notifier/services/core.py
2023-11-24 01:58:55 +03:00

63 lines
1.9 KiB
Python

from httpx import AsyncClient
from settings import API_BASE
from typing import List
from models.member import ChatMember
headers = {"Content-Type": "application/json"}
async def get_all_authors() -> List[ChatMember]:
query_name = "authorsAll"
query_type = "query"
operation = "AuthorsAll"
query_fields = "id slug userpic name"
gql = {
"query": query_type + " " + operation + " { " + query_name + " { " + query_fields + " } " + " }",
"operationName": operation,
"variables": None,
}
async with AsyncClient() as client:
try:
response = await client.post(API_BASE, headers=headers, json=gql)
print(f"[services.core] {query_name}: [{response.status_code}] {len(response.text)} bytes")
if response.status_code != 200:
return []
r = response.json()
if r:
return r.get("data", {}).get(query_name, [])
except Exception:
import traceback
traceback.print_exc()
return []
async def get_my_followings() -> List[ChatMember]:
query_name = "loadMySubscriptions"
query_type = "query"
operation = "LoadMySubscriptions"
query_fields = "id slug userpic name"
gql = {
"query": query_type + " " + operation + " { " + query_name + " { authors {" + query_fields + "} } " + " }",
"operationName": operation,
"variables": None,
}
async with AsyncClient() as client:
try:
response = await client.post(API_BASE, headers=headers, json=gql)
print(f"[services.core] {query_name}: [{response.status_code}] {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 []