45 lines
1.4 KiB
Python
45 lines
1.4 KiB
Python
from typing import Any, List
|
|
|
|
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):
|
|
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)
|