notifier/services/core.py

from typing import List, Any
import aiohttp
from settings import API_BASE

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


async def _request_endpoint(query_name, body):
    async with aiohttp.ClientSession() as session:
        try:
            async with session.post(API_BASE, headers=headers, json=body) as response:
                print(f"[services.core] {query_name}: [{response.status}] {len(await response.text())} bytes")
                if response.status != 200:
                    return []
                r = await response.json()
                if r:
                    return r.get("data", {}).get(query_name, {})
                else:
                    raise Exception("json response error")
        except Exception:
            import traceback

            traceback.print_exc()


async def get_followed_shouts(author_id: int) -> List[Any]:
    query_name = "load_shouts_followed"
    query_type = "query"
    operation = "GetFollowedShouts"
    query_fields = "id slug title"

    query = f"""{query_type} {operation}($author_id: Int!, limit: Int, offset: Int) {{
        {query_name}(author_id: $author_id, limit: $limit, offset: $offset) {{ {query_fields} }}
    }}"""

    body = {
        "query": query,
        "operationName": operation,
        "variables": {"author_id": author_id, "limit": 1000, "offset": 0},  # FIXME: too big limit
    }

    return await _request_endpoint(query_name, body)