discours.io/core
5
0
Fork 0
Code Pull Requests Actions Projects Activity

async-fix

Browse Source
This commit is contained in:
Tony Rewin 2023-10-06 01:45:32 +03:00
parent 12c43dbf32
commit 8de2eb385b
2 changed files with 18 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
requirements.txt
View File

@ -1,9 +1,9 @@
aredis>=1.1.8 aredis>=1.1.8
httpx>=0.23.0 httpx>=0.23.0
gql[requests] # TODO: replace with httpx when stable git+https://github.com/graphql-python/gql.git#master
git+https://github.com/encode/starlette.git#main git+https://github.com/encode/starlette.git#main
sqlalchemy>=1.4.41 sqlalchemy>=1.4.41
ariadne>=0.17.0 git+https://dev.discours.io/discours.io/ariadne.git#master
uvicorn>=0.18.3 uvicorn>=0.18.3
python-frontmatter~=1.0.0 python-frontmatter~=1.0.0
PyYAML>=5.4 PyYAML>=5.4

34
services/viewed.py
View File

@ -1,10 +1,10 @@
import asyncio import asyncio
import time import time
from datetime import timedelta, timezone, datetime from datetime import timedelta, timezone, datetime
from os import environ from os import environ, path
from gql import Client, gql from gql import Client, gql
from gql.transport.requests import RequestsHTTPTransport from gql.transport.httpx import HTTPXAsyncTransport
from services.db import local_session from services.db import local_session
from orm import Topic from orm import Topic
@ -39,14 +39,14 @@ load_pages = gql(
} } """ } } """
) )
schema_str = open("schemas/ackee.graphql").read() schema_str = open(path.dirname(__file__) + "/ackee.graphql").read()
token = environ.get("ACKEE_TOKEN", "") token = environ.get("ACKEE_TOKEN", "")
def create_client(headers=None, schema=None): def create_client(headers=None, schema=None):
return Client( return Client(
schema=schema, schema=schema,
transport=RequestsHTTPTransport( transport=HTTPXAsyncTransport(
url="https://ackee.discours.io/api", url="https://ackee.discours.io/api",
headers=headers, headers=headers,
), ),
@ -75,16 +75,17 @@ class ViewedStorage:
{"Authorization": "Bearer %s" % str(token)}, schema=schema_str {"Authorization": "Bearer %s" % str(token)}, schema=schema_str
) )
print( print(
"[stat] * authorized permanentely by ackee.discours.io: %s" % token "[stat.viewed] * authorized permanentely by ackee.discours.io: %s"
% token
) )
else: else:
print("[stat] * please set ACKEE_TOKEN") print("[stat.viewed] * please set ACKEE_TOKEN")
self.disabled = True self.disabled = True
@staticmethod @staticmethod
async def update_pages(): async def update_pages():
"""query all the pages from ackee sorted by views count""" """query all the pages from ackee sorted by views count"""
print("[stat] ⎧ updating ackee pages data ---") print("[stat.viewed] ⎧ updating ackee pages data ---")
start = time.time() start = time.time()
self = ViewedStorage self = ViewedStorage
try: try:
@ -100,12 +101,12 @@ class ViewedStorage:
await ViewedStorage.increment(slug, shouts[slug]) await ViewedStorage.increment(slug, shouts[slug])
except Exception: except Exception:
pass pass
print("[stat] ⎪ %d pages collected " % len(shouts.keys())) print("[stat.viewed] ⎪ %d pages collected " % len(shouts.keys()))
except Exception as e: except Exception as e:
raise e raise e
end = time.time() end = time.time()
print("[stat] ⎪ update_pages took %fs " % (end - start)) print("[stat.viewed] ⎪ update_pages took %fs " % (end - start))
@staticmethod @staticmethod
async def get_facts(): async def get_facts():
@ -196,29 +197,26 @@ class ViewedStorage:
self = ViewedStorage self = ViewedStorage
if self.disabled: if self.disabled:
return return
print("[stat] worker started")
while True: while True:
try: try:
print("[stat] - updating views...") print("[stat.viewed] - updating views...")
await self.update_pages() await self.update_pages()
failed = 0 failed = 0
except Exception: except Exception:
import traceback
traceback.print_exc()
failed += 1 failed += 1
print("[stat] - update failed #%d, wait 10 seconds" % failed) print("[stat.viewed] - update failed #%d, wait 10 seconds" % failed)
if failed > 3: if failed > 3:
print("[stat] - not trying to update anymore") print("[stat.viewed] - not trying to update anymore")
break break
if failed == 0: if failed == 0:
when = datetime.now(timezone.utc) + timedelta(seconds=self.period) when = datetime.now(timezone.utc) + timedelta(seconds=self.period)
t = format(when.astimezone().isoformat()) t = format(when.astimezone().isoformat())
print( print(
"[stat] ⎩ next update: %s" "[stat.viewed] ⎩ next update: %s"
% (t.split("T")[0] + " " + t.split("T")[1].split(".")[0]) % (t.split("T")[0] + " " + t.split("T")[1].split(".")[0])
) )
await asyncio.sleep(self.period) await asyncio.sleep(self.period)
else: else:
await asyncio.sleep(10) await asyncio.sleep(10)
print("[stat] - trying to update data again") print("[stat.viewed] - trying to update data again")