inbox/services/core.py

from typing import Any, List

from aiohttp import ClientSession
from models.member import ChatMember
from settings import API_BASE

headers = {"Content-Type": "application/json"}
authors_by_user = {}
authors_by_id = {}


async def _request_endpoint(query_name, body) -> Any:
    async with ClientSession() as session:
        async with session.post(API_BASE, headers=headers, json=body) as response:
            print(f"[services.core] {query_name} response: <{response.status}> {(await response.text())[:15]}..")
            if response.status == 200:
                r = await response.json()
                if r:
                    return r.get("data", {}).get(query_name, {})
            return []


async def get_all_authors() -> List[ChatMember]:
    if len(authors_by_user.keys()) == 0:
        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 } }",
            "variables": None,
        }

        # Make a request to load authors
        authors = await _request_endpoint(query_name, gql)

        for a in authors:
            authors_by_user[a["user"]] = a
            authors_by_id[a["id"]] = a

    return list(authors_by_id.values())


async def get_my_followed() -> List[ChatMember]:
    query_name = "get_my_followed"

    gql = {
        "query": "query { " + query_name + " { authors { id slug pic name } } }",
        "variables": None,
    }

    return await _request_endpoint(query_name, gql) or []


async def get_author(user: str = ""):
    return authors_by_user.get(user)