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

full-preload
Some checks failed
Deploy to core / deploy (push) Failing after 1m32s
Details

Browse Source
This commit is contained in:
Untone 2024-02-03 16:17:00 +03:00
parent 066770febc
commit 53ceac108f
1 changed files with 118 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

185
services/following.py
View File

@ -2,126 +2,171 @@ import asyncio
import logging import logging
import time import time
from orm.author import AuthorFollower from sqlalchemy import and_, joinedload
from orm.shout import ShoutReactionsFollower
from orm.topic import TopicFollower from orm.author import Author, AuthorFollower
from orm.shout import Shout, ShoutReactionsFollower
from orm.topic import Topic, TopicFollower
from services.db import local_session from services.db import local_session
logger = logging.getLogger('[services.following] ') logger = logging.getLogger('[services.following] ')
logger.setLevel(logging.DEBUG) logger.setLevel(logging.DEBUG)
MODEL_CLASSES = {'author': AuthorFollower, 'topic': TopicFollower, 'shout': ShoutReactionsFollower} MODEL_CLASSES = {'author': AuthorFollower, 'topic': TopicFollower, 'shout': ShoutReactionsFollower}
class FollowingResult: class FollowingResult:
def __init__(self, event, kind, payload): def __init__(self, event, kind, payload):
self.event = event self.event = event
self.kind = kind self.kind = kind
self.payload = payload self.payload = payload
class Following: class Following:
def __init__(self, kind, uid): def __init__(self, kind, uid):
self.kind = kind # author, topic, shout self.kind = kind # author, topic, shout
self.uid = uid self.uid = uid
self.queue = asyncio.Queue() self.queue = asyncio.Queue()
class FollowingManager: class FollowingManager:
lock = asyncio.Lock() lock = asyncio.Lock()
followers_by_kind = {'author': [], 'topic': [], 'shout': []} followers_by_kind = None
authors_by_follower = {} authors_by_follower = None
topics_by_follower = {} topics_by_follower = None
shouts_by_follower = {} shouts_by_follower = None
authors_by_id = None
shouts_by_id = None
topics_by_id = None
@staticmethod @staticmethod
async def preload(): async def preload():
logger.info(' preloading started...') ts = time.time()
ts = int(time.time())
async with FollowingManager.lock: async with FollowingManager.lock:
followers_by_kind = {'author': {}, 'topic': {}, 'shout': {}}
authors_by_follower = {}
topics_by_follower = {}
shouts_by_follower = {}
authors_by_id = {}
topics_by_id = {}
shouts_by_id = {}
with local_session() as session: with local_session() as session:
# Load followers_by_kind all_authors = session.query(Author).all()
for kind in FollowingManager.followers_by_kind.keys(): for author in all_authors:
authors_by_id[author.id] = author
all_topics = session.query(Topic).all()
for topic in all_topics:
topics_by_id[topic.id] = topic
all_shouts = session.query(Shout).filter(and_(Shout.published_at.is_not(None), Shout.deleted_at.is_(None))).all()
for shout in all_shouts:
shouts_by_id[shout.id] = shout
for kind in followers_by_kind.keys():
model_class = MODEL_CLASSES[kind] model_class = MODEL_CLASSES[kind]
followers = session.query(model_class.follower).distinct().all() followings = (
FollowingManager.followers_by_kind[kind] = [follower[0] for follower in followers] session.query(model_class.follower)
.distinct()
# Load authors_by_follower .options(joinedload(model_class.follower))
c = 0 .all()
for following in session.query(AuthorFollower).all():
FollowingManager.authors_by_follower[following.follower] = FollowingManager.authors_by_follower.get(
following.follower, []
) )
FollowingManager.authors_by_follower[following.follower].append(following.author) for following in followings:
c += 1 if kind == 'topic':
logger.info(f' {c} authors followings') followers_by_kind[kind][following.topic] = followers_by_kind[kind].get(following.topic, set())
followers_by_kind[kind][following.topic].add(following.follower)
elif kind == 'author':
followers_by_kind[kind][following.author] = followers_by_kind[kind].get(following.author, set())
followers_by_kind[kind][following.author].add(following.follower)
elif kind == 'shout':
followers_by_kind[kind][following.shout] = followers_by_kind[kind].get(following.shout, set())
followers_by_kind[kind][following.shout].add(following.follower)
# Load topics_by_follower # Load authors_by_follower, topics_by_follower, and shouts_by_follower
c = 0 for entity_kind in followers_by_kind.keys():
for following in session.query(TopicFollower).all(): followers_dict = followers_by_kind[entity_kind]
FollowingManager.topics_by_follower[following.follower] = FollowingManager.topics_by_follower.get( if followers_dict:
following.follower, [] entity_class = MODEL_CLASSES[entity_kind]
followings = (
session.query(entity_class)
.options(joinedload(entity_class.follower), joinedload(entity_class.entity))
.all()
) )
FollowingManager.topics_by_follower[following.follower].append(following.topic) for following in followings:
c += 1 follower_id = following.follower.id
logger.info(f' {c} topics followings') entity_id = following.entity.id
followers_dict.setdefault(follower_id, set()).add(entity_id)
# Load shouts_by_follower # Assign the loaded dictionaries to the class attributes
c = 0 FollowingManager.authors_by_follower = authors_by_follower
for following in session.query(ShoutReactionsFollower).all(): FollowingManager.topics_by_follower = topics_by_follower
FollowingManager.shouts_by_follower[following.follower] = FollowingManager.shouts_by_follower.get( FollowingManager.shouts_by_follower = shouts_by_follower
following.follower, [] FollowingManager.authors_by_id = authors_by_id
) FollowingManager.topics_by_id = topics_by_id
FollowingManager.shouts_by_follower[following.follower].append(following.shout) FollowingManager.shouts_by_id = shouts_by_id
c += 1
logger.info(f' {c} shouts followings') logger.info(f' preloaded in {time.time() - ts} msec')
logger.info(f' preloading finished at {(int(time.time()) - ts)/1000} secs')
@staticmethod @staticmethod
async def register(kind, uid): async def register(entity_kind, entity_id, follower_id):
async with FollowingManager.lock: self = FollowingManager
if uid not in FollowingManager.followers_by_kind[kind]:
FollowingManager.followers_by_kind[kind].append(uid)
@staticmethod
async def remove(kind, uid):
async with FollowingManager.lock:
FollowingManager.followers_by_kind[kind] = [
follower for follower in FollowingManager.followers_by_kind[kind] if follower != uid
]
@staticmethod
async def push(kind, payload):
try: try:
async with FollowingManager.lock: async with self.lock:
for entity in FollowingManager.followers_by_kind[kind]: if isinstance(self.authors_by_id, dict):
if payload.shout['created_by'] == entity: follower = self.authors_by_id.get(follower_id)
await entity.queue.put(payload) if follower and self.followers_by_kind:
except Exception as e: self.followers_by_kind[entity_kind][entity_id] = self.followers_by_kind[entity_kind].get(entity_id, set())
print(f'Error in push method: {e}') self.followers_by_kind[entity_kind][entity_id].add(follower)
if entity_kind == 'author' and self.authors_by_follower and self.authors_by_id:
author = self.authors_by_id.get(entity_id)
self.authors_by_follower.setdefault(follower_id, set()).add(author)
if entity_kind == 'topic' and self.topics_by_follower and self.topics_by_id:
topic = self.topics_by_id.get(entity_id)
self.topics_by_follower.setdefault(follower_id, set()).add(topic)
if entity_kind == 'shout' and self.shouts_by_follower and self.shouts_by_id:
shout = self.shouts_by_id.get(entity_id)
self.shouts_by_follower.setdefault(follower_id, set()).add(shout)
except Exception as exc:
logger.warn(exc)
@staticmethod
async def remove(entity_kind, entity_id, follower_id):
self = FollowingManager
async with self.lock:
if self.followers_by_kind and entity_kind in self.followers_by_kind and entity_id in self.followers_by_kind[entity_kind]:
try:
del self.followers_by_kind[entity_kind][entity_id]
if entity_kind == 'author' and self.authors_by_follower:
del self.authors_by_follower[follower_id][entity_id]
elif entity_kind == 'topic' and self.topics_by_follower:
del self.topics_by_follower[follower_id][entity_id]
elif entity_kind == 'shout' and self.shouts_by_follower:
del self.shouts_by_follower[follower_id][entity_id]
except Exception as exc:
logger.warn(exc)
if isinstance(self.authors_by_id, dict):
follower = self.authors_by_id.get(follower_id)
if follower:
self.followers_by_kind[entity_kind][entity_id].remove(follower)
@staticmethod @staticmethod
async def get_followers_by_kind(kind, target_id=None): async def get_followers_by_kind(kind, target_id=None):
async with FollowingManager.lock: async with FollowingManager.lock:
if FollowingManager.followers_by_kind:
return ( return (
FollowingManager.followers_by_kind[kind][target_id] FollowingManager.followers_by_kind[kind] if target_id is None else {target_id}
if target_id
else FollowingManager.followers_by_kind[kind]
) )
@staticmethod @staticmethod
async def get_authors_for(follower_id): async def get_authors_for(follower_id):
async with FollowingManager.lock: async with FollowingManager.lock:
return FollowingManager.authors_by_follower.get(follower_id, []) if FollowingManager.authors_by_follower:
return FollowingManager.authors_by_follower.get(follower_id, set())
@staticmethod @staticmethod
async def get_topics_for(follower_id): async def get_topics_for(follower_id):
async with FollowingManager.lock: async with FollowingManager.lock:
return FollowingManager.topics_by_follower.get(follower_id, []) if FollowingManager.topics_by_follower:
return FollowingManager.topics_by_follower.get(follower_id, set())
@staticmethod @staticmethod
async def get_shouts_for(follower_id): async def get_shouts_for(follower_id):
async with FollowingManager.lock: async with FollowingManager.lock:
return FollowingManager.shouts_by_follower.get(follower_id, []) if FollowingManager.shouts_by_follower:
return FollowingManager.shouts_by_follower.get(follower_id, set())