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

schemaup

Browse Source
This commit is contained in:
tonyrewin
2022-11-30 23:20:08 +03:00
parent d0b875c157 3ed1857f11
commit c86549507e
35 changed files with 617 additions and 782 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
services/auth/roles.py
View File

@@ -1,35 +0,0 @@
import asyncio
from sqlalchemy.orm import selectinload
from orm.rbac import Role
class RoleStorage:
roles = {}
lock = asyncio.Lock()
@staticmethod
def init(session):
self = RoleStorage
roles = session.query(Role).options(selectinload(Role.permissions)).all()
self.roles = dict([(role.id, role) for role in roles])
print("[auth.roles] %d precached" % len(roles))
@staticmethod
async def get_role(id):
self = RoleStorage
async with self.lock:
return self.roles.get(id)
@staticmethod
async def add_role(role):
self = RoleStorage
async with self.lock:
self.roles[id] = role
@staticmethod
async def del_role(id):
self = RoleStorage
async with self.lock:
del self.roles[id]

72
services/auth/users.py
View File

@@ -1,72 +0,0 @@
import asyncio
from sqlalchemy.orm import selectinload, exc
from orm.user import User
from base.orm import local_session
class UserStorage:
users = {}
lock = asyncio.Lock()
@staticmethod
def init(session):
self = UserStorage
users = (
session.query(User)
.options(selectinload(User.roles), selectinload(User.ratings))
.all()
)
self.users = dict([(user.id, user) for user in users])
print("[auth.users] %d precached" % len(self.users))
@staticmethod
async def get_user(id):
with local_session() as session:
try:
user = (
session.query(User).options(
selectinload(User.roles),
selectinload(User.ratings)
).filter(
User.id == id
).one()
)
return user
except exc.NoResultFound:
return None
@staticmethod
async def get_all_users():
self = UserStorage
async with self.lock:
aaa = list(self.users.values())
aaa.sort(key=lambda user: user.createdAt)
return aaa
@staticmethod
async def get_top_users():
self = UserStorage
async with self.lock:
aaa = list(self.users.values())
aaa.sort(key=lambda user: user.rating)
return aaa
@staticmethod
async def get_user_by_slug(slug):
self = UserStorage
async with self.lock:
for user in self.users.values():
if user.slug == slug:
return user
@staticmethod
async def add_user(user):
self = UserStorage
async with self.lock:
self.users[user.id] = user
@staticmethod
async def del_user(id):
self = UserStorage
async with self.lock:
del self.users[id]

14
services/main.py
View File

@@ -1,7 +1,3 @@
# from services.stat.reacted import ReactedStorage
from services.auth.roles import RoleStorage
from services.auth.users import UserStorage
from services.zine.topics import TopicStorage
from services.search import SearchService
from services.stat.viewed import ViewedStorage
from base.orm import local_session
@@ -9,11 +5,9 @@ from base.orm import local_session
async def storages_init():
with local_session() as session:
print('[main] initialize storages')
# ReactedStorage.init(session)
RoleStorage.init(session)
UserStorage.init(session)
TopicStorage.init(session)
print('[main] initialize SearchService')
await SearchService.init(session)
session.commit()
print('[main] SearchService initialized')
print('[main] initialize storages')
await ViewedStorage.init()
print('[main] storages initialized')

72
services/stat/topicstat.py
View File

@@ -1,72 +0,0 @@
import asyncio
import time
from base.orm import local_session
from orm.shout import Shout, ShoutTopic, ShoutAuthor
from orm.topic import TopicFollower
# from sqlalchemy.sql.expression import select
class TopicStat:
# by slugs
shouts_by_topic = {} # Shout object stored
authors_by_topic = {} # User
followers_by_topic = {} # User
#
lock = asyncio.Lock()
period = 30 * 60 # sec
@staticmethod
async def load_stat(session):
print("[stat.topics] ⎧ loading stat -------")
ts = time.time()
self = TopicStat
shout_topics = session.query(ShoutTopic, Shout).join(Shout).all() # ~ 10 secs
print("[stat.topics] ⎪ shout topics joined query took %fs " % (time.time() - ts))
print("[stat.topics] ⎪ indexing %d links..." % len(shout_topics))
for [shout_topic, shout] in shout_topics:
tpc = shout_topic.topic
self.shouts_by_topic[tpc] = self.shouts_by_topic.get(tpc, dict())
self.shouts_by_topic[tpc][shout.slug] = shout
self.authors_by_topic[tpc] = self.authors_by_topic.get(tpc, dict())
authors = session.query(
ShoutAuthor.user, ShoutAuthor.caption
).filter(
ShoutAuthor.shout == shout.slug
).all()
for a in authors:
self.authors_by_topic[tpc][a[0]] = a[1]
self.followers_by_topic = {}
followings = session.query(TopicFollower).all()
print("[stat.topics] ⎪ indexing %d followings..." % len(followings))
for flw in followings:
topic = flw.topic
userslug = flw.follower
self.followers_by_topic[topic] = self.followers_by_topic.get(topic, dict())
self.followers_by_topic[topic][userslug] = userslug
@staticmethod
async def get_shouts(topic):
self = TopicStat
async with self.lock:
return self.shouts_by_topic.get(topic, dict())
@staticmethod
async def worker():
self = TopicStat
first_run = True
while True:
try:
with local_session() as session:
ts = time.time()
async with self.lock:
await self.load_stat(session)
print("[stat.topics] ⎩ load_stat took %fs " % (time.time() - ts))
except Exception as err:
raise Exception(err)
if first_run:
# sleep for period + 1 min after first run
# to distribute load on server by workers with the same period
await asyncio.sleep(60)
first_run = False
await asyncio.sleep(self.period)

