from typing import Any, List import requests from models.member import ChatMember from settings import API_BASE headers = {"Content-Type": "application/json"} authors_by_user = {} authors_by_id = {} def _request_endpoint(query_name, body) -> Any: response = requests.post(API_BASE, headers=headers, json=body) print(f"[services.core] {query_name} response: <{response.status_code}> {response.text[:65]}..") if response.status_code == 200: r = response.json() if r: return r.get("data", {}).get(query_name, {}) return [] def get_all_authors() -> List[ChatMember]: if len(authors_by_user.keys()) == 0: print("[services.core] precaching authors...") query_name = "get_authors_all" # Check if authors are already cached if "authors" in authors_by_user: return authors_by_user["authors"] gql = { "query": "query { " + query_name + "{ id slug pic name user } }", "variables": None, } # Make a request to load authors authors = _request_endpoint(query_name, gql) for a in list(authors): authors_by_user.__setitem__(a["user"], a) authors_by_id.__setitem__(a["id"], a) print(f"[services.core] {len(authors)} authors precached") return list(authors_by_id.values()) def get_my_followed() -> List[ChatMember]: query_name = "get_my_followed" gql = { "query": "query { " + query_name + " { authors { id slug pic name } } }", "variables": None, } return _request_endpoint(query_name, gql) or []