64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
from httpx import AsyncClient
|
|
from settings import API_BASE
|
|
|
|
headers = {"Content-Type": "application/json"}
|
|
INTERNAL_AUTH_SERVER = "v2." in API_BASE
|
|
|
|
|
|
async def get_all_authors():
|
|
query_name = "authorsAll"
|
|
query_type = "query"
|
|
operation = "AuthorsAll"
|
|
query_fields = "id slug userpic name"
|
|
headers = {"Content-Type": "application/json"} # "Bearer " + removed
|
|
|
|
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)
|
|
except Exception:
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
print(f"[services.core] {query_name}: [{response.status_code}] {len(response.text)} bytes")
|
|
if response.status_code != 200:
|
|
return None
|
|
r = response.json()
|
|
return r.get("data", {}).get(query_name)
|
|
|
|
|
|
async def get_my_followings():
|
|
query_name = "loadMySubscriptions"
|
|
query_type = "query"
|
|
operation = "LoadMySubscriptions"
|
|
query_fields = "id slug userpic name"
|
|
headers = {"Content-Type": "application/json"} # "Bearer " + removed
|
|
|
|
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)
|
|
except Exception:
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
print(f"[services.core] {query_name}: [{response.status_code}] {len(response.text)} bytes")
|
|
if response.status_code != 200:
|
|
return None
|
|
r = response.json()
|
|
data = r.get("data")
|
|
if data:
|
|
d = data.get(query_name)
|
|
if d:
|
|
authors = d.get("authors", [])
|
|
return authors
|
|
return []
|