359 lines
17 KiB
Python
359 lines
17 KiB
Python
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from graphql import GraphQLResolveInfo
|
||
from sqlalchemy.sql import and_
|
||
|
||
from cache.cache import (
|
||
cache_author,
|
||
cache_topic,
|
||
get_cached_follower_authors,
|
||
get_cached_follower_topics,
|
||
)
|
||
from orm.author import Author, AuthorFollower
|
||
from orm.community import Community, CommunityFollower
|
||
from orm.shout import Shout, ShoutReactionsFollower
|
||
from orm.topic import Topic, TopicFollower
|
||
from resolvers.author import invalidate_authors_cache
|
||
from services.auth import login_required
|
||
from services.notify import notify_follower
|
||
from storage.db import local_session
|
||
from storage.redis import redis
|
||
from storage.schema import mutation, query
|
||
from utils.logger import root_logger as logger
|
||
|
||
|
||
def get_entity_field_name(entity_type: str) -> str:
|
||
"""Возвращает имя поля для связи с сущностью в модели подписчика"""
|
||
entity_field_mapping = {"author": "following", "topic": "topic", "community": "community", "shout": "shout"}
|
||
if entity_type not in entity_field_mapping:
|
||
msg = f"Unknown entity_type: {entity_type}"
|
||
raise ValueError(msg)
|
||
return entity_field_mapping[entity_type]
|
||
|
||
|
||
@mutation.field("follow")
|
||
@login_required
|
||
async def follow(
|
||
_: None, info: GraphQLResolveInfo, what: str, slug: str = "", entity_id: int | None = None
|
||
) -> dict[str, Any]:
|
||
logger.debug("Начало выполнения функции 'follow'")
|
||
viewer_id = info.context.get("author", {}).get("id")
|
||
follower_dict = info.context.get("author") or {}
|
||
|
||
# ✅ КРИТИЧНО: Инвалидируем кеш В САМОМ НАЧАЛЕ, если пользователь авторизован
|
||
# чтобы предотвратить чтение старых данных при последующей перезагрузке
|
||
if viewer_id:
|
||
entity_type = what.lower()
|
||
cache_key_pattern = f"author:follows-{entity_type}s:{viewer_id}"
|
||
await redis.execute("DEL", cache_key_pattern)
|
||
await redis.execute("DEL", f"author:id:{viewer_id}")
|
||
logger.debug(f"Инвалидирован кеш подписок follower'а: {cache_key_pattern}")
|
||
|
||
if not viewer_id:
|
||
return {"error": "Access denied"}
|
||
|
||
logger.debug(f"follower: {follower_dict}")
|
||
|
||
if not viewer_id or not follower_dict:
|
||
logger.warning("Неавторизованный доступ при попытке подписаться")
|
||
return {"error": "UnauthorizedError"}
|
||
|
||
follower_id = follower_dict.get("id")
|
||
logger.debug(f"follower_id: {follower_id}")
|
||
|
||
entity_classes = {
|
||
"AUTHOR": (Author, AuthorFollower, get_cached_follower_authors, cache_author),
|
||
"TOPIC": (Topic, TopicFollower, get_cached_follower_topics, cache_topic),
|
||
"COMMUNITY": (Community, CommunityFollower, None, None), # Нет методов кэша для сообщества
|
||
"SHOUT": (Shout, ShoutReactionsFollower, None, None), # Нет методов кэша для shout
|
||
}
|
||
|
||
if what not in entity_classes:
|
||
logger.error(f"Неверный тип для следования: {what}")
|
||
return {"error": "invalid follow type"}
|
||
|
||
entity_class, follower_class, get_cached_follows_method, cache_method = entity_classes[what]
|
||
entity_type = what.lower()
|
||
follows: list[dict[str, Any]] = []
|
||
error: str | None = None
|
||
|
||
# ✅ Сохраняем entity_id и error вне сессии для использования после её закрытия
|
||
entity_id_result: int | None = None
|
||
error_result: str | None = None
|
||
|
||
try:
|
||
logger.debug("Попытка получить сущность из базы данных")
|
||
with local_session() as session:
|
||
# Используем query для получения сущности
|
||
entity_query = session.query(entity_class)
|
||
|
||
# Проверяем наличие slug перед фильтрацией
|
||
if hasattr(entity_class, "slug"):
|
||
entity_query = entity_query.where(entity_class.slug == slug)
|
||
|
||
entity = entity_query.first()
|
||
|
||
if not entity:
|
||
logger.warning(f"{what.lower()} не найден по slug: {slug}")
|
||
return {"error": f"{what.lower()} not found"}
|
||
|
||
# Получаем ID сущности
|
||
if entity_id is None:
|
||
entity_id = getattr(entity, "id", None)
|
||
|
||
if not entity_id:
|
||
logger.warning(f"Не удалось получить ID для {what.lower()}")
|
||
return {"error": f"Cannot get ID for {what.lower()}"}
|
||
|
||
# Если это автор, учитываем фильтрацию данных
|
||
entity_dict = entity.dict() if hasattr(entity, "dict") else {}
|
||
|
||
logger.debug(f"entity_id: {entity_id}, entity_dict: {entity_dict}")
|
||
|
||
if entity_id is not None and isinstance(entity_id, int):
|
||
entity_field = get_entity_field_name(entity_type)
|
||
logger.debug(f"entity_type: {entity_type}, entity_field: {entity_field}")
|
||
|
||
existing_sub = (
|
||
session.query(follower_class)
|
||
.where(
|
||
follower_class.follower == follower_id, # type: ignore[attr-defined]
|
||
getattr(follower_class, entity_field) == entity_id, # type: ignore[attr-defined]
|
||
)
|
||
.first()
|
||
)
|
||
|
||
if existing_sub:
|
||
logger.info(f"Пользователь {follower_id} уже подписан на {what.lower()} с ID {entity_id}")
|
||
error_result = "already following"
|
||
# ✅ КРИТИЧНО: Не делаем return - продолжаем для получения списка подписок
|
||
else:
|
||
logger.debug("Добавление новой записи в базу данных")
|
||
sub = follower_class(follower=follower_id, **{entity_field: entity_id})
|
||
logger.debug(f"Создан объект подписки: {sub}")
|
||
session.add(sub)
|
||
session.commit()
|
||
logger.info(f"Пользователь {follower_id} подписался на {what.lower()} с ID {entity_id}")
|
||
|
||
if cache_method:
|
||
logger.debug("Обновление кэша сущности")
|
||
await cache_method(entity_dict)
|
||
|
||
if what == "AUTHOR":
|
||
logger.debug("Отправка уведомления автору о подписке")
|
||
if isinstance(follower_dict, dict) and isinstance(entity_id, int):
|
||
# Получаем ID созданной записи подписки
|
||
subscription_id = getattr(sub, "id", None) if "sub" in locals() else None
|
||
await notify_follower(
|
||
follower=follower_dict,
|
||
author_id=entity_id,
|
||
action="follow",
|
||
subscription_id=subscription_id,
|
||
)
|
||
|
||
# ✅ КРИТИЧНО: Инвалидируем кеш списка подписчиков автора
|
||
# чтобы новый подписчик сразу появился в списке
|
||
await redis.execute("DEL", f"author:followers:{entity_id}")
|
||
logger.debug(f"Инвалидирован кеш подписчиков автора: author:followers:{entity_id}")
|
||
|
||
# Инвалидируем кеш статистики авторов для обновления счетчиков подписчиков
|
||
logger.debug("Инвалидируем кеш статистики авторов")
|
||
await invalidate_authors_cache(entity_id)
|
||
|
||
entity_id_result = entity_id
|
||
|
||
# ✅ Получаем актуальный список подписок для возврата клиенту
|
||
# Кеш уже инвалидирован в начале функции, поэтому get_cached_follows_method
|
||
# вернет свежие данные из БД
|
||
if get_cached_follows_method and isinstance(follower_id, int):
|
||
logger.debug("Получение актуального списка подписок после закрытия сессии")
|
||
existing_follows = await get_cached_follows_method(follower_id)
|
||
logger.debug(
|
||
f"Получено подписок: {len(existing_follows)}, содержит target={entity_id_result in [f.get('id') for f in existing_follows] if existing_follows else False}"
|
||
)
|
||
|
||
# Если это авторы, получаем безопасную версию
|
||
if what == "AUTHOR":
|
||
follows_filtered = []
|
||
|
||
for author_data in existing_follows:
|
||
# Создаем объект автора для использования метода dict
|
||
temp_author = Author()
|
||
for key, value in author_data.items():
|
||
if (
|
||
hasattr(temp_author, key) and key != "username"
|
||
): # username - это свойство, нельзя устанавливать
|
||
setattr(temp_author, key, value)
|
||
# Добавляем отфильтрованную версию
|
||
follows_filtered.append(temp_author.dict())
|
||
|
||
follows = follows_filtered
|
||
else:
|
||
follows = existing_follows
|
||
|
||
logger.debug(f"Актуальный список подписок получен: {len(follows)} элементов")
|
||
|
||
return {f"{entity_type}s": follows, "error": error_result}
|
||
|
||
except Exception as exc:
|
||
logger.exception("Произошла ошибка в функции 'follow'")
|
||
return {"error": str(exc)}
|
||
|
||
|
||
@mutation.field("unfollow")
|
||
@login_required
|
||
async def unfollow(
|
||
_: None, info: GraphQLResolveInfo, what: str, slug: str = "", entity_id: int | None = None
|
||
) -> dict[str, Any]:
|
||
logger.debug("Начало выполнения функции 'unfollow'")
|
||
viewer_id = info.context.get("author", {}).get("id")
|
||
follower_dict = info.context.get("author") or {}
|
||
|
||
# ✅ КРИТИЧНО: Инвалидируем кеш В САМОМ НАЧАЛЕ, если пользователь авторизован
|
||
if viewer_id:
|
||
entity_type = what.lower()
|
||
cache_key_pattern = f"author:follows-{entity_type}s:{viewer_id}"
|
||
await redis.execute("DEL", cache_key_pattern)
|
||
await redis.execute("DEL", f"author:id:{viewer_id}")
|
||
logger.debug(f"Инвалидирован кеш подписок В НАЧАЛЕ операции unfollow: {cache_key_pattern}")
|
||
|
||
if not viewer_id:
|
||
return {"error": "Access denied"}
|
||
|
||
logger.debug(f"follower: {follower_dict}")
|
||
|
||
if not viewer_id or not follower_dict:
|
||
logger.warning("Неавторизованный доступ при попытке отписаться")
|
||
return {"error": "UnauthorizedError"}
|
||
|
||
follower_id = follower_dict.get("id")
|
||
logger.debug(f"follower_id: {follower_id}")
|
||
|
||
entity_classes = {
|
||
"AUTHOR": (Author, AuthorFollower, get_cached_follower_authors, cache_author),
|
||
"TOPIC": (Topic, TopicFollower, get_cached_follower_topics, cache_topic),
|
||
"COMMUNITY": (Community, CommunityFollower, None, None), # Нет методов кэша для сообщества
|
||
"SHOUT": (Shout, ShoutReactionsFollower, None, None), # Нет методов кэша для shout
|
||
}
|
||
|
||
if what not in entity_classes:
|
||
logger.error(f"Неверный тип для отписки: {what}")
|
||
return {"error": "invalid unfollow type"}
|
||
|
||
entity_class, follower_class, get_cached_follows_method, _cache_method = entity_classes[what]
|
||
entity_type = what.lower()
|
||
follows: list[dict[str, Any]] = []
|
||
|
||
try:
|
||
logger.debug("Попытка получить сущность из базы данных")
|
||
with local_session() as session:
|
||
# Используем query для получения сущности
|
||
entity_query = session.query(entity_class)
|
||
if hasattr(entity_class, "slug"):
|
||
entity_query = entity_query.where(entity_class.slug == slug)
|
||
|
||
entity = entity_query.first()
|
||
logger.debug(f"Полученная сущность: {entity}")
|
||
if not entity:
|
||
logger.warning(f"{what.lower()} не найден по slug: {slug}")
|
||
return {"error": f"{what.lower()} not found"}
|
||
|
||
if not entity_id:
|
||
entity_id = getattr(entity, "id", None)
|
||
if not entity_id:
|
||
logger.warning(f"Не удалось получить ID для {what.lower()}")
|
||
return {"error": f"Cannot get ID for {what.lower()}"}
|
||
|
||
logger.debug(f"entity_id: {entity_id}")
|
||
entity_field = get_entity_field_name(entity_type)
|
||
|
||
sub = (
|
||
session.query(follower_class)
|
||
.where(
|
||
and_(
|
||
follower_class.follower == follower_id, # type: ignore[attr-defined]
|
||
getattr(follower_class, entity_field) == entity_id, # type: ignore[attr-defined]
|
||
)
|
||
)
|
||
.first()
|
||
)
|
||
if not sub:
|
||
logger.warning(f"Подписка не найдена для {what.lower()} с ID {entity_id}")
|
||
return {"error": "Not following"}
|
||
|
||
logger.debug(f"Найдена подписка для удаления: {sub}")
|
||
session.delete(sub)
|
||
session.commit()
|
||
logger.info(f"Пользователь {follower_id} отписался от {what.lower()} с ID {entity_id}")
|
||
|
||
# Кеш подписок follower'а уже инвалидирован в начале функции
|
||
if get_cached_follows_method and isinstance(follower_id, int):
|
||
logger.debug("Получение актуального списка подписок из кэша")
|
||
follows = await get_cached_follows_method(follower_id)
|
||
logger.debug(f"Актуальный список подписок получен: {len(follows)} элементов")
|
||
else:
|
||
follows = []
|
||
|
||
if what == "AUTHOR" and isinstance(follower_dict, dict):
|
||
await notify_follower(follower=follower_dict, author_id=entity_id, action="unfollow")
|
||
|
||
# ✅ КРИТИЧНО: Инвалидируем кеш списка подписчиков автора
|
||
# чтобы отписавшийся сразу исчез из списка
|
||
await redis.execute("DEL", f"author:followers:{entity_id}")
|
||
logger.debug(f"Инвалидирован кеш подписчиков автора после unfollow: author:followers:{entity_id}")
|
||
|
||
# Инвалидируем кеш статистики авторов для обновления счетчиков подписчиков
|
||
logger.debug("Инвалидируем кеш статистики авторов после отписки")
|
||
await invalidate_authors_cache(entity_id)
|
||
|
||
return {f"{entity_type}s": follows, "error": None}
|
||
|
||
except Exception as exc:
|
||
logger.exception("Произошла ошибка в функции 'unfollow'")
|
||
return {"error": str(exc)}
|
||
|
||
|
||
@query.field("get_shout_followers")
|
||
def get_shout_followers(
|
||
_: None, _info: GraphQLResolveInfo, slug: str = "", shout_id: int | None = None
|
||
) -> list[dict[str, Any]]:
|
||
"""
|
||
Получает список подписчиков для шаута по slug или ID
|
||
|
||
Args:
|
||
_: GraphQL root
|
||
_info: GraphQL context info
|
||
slug: Slug шаута (опционально)
|
||
shout_id: ID шаута (опционально)
|
||
|
||
Returns:
|
||
Список подписчиков шаута
|
||
"""
|
||
if not slug and not shout_id:
|
||
return []
|
||
|
||
with local_session() as session:
|
||
# Если slug не указан, ищем шаут по ID
|
||
if not slug and shout_id is not None:
|
||
shout = session.query(Shout).where(Shout.id == shout_id).first()
|
||
else:
|
||
# Ищем шаут по slug
|
||
shout = session.query(Shout).where(Shout.slug == slug).first()
|
||
|
||
if not shout:
|
||
return []
|
||
|
||
# Получаем подписчиков шаута
|
||
followers_query = (
|
||
session.query(Author)
|
||
.join(ShoutReactionsFollower, Author.id == ShoutReactionsFollower.follower)
|
||
.where(ShoutReactionsFollower.shout == shout.id)
|
||
)
|
||
|
||
followers = followers_query.all()
|
||
|
||
# Возвращаем безопасную версию данных
|
||
return [follower.dict() for follower in followers]
|