inbox/services/core.py

from typing import List
import asyncio
import requests
from datetime import datetime, timezone, timedelta

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}..")

    if response.status_code == 200:
        try:
            r = response.json()
            result = r.get("data", {}).get(query_name, {})
            if result:
                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},
    }

    return _request_endpoint(query_name, gql)


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", [])


class CacheStorage:
    lock = asyncio.Lock()
    period = 5 * 60  # every 5 mins
    client = None
    authors = []
    authors_by_user = {}
    authors_by_id = {}

    @staticmethod
    async def init():
        """graphql client connection using permanent token"""
        self = CacheStorage
        async with self.lock:
            task = asyncio.create_task(self.worker())
            print(task)

    @staticmethod
    async def update_authors():
        self = CacheStorage
        async with self.lock:
            result = get_all_authors()
            print(f"[services.core] loaded {len(result)}")
            if result:
                CacheStorage.authors = result
                for a in result:
                    self.authors_by_user[a.user] = a
                    self.authors_by_id[a.id] = a

    @staticmethod
    async def worker():
        """async task worker"""
        failed = 0
        self = CacheStorage
        while True:
            try:
                print("[services.core] - updating views...")
                await self.update_authors()
                failed = 0
            except Exception:
                failed += 1
                print("[services.core] - update failed #%d, wait 10 seconds" % failed)
                if failed > 3:
                    print("[services.core] - not trying to update anymore")
                    break
            if failed == 0:
                when = datetime.now(timezone.utc) + timedelta(seconds=self.period)
                t = format(when.astimezone().isoformat())
                print("[services.core] ⎩ next update: %s" % (t.split("T")[0] + " " + t.split("T")[1].split(".")[0]))
                await asyncio.sleep(self.period)
            else:
                await asyncio.sleep(10)
                print("[services.core] - trying to update data again")