240 lines
9.7 KiB
Python
240 lines
9.7 KiB
Python
import asyncio
|
||
import json
|
||
from typing import List
|
||
|
||
from sqlalchemy import select, join
|
||
from orm.author import Author, AuthorFollower
|
||
from orm.topic import Topic, TopicFollower
|
||
from services.db import local_session
|
||
from services.encoders import CustomJSONEncoder
|
||
from services.rediscache import redis
|
||
from services.logger import root_logger as logger
|
||
|
||
DEFAULT_FOLLOWS = {
|
||
"topics": [],
|
||
"authors": [],
|
||
"communities": [{"id": 1, "name": "Дискурс", "slug": "discours", "pic": ""}],
|
||
}
|
||
|
||
|
||
# Кэширование данных темы
|
||
async def cache_topic(topic: dict):
|
||
payload = json.dumps(topic, cls=CustomJSONEncoder)
|
||
# Одновременное кэширование по id и slug для быстрого доступа
|
||
await asyncio.gather(
|
||
redis.set(f"topic:id:{topic['id']}", payload), redis.set(f"topic:slug:{topic['slug']}", payload)
|
||
)
|
||
|
||
|
||
# Кэширование данных автора
|
||
async def cache_author(author: dict):
|
||
payload = json.dumps(author, cls=CustomJSONEncoder)
|
||
# Кэширование данных автора по user и id
|
||
await asyncio.gather(
|
||
redis.set(f"author:user:{author['user'].strip()}", str(author["id"])),
|
||
redis.set(f"author:id:{author['id']}", payload),
|
||
)
|
||
|
||
|
||
# Кэширование данных о подписках
|
||
async def cache_follows(follower_id: int, entity_type: str, entity_id: int, is_insert=True):
|
||
key = f"author:follows-{entity_type}s:{follower_id}"
|
||
follows_str = await redis.get(key)
|
||
follows = json.loads(follows_str) if follows_str else []
|
||
if is_insert:
|
||
if entity_id not in follows:
|
||
follows.append(entity_id)
|
||
else:
|
||
follows = [eid for eid in follows if eid != entity_id]
|
||
await redis.set(key, json.dumps(follows, cls=CustomJSONEncoder))
|
||
update_follower_stat(follower_id, entity_type, len(follows))
|
||
|
||
|
||
# Обновление статистики подписчика
|
||
async def update_follower_stat(follower_id, entity_type, count):
|
||
follower_key = f"author:id:{follower_id}"
|
||
follower_str = await redis.get(follower_key)
|
||
follower = json.loads(follower_str) if follower_str else None
|
||
if follower:
|
||
follower["stat"] = {f"{entity_type}s": count}
|
||
await cache_author(follower)
|
||
|
||
|
||
# Получение автора из кэша
|
||
async def get_cached_author(author_id: int):
|
||
author_key = f"author:id:{author_id}"
|
||
result = await redis.get(author_key)
|
||
if result:
|
||
return json.loads(result)
|
||
# Загрузка из базы данных, если не найдено в кэше
|
||
with local_session() as session:
|
||
author = session.execute(select(Author).where(Author.id == author_id)).scalar_one_or_none()
|
||
if author:
|
||
await cache_author(author.dict())
|
||
return author.dict()
|
||
return None
|
||
|
||
|
||
# Получение темы по slug из кэша
|
||
async def get_cached_topic_by_slug(slug: str):
|
||
topic_key = f"topic:slug:{slug}"
|
||
result = await redis.get(topic_key)
|
||
if result:
|
||
return json.loads(result)
|
||
# Загрузка из базы данных, если не найдено в кэше
|
||
with local_session() as session:
|
||
topic = session.execute(select(Topic).where(Topic.slug == slug)).scalar_one_or_none()
|
||
if topic:
|
||
await cache_topic(topic.dict())
|
||
return topic.dict()
|
||
return None
|
||
|
||
|
||
# Получение списка авторов по ID из кэша
|
||
async def get_cached_authors_by_ids(author_ids: List[int]) -> List[dict]:
|
||
# Одновременное получение данных всех авторов
|
||
keys = [f"author:id:{author_id}" for author_id in author_ids]
|
||
results = await asyncio.gather(*(redis.get(key) for key in keys))
|
||
authors = [json.loads(result) if result else None for result in results]
|
||
# Загрузка отсутствующих авторов из базы данных и кэширование
|
||
missing_indices = [index for index, author in enumerate(authors) if author is None]
|
||
if missing_indices:
|
||
missing_ids = [author_ids[index] for index in missing_indices]
|
||
with local_session() as session:
|
||
query = select(Author).where(Author.id.in_(missing_ids))
|
||
missing_authors = session.execute(query).scalars().all()
|
||
await asyncio.gather(*(cache_author(author.dict()) for author in missing_authors))
|
||
authors = [author.dict() for author in missing_authors]
|
||
return authors
|
||
|
||
|
||
async def get_cached_topic_followers(topic_id: int):
|
||
# Попытка извлечь кэшированные данные
|
||
cached = await redis.get(f"topic:followers:{topic_id}")
|
||
if cached:
|
||
followers = json.loads(cached)
|
||
logger.debug(f"Cached followers for topic#{topic_id}: {len(followers)}")
|
||
return followers
|
||
|
||
# Загрузка из базы данных и кэширование результатов
|
||
with local_session() as session:
|
||
followers_ids = [
|
||
f[0]
|
||
for f in session.query(Author.id)
|
||
.join(TopicFollower, TopicFollower.follower == Author.id)
|
||
.filter(TopicFollower.topic == topic_id)
|
||
.all()
|
||
]
|
||
await redis.set(f"topic:followers:{topic_id}", json.dumps(followers_ids))
|
||
followers = await get_cached_authors_by_ids(followers_ids)
|
||
return followers
|
||
|
||
|
||
async def get_cached_author_followers(author_id: int):
|
||
# Проверяем кэш на наличие данных
|
||
cached = await redis.get(f"author:followers:{author_id}")
|
||
if cached:
|
||
followers_ids = json.loads(cached)
|
||
followers = await get_cached_authors_by_ids(followers_ids)
|
||
logger.debug(f"Cached followers for author#{author_id}: {len(followers)}")
|
||
return followers
|
||
|
||
# Запрос в базу данных если кэш пуст
|
||
with local_session() as session:
|
||
followers_ids = [
|
||
f[0]
|
||
for f in session.query(Author.id)
|
||
.join(AuthorFollower, AuthorFollower.follower == Author.id)
|
||
.filter(AuthorFollower.author == author_id, Author.id != author_id)
|
||
.all()
|
||
]
|
||
await redis.set(f"author:followers:{author_id}", json.dumps(followers_ids))
|
||
followers = await get_cached_authors_by_ids(followers_ids)
|
||
return followers
|
||
|
||
|
||
async def get_cached_follower_authors(author_id: int):
|
||
# Попытка получить авторов из кэша
|
||
cached = await redis.get(f"author:follows-authors:{author_id}")
|
||
if cached:
|
||
authors_ids = json.loads(cached)
|
||
else:
|
||
# Запрос авторов из базы данных
|
||
with local_session() as session:
|
||
authors_ids = [
|
||
a[0]
|
||
for a in session.execute(
|
||
select(Author.id)
|
||
.select_from(join(Author, AuthorFollower, Author.id == AuthorFollower.author))
|
||
.where(AuthorFollower.follower == author_id)
|
||
).all()
|
||
]
|
||
await redis.set(f"author:follows-authors:{author_id}", json.dumps(authors_ids))
|
||
|
||
authors = await get_cached_authors_by_ids(authors_ids)
|
||
return authors
|
||
|
||
|
||
async def get_cached_follower_topics(author_id: int):
|
||
# Попытка получить темы из кэша
|
||
cached = await redis.get(f"author:follows-topics:{author_id}")
|
||
if cached:
|
||
topics_ids = json.loads(cached)
|
||
else:
|
||
# Загрузка тем из базы данных и их кэширование
|
||
with local_session() as session:
|
||
topics_ids = [
|
||
t[0]
|
||
for t in session.query(Topic.id)
|
||
.join(TopicFollower, TopicFollower.topic == Topic.id)
|
||
.where(TopicFollower.follower == author_id)
|
||
.all()
|
||
]
|
||
await redis.set(f"author:follows-topics:{author_id}", json.dumps(topics_ids))
|
||
|
||
topics = []
|
||
for topic_id in topics_ids:
|
||
topic_str = await redis.get(f"topic:id:{topic_id}")
|
||
if topic_str:
|
||
topic = json.loads(topic_str)
|
||
if topic and topic not in topics:
|
||
topics.append(topic)
|
||
|
||
logger.debug(f"Cached topics for author#{author_id}: {len(topics)}")
|
||
return topics
|
||
|
||
|
||
async def get_cached_author_by_user_id(user_id: str):
|
||
"""
|
||
Получает информацию об авторе по его user_id, сначала проверяя кэш, а затем базу данных.
|
||
|
||
Args:
|
||
user_id (str): Идентификатор пользователя, по которому нужно получить автора.
|
||
|
||
Returns:
|
||
dict: Словарь с данными автора или None, если автор не найден.
|
||
"""
|
||
# Пытаемся найти ID автора по user_id в кэше Redis
|
||
author_id = await redis.get(f"author:user:{user_id.strip()}")
|
||
if author_id:
|
||
# Если ID найден, получаем полные данные автора по его ID
|
||
author_data = await redis.get(f"author:id:{author_id}")
|
||
if author_data:
|
||
return json.loads(author_data)
|
||
|
||
# Если данные в кэше не найдены, выполняем запрос к базе данных
|
||
with local_session() as session:
|
||
author = session.execute(select(Author).where(Author.user == user_id)).scalar_one_or_none()
|
||
|
||
if author:
|
||
# Кэшируем полученные данные автора
|
||
author_dict = author.dict()
|
||
await asyncio.gather(
|
||
redis.set(f"author:user:{user_id.strip()}", str(author.id)),
|
||
redis.set(f"author:id:{author.id}", json.dumps(author_dict)),
|
||
)
|
||
return author_dict
|
||
|
||
# Возвращаем None, если автор не найден
|
||
return None
|