66 lines
1.8 KiB
Python
66 lines
1.8 KiB
Python
from typing import List
|
|
|
|
import requests
|
|
from models.member import ChatMember
|
|
from settings import API_BASE
|
|
from datetime import datetime, timedelta
|
|
|
|
# Global variables to store cached authors data and last update time
|
|
cached_authors = None
|
|
last_update_time = None
|
|
update_interval = timedelta(minutes=2)
|
|
|
|
|
|
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[:65]}..")
|
|
|
|
if response.status_code == 200:
|
|
try:
|
|
r = response.json()
|
|
result = r.get("data", {}).get(query_name, {})
|
|
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},
|
|
}
|
|
|
|
author = _request_endpoint(query_name, gql).popitem()
|
|
|
|
return author
|
|
|
|
|
|
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", [])
|