From 49ac3e97e5b4a10235f2a4679eb67dce820f15f9 Mon Sep 17 00:00:00 2001 From: Igor Lobanov Date: Mon, 28 Nov 2022 21:29:02 +0100 Subject: [PATCH] topic stat via queries --- migration/tables/topics.py | 5 +- migration/tables/users.py | 3 +- resolvers/zine/load.py | 88 +++++++++-------------------- resolvers/zine/profile.py | 67 ++++++++++------------ resolvers/zine/reactions.py | 10 +--- resolvers/zine/topics.py | 109 +++++++++++++++++++++++++----------- services/zine/topics.py | 97 -------------------------------- 7 files changed, 139 insertions(+), 240 deletions(-) delete mode 100644 services/zine/topics.py diff --git a/migration/tables/topics.py b/migration/tables/topics.py index 15fcf245..3287adb7 100644 --- a/migration/tables/topics.py +++ b/migration/tables/topics.py @@ -9,9 +9,10 @@ def migrate(entry): topic_dict = { "slug": entry["slug"], "oid": entry["_id"], - "title": entry["title"].replace(" ", " ") + "title": entry["title"].replace(" ", " "), + "body": extract_md(html2text(body_orig), entry["_id"]) } - topic_dict["body"] = extract_md(html2text(body_orig), entry["_id"]) + with local_session() as session: slug = topic_dict["slug"] topic = session.query(Topic).filter(Topic.slug == slug).first() or Topic.create( diff --git a/migration/tables/users.py b/migration/tables/users.py index 6db7a243..bca86a76 100644 --- a/migration/tables/users.py +++ b/migration/tables/users.py @@ -23,8 +23,9 @@ def migrate(entry): "notifications": [], "links": [], "name": "anonymous", + "password": entry["services"]["password"].get("bcrypt") } - user_dict["password"] = entry["services"]["password"].get("bcrypt") + if "updatedAt" in entry: user_dict["updatedAt"] = parse(entry["updatedAt"]) if "wasOnineAt" in entry: diff --git a/resolvers/zine/load.py b/resolvers/zine/load.py index 79dcffb9..1408d135 100644 --- a/resolvers/zine/load.py +++ b/resolvers/zine/load.py @@ -9,72 +9,47 @@ from orm.shout import Shout, ShoutAuthor from orm.reaction import Reaction, ReactionKind -def add_viewed_stat_column(q): - return q.outerjoin(ViewedEntry).add_columns(sa.func.sum(ViewedEntry.amount).label('viewed_stat')) +def add_stat_columns(q): + q = q.outerjoin(ViewedEntry).add_columns(sa.func.sum(ViewedEntry.amount).label('viewed_stat')) - -def add_reacted_stat_column(q): aliased_reaction = aliased(Reaction) - return q.outerjoin(aliased_reaction).add_columns(sa.func.count(aliased_reaction.id).label('reacted_stat')) + q = q.outerjoin(aliased_reaction).add_columns( + sa.func.sum( + aliased_reaction.id + ).label('reacted_stat'), + sa.func.sum( + case( + (aliased_reaction.body.is_not(None), 1), + else_=0 + ) + ).label('commented_stat'), + sa.func.sum(case( + (aliased_reaction.kind == ReactionKind.AGREE, 1), + (aliased_reaction.kind == ReactionKind.DISAGREE, -1), + (aliased_reaction.kind == ReactionKind.PROOF, 1), + (aliased_reaction.kind == ReactionKind.DISPROOF, -1), + (aliased_reaction.kind == ReactionKind.ACCEPT, 1), + (aliased_reaction.kind == ReactionKind.REJECT, -1), + (aliased_reaction.kind == ReactionKind.LIKE, 1), + (aliased_reaction.kind == ReactionKind.DISLIKE, -1), + else_=0) + ).label('rating_stat')) -def add_commented_stat_column(q): - aliased_reaction = aliased(Reaction) - return q.outerjoin( - aliased_reaction, - aliased_reaction.shout == Shout.slug and aliased_reaction.body.is_not(None) - ).add_columns(sa.func.count(aliased_reaction.id).label('commented_stat')) - - -def add_rating_stat_column(q): - return q.outerjoin(Reaction).add_columns(sa.func.sum(case( - (Reaction.kind == ReactionKind.AGREE, 1), - (Reaction.kind == ReactionKind.DISAGREE, -1), - (Reaction.kind == ReactionKind.PROOF, 1), - (Reaction.kind == ReactionKind.DISPROOF, -1), - (Reaction.kind == ReactionKind.ACCEPT, 1), - (Reaction.kind == ReactionKind.REJECT, -1), - (Reaction.kind == ReactionKind.LIKE, 1), - (Reaction.kind == ReactionKind.DISLIKE, -1), - else_=0 - )).label('rating_stat')) - - -# def calc_reactions(q): -# aliased_reaction = aliased(Reaction) -# return q.join(aliased_reaction).add_columns( -# sa.func.sum(case( -# (aliased_reaction.kind == ReactionKind.AGREE, 1), -# (aliased_reaction.kind == ReactionKind.DISAGREE, -1), -# (aliased_reaction.kind == ReactionKind.PROOF, 1), -# (aliased_reaction.kind == ReactionKind.DISPROOF, -1), -# (aliased_reaction.kind == ReactionKind.ACCEPT, 1), -# (aliased_reaction.kind == ReactionKind.REJECT, -1), -# (aliased_reaction.kind == ReactionKind.LIKE, 1), -# (aliased_reaction.kind == ReactionKind.DISLIKE, -1), -# else_=0) -# ).label('rating'), -# sa.func.sum( -# case( -# (aliased_reaction.body.is_not(None), 1), -# else_=0 -# ) -# ).label('commented'), -# sa.func.sum( -# aliased_reaction.id -# ).label('reacted') -# ) + return q def apply_filters(q, filters, user=None): - filters = {} if filters is None else filters + if filters.get("reacted") and user: q.join(Reaction, Reaction.createdBy == user.slug) + v = filters.get("visibility") if v == "public": q = q.filter(Shout.visibility == filters.get("visibility")) if v == "community": q = q.filter(Shout.visibility.in_(["public", "community"])) + if filters.get("layout"): q = q.filter(Shout.layout == filters.get("layout")) if filters.get("author"): @@ -88,14 +63,7 @@ def apply_filters(q, filters, user=None): if filters.get("days"): before = datetime.now(tz=timezone.utc) - timedelta(days=int(filters.get("days")) or 30) q = q.filter(Shout.createdAt > before) - return q - -def add_stat_columns(q): - q = add_viewed_stat_column(q) - q = add_reacted_stat_column(q) - q = add_commented_stat_column(q) - q = add_rating_stat_column(q) return q @@ -162,7 +130,7 @@ async def load_shouts_by(_, info, options): q = add_stat_columns(q) user = info.context["request"].user - q = apply_filters(q, options.get("filters"), user) + q = apply_filters(q, options.get("filters", {}), user) order_by = options.get("order_by", Shout.createdAt) if order_by == 'reacted': diff --git a/resolvers/zine/profile.py b/resolvers/zine/profile.py index 115f1af1..b28e7c24 100644 --- a/resolvers/zine/profile.py +++ b/resolvers/zine/profile.py @@ -1,22 +1,24 @@ from typing import List from datetime import datetime, timedelta, timezone -from sqlalchemy import and_, func +from sqlalchemy import and_, func, select from sqlalchemy.orm import selectinload from auth.authenticate import login_required from base.orm import local_session from base.resolvers import mutation, query from orm.reaction import Reaction -from orm.shout import ShoutAuthor +from orm.shout import ShoutAuthor, ShoutTopic from orm.topic import Topic, TopicFollower from orm.user import AuthorFollower, Role, User, UserRating, UserRole # from .community import followed_communities from resolvers.inbox.unread import get_total_unread_counter +from resolvers.zine.topics import followed_by_user + async def user_subscriptions(slug: str): return { - "unread": await get_total_unread_counter(slug), # unread inbox messages counter + "unread": await get_total_unread_counter(slug), # unread inbox messages counter "topics": [t.slug for t in await followed_topics(slug)], # followed topics slugs "authors": [a.slug for a in await followed_authors(slug)], # followed authors slugs "reactions": await followed_reactions(slug) @@ -66,17 +68,7 @@ async def get_followed_topics(_, info, slug) -> List[Topic]: async def followed_topics(slug): - topics = [] - with local_session() as session: - topics = ( - session.query(Topic) - .join(TopicFollower) - .where(TopicFollower.follower == slug) - .all() - ) - for topic in topics: - topic.stat = await get_topic_stat(topic.slug) - return topics + return followed_by_user(slug) @query.field("userFollowedAuthors") @@ -89,9 +81,9 @@ async def followed_authors(slug) -> List[User]: with local_session() as session: authors = ( session.query(User) - .join(AuthorFollower, User.slug == AuthorFollower.author) - .where(AuthorFollower.follower == slug) - .all() + .join(AuthorFollower, User.slug == AuthorFollower.author) + .where(AuthorFollower.follower == slug) + .all() ) for author in authors: author.stat = await get_author_stat(author.slug) @@ -103,9 +95,9 @@ async def user_followers(_, _info, slug) -> List[User]: with local_session() as session: users = ( session.query(User) - .join(AuthorFollower, User.slug == AuthorFollower.follower) - .where(AuthorFollower.author == slug) - .all() + .join(AuthorFollower, User.slug == AuthorFollower.follower) + .where(AuthorFollower.author == slug) + .all() ) return users @@ -115,10 +107,10 @@ async def get_user_roles(slug): user = session.query(User).where(User.slug == slug).first() roles = ( session.query(Role) - .options(selectinload(Role.permissions)) - .join(UserRole) - .where(UserRole.user_id == user.id) - .all() + .options(selectinload(Role.permissions)) + .join(UserRole) + .where(UserRole.user_id == user.id) + .all() ) return roles @@ -144,8 +136,8 @@ async def rate_user(_, info, rated_userslug, value): with local_session() as session: rating = ( session.query(UserRating) - .filter(and_(UserRating.rater == user.slug, UserRating.user == rated_userslug)) - .first() + .filter(and_(UserRating.rater == user.slug, UserRating.user == rated_userslug)) + .first() ) if rating: rating.value = value @@ -171,12 +163,12 @@ def author_unfollow(user, slug): with local_session() as session: flw = ( session.query(AuthorFollower) - .filter( + .filter( and_( AuthorFollower.follower == user.slug, AuthorFollower.author == slug ) ) - .first() + .first() ) if not flw: raise Exception("[resolvers.profile] follower not exist, cant unfollow") @@ -204,7 +196,6 @@ async def get_author(_, _info, slug): @query.field("loadAuthorsBy") async def load_authors_by(_, info, by, limit, offset): - authors = [] with local_session() as session: aq = session.query(User) if by.get("slug"): @@ -212,24 +203,26 @@ async def load_authors_by(_, info, by, limit, offset): elif by.get("name"): aq = aq.filter(User.name.ilike(f"%{by['name']}%")) elif by.get("topic"): - aaa = list(map(lambda a: a.slug, TopicStat.authors_by_topic.get(by["topic"]))) - aq = aq.filter(User.name._in(aaa)) + aq = aq.join(ShoutAuthor).join(ShoutTopic).where(ShoutTopic.topic == by["topic"]) if by.get("lastSeen"): # in days days_before = datetime.now(tz=timezone.utc) - timedelta(days=by["lastSeen"]) aq = aq.filter(User.lastSeen > days_before) elif by.get("createdAt"): # in days days_before = datetime.now(tz=timezone.utc) - timedelta(days=by["createdAt"]) aq = aq.filter(User.createdAt > days_before) + aq = aq.group_by( User.id ).order_by( by.get("order") or "createdAt" ).limit(limit).offset(offset) + print(aq) - authors = list(map(lambda r: r.User, session.execute(aq))) - if by.get("stat"): - for a in authors: - a.stat = await get_author_stat(a.slug) - authors = list(set(authors)) - # authors = sorted(authors, key=lambda a: a["stat"].get(by.get("stat"))) + + authors = [] + for [author] in session.execute(aq): + if by.get("stat"): + author.stat = await get_author_stat(author.slug) + authors.append(author) + return authors diff --git a/resolvers/zine/reactions.py b/resolvers/zine/reactions.py index 5647c253..dafb1db4 100644 --- a/resolvers/zine/reactions.py +++ b/resolvers/zine/reactions.py @@ -10,15 +10,6 @@ from orm.shout import Shout, ShoutReactionsFollower from orm.user import User -async def get_reaction_stat(reaction_id): - return { - # "viewed": await ViewedStorage.get_reaction(reaction_id), - "reacted": len(await ReactedStorage.get_reaction(reaction_id)), - "rating": await ReactedStorage.get_reaction_rating(reaction_id), - "commented": len(await ReactedStorage.get_reaction_comments(reaction_id)), - } - - def reactions_follow(user: User, slug: str, auto=False): with local_session() as session: following = ( @@ -205,6 +196,7 @@ async def delete_reaction(_, info, rid): session.commit() return {} + @query.field("loadReactionsBy") async def load_reactions_by(_, _info, by, limit=50, offset=0): """ diff --git a/resolvers/zine/topics.py b/resolvers/zine/topics.py index 881299b1..f448a3b2 100644 --- a/resolvers/zine/topics.py +++ b/resolvers/zine/topics.py @@ -1,52 +1,94 @@ import sqlalchemy as sa -from sqlalchemy import and_, select +from sqlalchemy import and_, select, distinct + from auth.authenticate import login_required from base.orm import local_session from base.resolvers import mutation, query +from orm.shout import ShoutTopic, ShoutAuthor from orm.topic import Topic, TopicFollower from orm import Shout -async def get_topic_stat(slug): - return { - "shouts": len(TopicStat.shouts_by_topic.get(slug, {}).keys()), - "authors": len(TopicStat.authors_by_topic.get(slug, {}).keys()), - "followers": len(TopicStat.followers_by_topic.get(slug, {}).keys()) + +def add_topic_stat_columns(q): + q = q.outerjoin(ShoutTopic, Topic.slug == ShoutTopic.topic).add_columns( + sa.func.count(distinct(ShoutTopic.shout)).label('shouts_stat') + ).outerjoin(ShoutAuthor, ShoutTopic.shout == ShoutAuthor.shout).add_columns( + sa.func.count(distinct(ShoutAuthor.user)).label('authors_stat') + ).outerjoin(TopicFollower, + and_( + TopicFollower.topic == Topic.slug, + TopicFollower.follower == ShoutAuthor.user + )).add_columns( + sa.func.count(distinct(TopicFollower.follower)).label('followers_stat') + ) + + q = q.group_by(Topic.id) + + return q + + +def add_stat(topic, stat_columns): + [shouts_stat, authors_stat, followers_stat] = stat_columns + topic.stat = { + "shouts": shouts_stat, + "authors": authors_stat, + "followers": followers_stat } + return topic + + +def get_topics_from_query(q): + topics = [] + with local_session() as session: + for [topic, *stat_columns] in session.execute(q): + topic = add_stat(topic, stat_columns) + topics.append(topic) + + return topics + + +def followed_by_user(user_slug): + q = select(Topic) + q = add_topic_stat_columns(q) + q = q.where(TopicFollower.follower == user_slug) + + return get_topics_from_query(q) + + @query.field("topicsAll") async def topics_all(_, _info): - topics = await TopicStorage.get_topics_all() - for topic in topics: - topic.stat = await get_topic_stat(topic.slug) - return topics + q = select(Topic) + q = add_topic_stat_columns(q) + + return get_topics_from_query(q) @query.field("topicsByCommunity") async def topics_by_community(_, info, community): - topics = await TopicStorage.get_topics_by_community(community) - for topic in topics: - topic.stat = await get_topic_stat(topic.slug) - return topics + q = select(Topic).where(Topic.community == community) + q = add_topic_stat_columns(q) + + return get_topics_from_query(q) @query.field("topicsByAuthor") async def topics_by_author(_, _info, author): - shouts = TopicStorage.get_topics_by_author(author) - author_topics = set() - for s in shouts: - for tpc in s.topics: - tpc = await TopicStorage.topics[tpc.slug] - tpc.stat = await get_topic_stat(tpc.slug) - author_topics.add(tpc) - return list(author_topics) + q = select(Topic) + q = add_topic_stat_columns(q) + q = q.where(ShoutAuthor.user == author) + + return get_topics_from_query(q) @query.field("getTopic") async def get_topic(_, _info, slug): - t = TopicStorage.topics[slug] - t.stat = await get_topic_stat(slug) - return t + q = select(Topic).where(Topic.slug == slug) + q = add_topic_stat_columns(q) + + topics = get_topics_from_query(q) + return topics[0] @mutation.field("createTopic") @@ -57,7 +99,7 @@ async def create_topic(_, _info, inp): new_topic = Topic.create(**inp) session.add(new_topic) session.commit() - await TopicStorage.update_topic(new_topic) + return {"topic": new_topic} @@ -72,7 +114,7 @@ async def update_topic(_, _info, inp): else: topic.update(**inp) session.commit() - await TopicStorage.update_topic(topic.slug) + return {"topic": topic} @@ -81,7 +123,6 @@ async def topic_follow(user, slug): following = TopicFollower.create(topic=slug, follower=user.slug) session.add(following) session.commit() - await TopicStorage.update_topic(slug) async def topic_unfollow(user, slug): @@ -99,13 +140,13 @@ async def topic_unfollow(user, slug): else: session.delete(sub) session.commit() - await TopicStorage.update_topic(slug) @query.field("topicsRandom") async def topics_random(_, info, amount=12): - with local_session() as session: - q = select(Topic).join(Shout).group_by(Topic.id).having(sa.func.count(Shout.id) > 2).order_by( - sa.func.random()).limit(amount) - random_topics = list(map(lambda result_item: result_item.Topic, session.execute(q))) - return random_topics + q = select(Topic) + q = add_topic_stat_columns(q) + q = q.join(Shout, ShoutTopic.shout == Shout.slug).group_by(Topic.id).having(sa.func.count(Shout.id) > 2) + q = q.order_by(sa.func.random()).limit(amount) + + return get_topics_from_query(q) diff --git a/services/zine/topics.py b/services/zine/topics.py deleted file mode 100644 index 0aa6af7a..00000000 --- a/services/zine/topics.py +++ /dev/null @@ -1,97 +0,0 @@ -import asyncio -from base.orm import local_session -from orm.topic import Topic -from orm.shout import Shout -import sqlalchemy as sa -from sqlalchemy import select - - -class TopicStorage: - topics = {} - lock = asyncio.Lock() - random_topics = [] - - @staticmethod - def init(session): - self = TopicStorage - topics = session.query(Topic) - self.topics = dict([(topic.slug, topic) for topic in topics]) - for tpc in self.topics.values(): - # self.load_parents(tpc) - pass - - print("[zine.topics] %d precached" % len(self.topics.keys())) - - # @staticmethod - # def load_parents(topic): - # self = TopicStorage - # parents = [] - # for parent in self.topics.values(): - # if topic.slug in parent.children: - # parents.append(parent.slug) - # topic.parents = parents - # return topic - - @staticmethod - def get_random_topics(amount): - return TopicStorage.random_topics[0:amount] - - @staticmethod - def renew_topics_random(): - with local_session() as session: - q = select(Topic).join(Shout).group_by(Topic.id).having(sa.func.count(Shout.id) > 2).order_by( - sa.func.random()).limit(50) - TopicStorage.random_topics = list(map( - lambda result_item: result_item.Topic, session.execute(q) - )) - - @staticmethod - async def worker(): - self = TopicStorage - async with self.lock: - while True: - try: - self.renew_topics_random() - except Exception as err: - print("[zine.topics] error %s" % (err)) - await asyncio.sleep(300) # 5 mins - - @staticmethod - async def get_topics_all(): - self = TopicStorage - async with self.lock: - return list(self.topics.values()) - - @staticmethod - async def get_topics_by_slugs(slugs): - self = TopicStorage - async with self.lock: - if not slugs: - return self.topics.values() - topics = filter(lambda topic: topic.slug in slugs, self.topics.values()) - return list(topics) - - @staticmethod - async def get_topics_by_community(community): - self = TopicStorage - async with self.lock: - topics = filter( - lambda topic: topic.community == community, self.topics.values() - ) - return list(topics) - - @staticmethod - async def get_topics_by_author(author): - self = TopicStorage - async with self.lock: - topics = filter( - lambda topic: topic.community == author, self.topics.values() - ) - return list(topics) - - @staticmethod - async def update_topic(topic): - self = TopicStorage - async with self.lock: - self.topics[topic.slug] = topic - # self.load_parents(topic)