inbox/services/core.py

from typing import Any, List

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

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


async def _request_endpoint(query_name, body) -> Any:
    async with aiohttp.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()}")
            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]:
    query_name = "authorsAll"
    query_type = "query"
    operation = "AuthorsAll"
    query_fields = "id slug pic name"

    gql = {
        "query": query_type + " " + operation + " { " + query_name + " { " + query_fields + " } " + " }",
        "operationName": operation,
        "variables": None,
    }

    return await _request_endpoint(query_name, gql)


async def get_my_followed() -> List[ChatMember]:
    query_name = "get_my_followed"
    query_type = "query"
    operation = "GetMyFollowed"
    query_fields = "id slug pic name"

    gql = {
        "query": query_type + " " + operation + " { " + query_name + " { authors {" + query_fields + "} } " + " }",
        "operationName": operation,
        "variables": None,
    }

    async with ClientSession() as client:
        try:
            response: ClientResponse = await client.post(API_BASE, headers=headers, json=gql)
            print(f"[services.core] {query_name}: [{response.status}] {len(response.text)} bytes")
            if response.status_code != 200:
                return []
            r = response.json()
            if r:
                return r.get("data", {}).get(query_name, {}).get("authors", [])
        except Exception:
            import traceback

            traceback.print_exc()
    return []


async def get_author(author_id: int = None, slug: str = "", user: str = ""):
    query_name = "get_author(author_id: $author_id, slug: $slug, user: $user)"
    query_type = "query"
    operation = "GetAuthor($author_id: Int, $slug: String, $user: String)"
    query_fields = "id slug pic name"
    vars = {}
    if author_id:
        vars["author_id"] = author_id
    elif slug:
        vars["slug"] = slug
    elif user:
        vars["user"] = user

    gql = {
        "query": query_type + " " + operation + " { " + query_name + " { " + query_fields + "} " + " }",
        "operationName": operation,
        "variables": None if vars == {} else vars,
    }

    r = await _request_endpoint(query_name, gql)