topic stat via queries

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(

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:

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):
+def add_stat_columns(q):
-    return q.outerjoin(ViewedEntry).add_columns(sa.func.sum(ViewedEntry.amount).label('viewed_stat'))
+    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):
+    return 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')
-#     )
 
 
 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':

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 = []
+    return followed_by_user(slug)
-    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
 
 
 @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)
+                .join(AuthorFollower, User.slug == AuthorFollower.author)
-            .where(AuthorFollower.follower == slug)
+                .where(AuthorFollower.follower == slug)
-            .all()
+                .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)
+                .join(AuthorFollower, User.slug == AuthorFollower.follower)
-            .where(AuthorFollower.author == slug)
+                .where(AuthorFollower.author == slug)
-            .all()
+                .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))
+                .options(selectinload(Role.permissions))
-            .join(UserRole)
+                .join(UserRole)
-            .where(UserRole.user_id == user.id)
+                .where(UserRole.user_id == user.id)
-            .all()
+                .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))
+                .filter(and_(UserRating.rater == user.slug, UserRating.user == rated_userslug))
-            .first()
+                .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.join(ShoutAuthor).join(ShoutTopic).where(ShoutTopic.topic == by["topic"])
-            aq = aq.filter(User.name._in(aaa))
         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"):
+        authors = []
-            for a in authors:
+        for [author] in session.execute(aq):
-                a.stat = await get_author_stat(a.slug)
+            if by.get("stat"):
-    authors = list(set(authors))
+                author.stat = await get_author_stat(author.slug)
-    # authors = sorted(authors, key=lambda a: a["stat"].get(by.get("stat")))
+            authors.append(author)
+
     return authors

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):
     """

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 {
+def add_topic_stat_columns(q):
-        "shouts": len(TopicStat.shouts_by_topic.get(slug, {}).keys()),
+    q = q.outerjoin(ShoutTopic, Topic.slug == ShoutTopic.topic).add_columns(
-        "authors": len(TopicStat.authors_by_topic.get(slug, {}).keys()),
+        sa.func.count(distinct(ShoutTopic.shout)).label('shouts_stat')
-        "followers": len(TopicStat.followers_by_topic.get(slug, {}).keys())
+    ).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()
+    q = select(Topic)
-    for topic in topics:
+    q = add_topic_stat_columns(q)
-        topic.stat = await get_topic_stat(topic.slug)
+
-    return topics
+    return get_topics_from_query(q)
 
 
 @query.field("topicsByCommunity")
 async def topics_by_community(_, info, community):
-    topics = await TopicStorage.get_topics_by_community(community)
+    q = select(Topic).where(Topic.community == community)
-    for topic in topics:
+    q = add_topic_stat_columns(q)
-        topic.stat = await get_topic_stat(topic.slug)
+
-    return topics
+    return get_topics_from_query(q)
 
 
 @query.field("topicsByAuthor")
 async def topics_by_author(_, _info, author):
-    shouts = TopicStorage.get_topics_by_author(author)
+    q = select(Topic)
-    author_topics = set()
+    q = add_topic_stat_columns(q)
-    for s in shouts:
+    q = q.where(ShoutAuthor.user == author)
-        for tpc in s.topics:
+
-            tpc = await TopicStorage.topics[tpc.slug]
+    return get_topics_from_query(q)
-            tpc.stat = await get_topic_stat(tpc.slug)
-            author_topics.add(tpc)
-    return list(author_topics)
 
 
 @query.field("getTopic")
 async def get_topic(_, _info, slug):
-    t = TopicStorage.topics[slug]
+    q = select(Topic).where(Topic.slug == slug)
-    t.stat = await get_topic_stat(slug)
+    q = add_topic_stat_columns(q)
-    return t
+
+    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)
-        q = select(Topic).join(Shout).group_by(Topic.id).having(sa.func.count(Shout.id) > 2).order_by(
+    q = add_topic_stat_columns(q)
-            sa.func.random()).limit(amount)
+    q = q.join(Shout, ShoutTopic.shout == Shout.slug).group_by(Topic.id).having(sa.func.count(Shout.id) > 2)
-        random_topics = list(map(lambda result_item: result_item.Topic, session.execute(q)))
+    q = q.order_by(sa.func.random()).limit(amount)
-        return random_topics
+
+    return get_topics_from_query(q)

services/zine/topics.py

@@ -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)