26
services/stat/viewed.py
View File

@@ -5,7 +5,9 @@ from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport
from base.orm import local_session
from sqlalchemy import func
from orm.shout import ShoutTopic
from orm import User, Topic
from orm.shout import ShoutTopic, Shout
from orm.viewed import ViewedEntry
from ssl import create_default_context
from os import environ, path
@@ -113,6 +115,7 @@ class ViewedStorage:
async with self.lock:
return self.client.execute_async(load_facts)
# unused yet
@staticmethod
async def get_shout(shout_slug):
""" getting shout views metric by slug """
@@ -123,8 +126,9 @@ class ViewedStorage:
shout_views = 0
with local_session() as session:
try:
shout = session.query(Shout).where(Shout.slug == shout_slug).one()
shout_views = session.query(func.sum(ViewedEntry.amount)).where(
ViewedEntry.shout == shout_slug
ViewedEntry.shout == shout.id
).all()[0][0]
self.by_shouts[shout_slug] = shout_views
self.update_topics(session, shout_slug)
@@ -147,11 +151,12 @@ class ViewedStorage:
def update_topics(session, shout_slug):
""" updates topics counters by shout slug """
self = ViewedStorage
for t in session.query(ShoutTopic).where(ShoutTopic.shout == shout_slug).all():
tpc = t.topic
if not self.by_topics.get(tpc):
self.by_topics[tpc] = {}
self.by_topics[tpc][shout_slug] = self.by_shouts[shout_slug]
for [shout_topic, topic] in session.query(ShoutTopic, Topic).join(Topic).join(Shout).where(
Shout.slug == shout_slug
).all():
if not self.by_topics.get(topic.slug):
self.by_topics[topic.slug] = {}
self.by_topics[topic.slug][shout_slug] = self.by_shouts[shout_slug]
@staticmethod
async def increment(shout_slug, amount=1, viewer='anonymous'):
@@ -159,9 +164,12 @@ class ViewedStorage:
self = ViewedStorage
async with self.lock:
with local_session() as session:
shout = session.query(Shout).where(Shout.slug == shout_slug).one()
viewer = session.query(User).where(User.slug == viewer).one()
viewed = ViewedEntry.create(**{
"viewer": viewer,
"shout": shout_slug,
"viewerId": viewer.id,
"shout": shout.id,
"amount": amount
})
session.add(viewed)

49
services/zine/shoutauthor.py
View File

@@ -1,49 +0,0 @@
import asyncio
import time
from base.orm import local_session
from orm.shout import ShoutAuthor
class ShoutAuthorStorage:
authors_by_shout = {}
lock = asyncio.Lock()
# period = 30 * 60 # sec
@staticmethod
async def load_captions(session):
self = ShoutAuthorStorage
sas = session.query(ShoutAuthor).all()
for sa in sas:
self.authors_by_shout[sa.shout] = self.authors_by_shout.get(sa.shout, {})
self.authors_by_shout[sa.shout][sa.user] = sa.caption
print("[zine.authors] ⎧ %d shouts indexed by authors" % len(self.authors_by_shout))
@staticmethod
async def get_author_caption(shout, author):
self = ShoutAuthorStorage
async with self.lock:
return self.authors_by_shout.get(shout, {}).get(author)
@staticmethod
async def set_author_caption(shout, author, caption):
self = ShoutAuthorStorage
async with self.lock:
self.authors_by_shout[shout] = self.authors_by_shout.get(shout, {})
self.authors_by_shout[shout][author] = caption
return {
"error": None,
}
@staticmethod
async def worker():
self = ShoutAuthorStorage
async with self.lock:
# while True:
try:
with local_session() as session:
ts = time.time()
await self.load_captions(session)
print("[zine.authors] ⎩ load_captions took %fs " % (time.time() - ts))
except Exception as err:
print("[zine.authors] ⎩ error indexing by author: %s" % (err))
# await asyncio.sleep(self.period)

97
services/zine/topics.py
View File

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