This commit is contained in:
parent
bdf78bb45d
commit
b0136fd9bc
|
@ -10,7 +10,7 @@ from orm.author import Author, AuthorFollower
|
||||||
from orm.shout import ShoutAuthor, ShoutTopic
|
from orm.shout import ShoutAuthor, ShoutTopic
|
||||||
from orm.topic import Topic
|
from orm.topic import Topic
|
||||||
from resolvers.stat import get_with_stat, author_follows_authors, author_follows_topics
|
from resolvers.stat import get_with_stat, author_follows_authors, author_follows_topics
|
||||||
from services.cache import update_author_cache, update_author_followers_cache
|
from services.cache import set_author_cache, update_author_followers_cache
|
||||||
from services.auth import login_required
|
from services.auth import login_required
|
||||||
from services.db import local_session
|
from services.db import local_session
|
||||||
from services.rediscache import redis
|
from services.rediscache import redis
|
||||||
|
@ -59,7 +59,7 @@ async def get_author(_, _info, slug='', author_id=None):
|
||||||
author_dict = author.dict()
|
author_dict = author.dict()
|
||||||
logger.debug(f'author to be stored: {author_dict}')
|
logger.debug(f'author to be stored: {author_dict}')
|
||||||
if author:
|
if author:
|
||||||
await update_author_cache(author_dict)
|
await set_author_cache(author_dict)
|
||||||
return author_dict
|
return author_dict
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
import traceback
|
import traceback
|
||||||
|
@ -88,7 +88,7 @@ async def get_author_by_user_id(user_id: str):
|
||||||
|
|
||||||
[author] = get_with_stat(q)
|
[author] = get_with_stat(q)
|
||||||
if author:
|
if author:
|
||||||
await update_author_cache(author.dict())
|
await set_author_cache(author.dict())
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
logger.error(exc)
|
logger.error(exc)
|
||||||
return author
|
return author
|
||||||
|
|
|
@ -26,6 +26,7 @@ from services.rediscache import redis
|
||||||
@mutation.field('follow')
|
@mutation.field('follow')
|
||||||
@login_required
|
@login_required
|
||||||
async def follow(_, info, what, slug):
|
async def follow(_, info, what, slug):
|
||||||
|
follows = None
|
||||||
try:
|
try:
|
||||||
user_id = info.context['user_id']
|
user_id = info.context['user_id']
|
||||||
follower_query = select(Author).select_from(Author).filter(Author.user == user_id)
|
follower_query = select(Author).select_from(Author).filter(Author.user == user_id)
|
||||||
|
@ -36,14 +37,14 @@ async def follow(_, info, what, slug):
|
||||||
author_query = select(Author).select_from(Author).where(Author.slug == slug)
|
author_query = select(Author).select_from(Author).where(Author.slug == slug)
|
||||||
[author] = get_with_stat(author_query)
|
[author] = get_with_stat(author_query)
|
||||||
if author:
|
if author:
|
||||||
await update_follows_for_author(follower, 'author', author, True)
|
follows = await update_follows_for_author(follower, 'author', author, True)
|
||||||
await update_followers_for_author(follower, author, True)
|
await update_followers_for_author(follower, author, True)
|
||||||
await notify_follower(follower.dict(), author.id, 'unfollow')
|
await notify_follower(follower.dict(), author.id, 'unfollow')
|
||||||
elif what == 'TOPIC':
|
elif what == 'TOPIC':
|
||||||
topic_query = select(Topic).where(Topic.slug == slug)
|
topic_query = select(Topic).where(Topic.slug == slug)
|
||||||
[topic] = get_with_stat(topic_query)
|
[topic] = get_with_stat(topic_query)
|
||||||
if topic:
|
if topic:
|
||||||
await update_follows_for_author(follower, 'topic', topic, True)
|
follows = await update_follows_for_author(follower, 'topic', topic, True)
|
||||||
topic_unfollow(follower.id, slug)
|
topic_unfollow(follower.id, slug)
|
||||||
elif what == 'COMMUNITY':
|
elif what == 'COMMUNITY':
|
||||||
community_follow(follower.id, slug)
|
community_follow(follower.id, slug)
|
||||||
|
@ -52,7 +53,7 @@ async def follow(_, info, what, slug):
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.debug(info, what, slug)
|
logger.debug(info, what, slug)
|
||||||
logger.error(e)
|
logger.error(e)
|
||||||
return {'error': str(e)}
|
return {'error': str(e), f'{what}s': follows}
|
||||||
|
|
||||||
return {}
|
return {}
|
||||||
|
|
||||||
|
@ -60,8 +61,9 @@ async def follow(_, info, what, slug):
|
||||||
@mutation.field('unfollow')
|
@mutation.field('unfollow')
|
||||||
@login_required
|
@login_required
|
||||||
async def unfollow(_, info, what, slug):
|
async def unfollow(_, info, what, slug):
|
||||||
user_id = info.context['user_id']
|
follows = None
|
||||||
try:
|
try:
|
||||||
|
user_id = info.context.get('user_id')
|
||||||
follower_query = select(Author).filter(Author.user == user_id)
|
follower_query = select(Author).filter(Author.user == user_id)
|
||||||
[follower] = get_with_stat(follower_query)
|
[follower] = get_with_stat(follower_query)
|
||||||
if follower:
|
if follower:
|
||||||
|
@ -71,20 +73,20 @@ async def unfollow(_, info, what, slug):
|
||||||
[author] = get_with_stat(author_query)
|
[author] = get_with_stat(author_query)
|
||||||
if author:
|
if author:
|
||||||
await update_follows_for_author(follower, 'author', author, False)
|
await update_follows_for_author(follower, 'author', author, False)
|
||||||
await update_followers_for_author(follower, author, False)
|
follows = await update_followers_for_author(follower, author, False)
|
||||||
await notify_follower(follower.dict(), author.id, 'unfollow')
|
await notify_follower(follower.dict(), author.id, 'unfollow')
|
||||||
elif what == 'TOPIC':
|
elif what == 'TOPIC':
|
||||||
topic_query = select(Topic).where(Topic.slug == slug)
|
topic_query = select(Topic).where(Topic.slug == slug)
|
||||||
[topic] = get_with_stat(topic_query)
|
[topic] = get_with_stat(topic_query)
|
||||||
if topic:
|
if topic:
|
||||||
await update_follows_for_author(follower, 'topic', topic, False)
|
follows = await update_follows_for_author(follower, 'topic', topic, False)
|
||||||
topic_unfollow(follower.id, slug)
|
topic_unfollow(follower.id, slug)
|
||||||
elif what == 'COMMUNITY':
|
elif what == 'COMMUNITY':
|
||||||
community_unfollow(follower.id, slug)
|
community_unfollow(follower.id, slug)
|
||||||
elif what == 'REACTIONS':
|
elif what == 'REACTIONS':
|
||||||
reactions_unfollow(follower.id, slug)
|
reactions_unfollow(follower.id, slug)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
return {'error': str(e)}
|
return {'error': str(e), f'{what}s': follows}
|
||||||
|
|
||||||
return {}
|
return {}
|
||||||
|
|
||||||
|
|
|
@ -19,7 +19,7 @@ DEFAULT_FOLLOWS = {
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
async def update_author_cache(author: dict, ttl=25 * 60 * 60):
|
async def set_author_cache(author: dict, ttl=25 * 60 * 60):
|
||||||
payload = json.dumps(author)
|
payload = json.dumps(author)
|
||||||
await redis.execute('SETEX', f'user:{author.get("user")}:author', ttl, payload)
|
await redis.execute('SETEX', f'user:{author.get("user")}:author', ttl, payload)
|
||||||
await redis.execute('SETEX', f'id:{author.get("id")}:author', ttl, payload)
|
await redis.execute('SETEX', f'id:{author.get("id")}:author', ttl, payload)
|
||||||
|
@ -30,7 +30,7 @@ async def update_author_followers_cache(author_id: int, followers, ttl=25 * 60 *
|
||||||
await redis.execute('SETEX', f'author:{author_id}:followers', ttl, payload)
|
await redis.execute('SETEX', f'author:{author_id}:followers', ttl, payload)
|
||||||
|
|
||||||
|
|
||||||
async def update_follows_topics_cache(follows, author_id: int, ttl=25 * 60 * 60):
|
async def set_follows_topics_cache(follows, author_id: int, ttl=25 * 60 * 60):
|
||||||
try:
|
try:
|
||||||
payload = json.dumps(follows)
|
payload = json.dumps(follows)
|
||||||
await redis.execute('SETEX', f'author:{author_id}:follows-topics', ttl, payload)
|
await redis.execute('SETEX', f'author:{author_id}:follows-topics', ttl, payload)
|
||||||
|
@ -42,7 +42,7 @@ async def update_follows_topics_cache(follows, author_id: int, ttl=25 * 60 * 60)
|
||||||
logger.error(exc)
|
logger.error(exc)
|
||||||
|
|
||||||
|
|
||||||
async def update_follows_authors_cache(follows, author_id: int, ttl=25 * 60 * 60):
|
async def set_follows_authors_cache(follows, author_id: int, ttl=25 * 60 * 60):
|
||||||
try:
|
try:
|
||||||
payload = json.dumps(follows)
|
payload = json.dumps(follows)
|
||||||
await redis.execute('SETEX', f'author:{author_id}:follows-authors', ttl, payload)
|
await redis.execute('SETEX', f'author:{author_id}:follows-authors', ttl, payload)
|
||||||
|
@ -53,6 +53,35 @@ async def update_follows_authors_cache(follows, author_id: int, ttl=25 * 60 * 60
|
||||||
logger.error(exc)
|
logger.error(exc)
|
||||||
|
|
||||||
|
|
||||||
|
async def update_follows_for_author(follower: Author, entity_type: str, entity: dict, is_insert: bool):
|
||||||
|
redis_key = f'author:{follower.id}:follows-{entity_type}s'
|
||||||
|
follows_str = await redis.get(redis_key)
|
||||||
|
follows = json.loads(follows_str) if follows_str else []
|
||||||
|
if is_insert:
|
||||||
|
follows.append(entity)
|
||||||
|
else:
|
||||||
|
# Remove the entity from follows
|
||||||
|
follows = [e for e in follows if e['id'] != entity['id']]
|
||||||
|
if entity_type == 'topic':
|
||||||
|
await set_follows_topics_cache(follows, follower.id)
|
||||||
|
if entity_type == 'author':
|
||||||
|
await set_follows_authors_cache(follows, follower.id)
|
||||||
|
return follows
|
||||||
|
|
||||||
|
|
||||||
|
async def update_followers_for_author(follower: Author, author: Author, is_insert: bool):
|
||||||
|
redis_key = f'author:{author.id}:followers'
|
||||||
|
followers_str = await redis.get(redis_key)
|
||||||
|
followers = json.loads(followers_str) if followers_str else []
|
||||||
|
if is_insert:
|
||||||
|
followers.append(follower)
|
||||||
|
else:
|
||||||
|
# Remove the entity from followers
|
||||||
|
followers = [e for e in followers if e['id'] != author.id]
|
||||||
|
await update_author_followers_cache(author.id, followers)
|
||||||
|
return followers
|
||||||
|
|
||||||
|
|
||||||
@event.listens_for(Shout, 'after_insert')
|
@event.listens_for(Shout, 'after_insert')
|
||||||
@event.listens_for(Shout, 'after_update')
|
@event.listens_for(Shout, 'after_update')
|
||||||
def after_shouts_update(mapper, connection, shout: Shout):
|
def after_shouts_update(mapper, connection, shout: Shout):
|
||||||
|
@ -65,7 +94,7 @@ def after_shouts_update(mapper, connection, shout: Shout):
|
||||||
)
|
)
|
||||||
|
|
||||||
for author_with_stat in get_with_stat(authors_query):
|
for author_with_stat in get_with_stat(authors_query):
|
||||||
asyncio.create_task(update_author_cache(author_with_stat.dict()))
|
asyncio.create_task(set_author_cache(author_with_stat.dict()))
|
||||||
|
|
||||||
|
|
||||||
@event.listens_for(Reaction, 'after_insert')
|
@event.listens_for(Reaction, 'after_insert')
|
||||||
|
@ -91,7 +120,7 @@ def after_reaction_insert(mapper, connection, reaction: Reaction):
|
||||||
)
|
)
|
||||||
|
|
||||||
for author_with_stat in get_with_stat(author_query):
|
for author_with_stat in get_with_stat(author_query):
|
||||||
asyncio.create_task(update_author_cache(author_with_stat.dict()))
|
asyncio.create_task(set_author_cache(author_with_stat.dict()))
|
||||||
|
|
||||||
shout = connection.execute(select(Shout).select_from(Shout).where(Shout.id == reaction.shout)).first()
|
shout = connection.execute(select(Shout).select_from(Shout).where(Shout.id == reaction.shout)).first()
|
||||||
if shout:
|
if shout:
|
||||||
|
@ -105,7 +134,7 @@ def after_reaction_insert(mapper, connection, reaction: Reaction):
|
||||||
def after_author_update(mapper, connection, author: Author):
|
def after_author_update(mapper, connection, author: Author):
|
||||||
q = select(Author).where(Author.id == author.id)
|
q = select(Author).where(Author.id == author.id)
|
||||||
[author_with_stat] = get_with_stat(q)
|
[author_with_stat] = get_with_stat(q)
|
||||||
asyncio.create_task(update_author_cache(author_with_stat.dict()))
|
asyncio.create_task(set_author_cache(author_with_stat.dict()))
|
||||||
|
|
||||||
|
|
||||||
@event.listens_for(TopicFollower, 'after_insert')
|
@event.listens_for(TopicFollower, 'after_insert')
|
||||||
|
@ -136,32 +165,6 @@ def after_author_follower_delete(mapper, connection, target: AuthorFollower):
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
async def update_follows_for_author(follower: Author, entity_type: str, entity: dict, is_insert: bool):
|
|
||||||
ttl = 25 * 60 * 60
|
|
||||||
redis_key = f'author:{follower.id}:follows-{entity_type}s'
|
|
||||||
follows_str = await redis.get(redis_key)
|
|
||||||
follows = json.loads(follows_str) if follows_str else []
|
|
||||||
if is_insert:
|
|
||||||
follows.append(entity)
|
|
||||||
else:
|
|
||||||
# Remove the entity from follows
|
|
||||||
follows = [e for e in follows if e['id'] != entity['id']]
|
|
||||||
await redis.execute('SETEX', redis_key, ttl, json.dumps(follows))
|
|
||||||
|
|
||||||
|
|
||||||
async def update_followers_for_author(follower: Author, author: Author, is_insert: bool):
|
|
||||||
ttl = 25 * 60 * 60
|
|
||||||
redis_key = f'author:{author.id}:followers'
|
|
||||||
followers_str = await redis.get(redis_key)
|
|
||||||
followers = json.loads(followers_str) if followers_str else []
|
|
||||||
if is_insert:
|
|
||||||
followers.append(follower)
|
|
||||||
else:
|
|
||||||
# Remove the entity from followers
|
|
||||||
followers = [e for e in followers if e['id'] != author.id]
|
|
||||||
await redis.execute('SETEX', redis_key, ttl, json.dumps(followers))
|
|
||||||
|
|
||||||
|
|
||||||
async def handle_author_follower_change(
|
async def handle_author_follower_change(
|
||||||
connection, author_id: int, follower_id: int, is_insert: bool
|
connection, author_id: int, follower_id: int, is_insert: bool
|
||||||
):
|
):
|
||||||
|
@ -170,14 +173,14 @@ async def handle_author_follower_change(
|
||||||
follower_query = select(Author).select_from(Author).filter(Author.id == follower_id)
|
follower_query = select(Author).select_from(Author).filter(Author.id == follower_id)
|
||||||
follower = get_with_stat(follower_query)
|
follower = get_with_stat(follower_query)
|
||||||
if follower and author:
|
if follower and author:
|
||||||
_ = asyncio.create_task(update_author_cache(author.dict()))
|
_ = asyncio.create_task(set_author_cache(author.dict()))
|
||||||
follows_authors = await redis.execute('GET', f'author:{follower_id}:follows-authors')
|
follows_authors = await redis.execute('GET', f'author:{follower_id}:follows-authors')
|
||||||
if follows_authors:
|
if follows_authors:
|
||||||
follows_authors = json.loads(follows_authors)
|
follows_authors = json.loads(follows_authors)
|
||||||
if not any(x.get('id') == author.id for x in follows_authors):
|
if not any(x.get('id') == author.id for x in follows_authors):
|
||||||
follows_authors.append(author.dict())
|
follows_authors.append(author.dict())
|
||||||
_ = asyncio.create_task(update_follows_authors_cache(follows_authors, follower_id))
|
_ = asyncio.create_task(set_follows_authors_cache(follows_authors, follower_id))
|
||||||
_ = asyncio.create_task(update_author_cache(follower.dict()))
|
_ = asyncio.create_task(set_author_cache(follower.dict()))
|
||||||
await update_follows_for_author(
|
await update_follows_for_author(
|
||||||
connection,
|
connection,
|
||||||
follower,
|
follower,
|
||||||
|
@ -203,13 +206,13 @@ async def handle_topic_follower_change(
|
||||||
follower_query = select(Author).filter(Author.id == follower_id)
|
follower_query = select(Author).filter(Author.id == follower_id)
|
||||||
follower = get_with_stat(follower_query)
|
follower = get_with_stat(follower_query)
|
||||||
if follower and topic:
|
if follower and topic:
|
||||||
_ = asyncio.create_task(update_author_cache(follower.dict()))
|
_ = asyncio.create_task(set_author_cache(follower.dict()))
|
||||||
follows_topics = await redis.execute('GET', f'author:{follower_id}:follows-topics')
|
follows_topics = await redis.execute('GET', f'author:{follower_id}:follows-topics')
|
||||||
if follows_topics:
|
if follows_topics:
|
||||||
follows_topics = json.loads(follows_topics)
|
follows_topics = json.loads(follows_topics)
|
||||||
if not any(x.get('id') == topic.id for x in follows_topics):
|
if not any(x.get('id') == topic.id for x in follows_topics):
|
||||||
follows_topics.append(topic)
|
follows_topics.append(topic)
|
||||||
_ = asyncio.create_task(update_follows_topics_cache(follows_topics, follower_id))
|
_ = asyncio.create_task(set_follows_topics_cache(follows_topics, follower_id))
|
||||||
await update_follows_for_author(
|
await update_follows_for_author(
|
||||||
follower,
|
follower,
|
||||||
'topic',
|
'topic',
|
||||||
|
|
Loading…
Reference in New Issue
Block a user