106 lines
2.9 KiB
Python
106 lines
2.9 KiB
Python
import asyncio
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import json
|
|
|
|
from services.logger import root_logger as logger
|
|
from services.rediscache import redis
|
|
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
|
|
async def get_all_authors():
|
|
authors = []
|
|
redis_key = "user:*"
|
|
|
|
result = await redis.execute("GET", redis_key)
|
|
if isinstance(result, str):
|
|
authors = json.loads(result)
|
|
|
|
return authors
|
|
|
|
|
|
async def get_author_by_user(user: str):
|
|
author = None
|
|
redis_key = f"user:{user}"
|
|
|
|
result = await redis.execute("GET", redis_key)
|
|
if isinstance(result, str):
|
|
author = json.loads(result)
|
|
|
|
return author
|
|
|
|
|
|
async def get_author_followed(author_id: int):
|
|
authors = []
|
|
redis_key = f"author:{author_id}:follows-authors"
|
|
|
|
result = await redis.execute("GET", redis_key)
|
|
if isinstance(result, str):
|
|
authors = json.loads(result)
|
|
|
|
return authors
|
|
|
|
|
|
class CacheStorage:
|
|
lock = asyncio.Lock()
|
|
period = 5 * 60 # every 5 mins
|
|
client = None
|
|
authors = []
|
|
authors_by_user = {}
|
|
authors_by_id = {}
|
|
|
|
@staticmethod
|
|
async def init():
|
|
"""graphql client connection using permanent token"""
|
|
self = CacheStorage
|
|
async with self.lock:
|
|
task = asyncio.create_task(self.worker())
|
|
logger.info(task)
|
|
|
|
@staticmethod
|
|
async def update_authors():
|
|
self = CacheStorage
|
|
async with self.lock:
|
|
result = get_all_authors()
|
|
logger.info(f"cache loaded {len(result)}")
|
|
if result:
|
|
CacheStorage.authors = result
|
|
for a in result:
|
|
user_id = a.get("user")
|
|
author_id = str(a.get("id"))
|
|
self.authors_by_user[user_id] = a
|
|
self.authors_by_id[author_id] = a
|
|
|
|
@staticmethod
|
|
async def worker():
|
|
"""async task worker"""
|
|
failed = 0
|
|
self = CacheStorage
|
|
while True:
|
|
try:
|
|
logger.info(" - updating profiles data...")
|
|
await self.update_authors()
|
|
failed = 0
|
|
except Exception as er:
|
|
failed += 1
|
|
logger.error(f"{er} - update failed #{failed}, wait 10 seconds")
|
|
if failed > 3:
|
|
logger.error(" - not trying to update anymore")
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
break
|
|
if failed == 0:
|
|
when = datetime.now(timezone.utc) + timedelta(seconds=self.period)
|
|
t = format(when.astimezone().isoformat())
|
|
logger.info(
|
|
" ⎩ next update: %s"
|
|
% (t.split("T")[0] + " " + t.split("T")[1].split(".")[0])
|
|
)
|
|
await asyncio.sleep(self.period)
|
|
else:
|
|
await asyncio.sleep(10)
|
|
logger.info(" - trying to update data again")
|