from typing import List import requests from models.member import ChatMember from settings import API_BASE def _request_endpoint(query_name, body) -> dict: print(f"[services.core] requesting {query_name}...") response = requests.post(API_BASE, headers={"Content-Type": "application/json"}, json=body) print(f"[services.core] {query_name} response: <{response.status_code}> {response.text[:65]}..") if response.status_code == 200: try: r = response.json() result = r.get("data", {}).get(query_name, {}) print(f"[services.core] entries amount in result: {len(result)} ") return result except ValueError as e: print(f"[services.core] Error decoding JSON response: {e}") return {} def get_all_authors(): query_name = "get_authors_all" gql = { "query": "query { " + query_name + "{ id slug pic name user } }", "variables": None, } return _request_endpoint(query_name, gql) def get_author_by_user(user: str): operation = "GetAuthorId" query_name = "get_author_id" gql = { "query": f"query {operation}($user: String!) {{ {query_name}(user: $user){{ id }} }}", "operationName": operation, "variables": {"user": user}, } author = _request_endpoint(query_name, gql)[0] return author def get_my_followed() -> List[ChatMember]: query_name = "get_my_followed" gql = { "query": "query { " + query_name + " { authors { id slug pic name } } }", "variables": None, } result = _request_endpoint(query_name, gql) return result.get("authors", [])