core/resolvers/feed.py

from typing import Any

from graphql import GraphQLResolveInfo
from sqlalchemy import Select, and_, select

from orm.author import Author, AuthorFollower
from orm.shout import Shout, ShoutAuthor, ShoutReactionsFollower, ShoutTopic
from orm.topic import Topic, TopicFollower
from resolvers.reader import (
    apply_options,
    get_shouts_with_links,
    has_field,
    query_with_stat,
)
from services.auth import login_required
from storage.db import local_session
from storage.schema import query
from utils.logger import root_logger as logger


@query.field("load_shouts_coauthored")
@login_required
async def load_shouts_coauthored(_: None, info: GraphQLResolveInfo, options: dict) -> list[Shout]:
    """
    Загрузка публикаций, написанных в соавторстве с пользователем.

    :param info: Информаци о контексте GraphQL.
    :param options: Опции фильтрации и сортировки.
    :return: Список публикаций в соавтостве.
    """
    author_id = info.context.get("author", {}).get("id")
    if not author_id:
        return []
    q = query_with_stat(info)
    q = q.where(Shout.authors.any(id=author_id))
    q, limit, offset = apply_options(q, options)
    return get_shouts_with_links(info, q, limit, offset=offset)


@query.field("load_shouts_discussed")
@login_required
async def load_shouts_discussed(_: None, info: GraphQLResolveInfo, options: dict) -> list[Shout]:
    """
    Загрузка публикаций, которые обсуждались пользователем.

    :param info: Информация о контексте GraphQL.
    :param options: Опции фильтрации и сортировки.
    :return: Список публикаций, обсужденых пользователем.
    """
    author_id = info.context.get("author", {}).get("id")
    if not author_id:
        return []
    q = query_with_stat(info)
    options["filters"]["commented"] = True
    q, limit, offset = apply_options(q, options, author_id)
    return get_shouts_with_links(info, q, limit, offset=offset)


def shouts_by_follower(info: GraphQLResolveInfo, follower_id: int, options: dict[str, Any]) -> list[Shout]:
    """
    Загружает публикации, на которые подписан автор.

    - по авторам
    - по темам
    - по реакциям

    :param info: Информация о контексте GraphQL.
    :param follower_id: Идентификатор автора.
    :param options: Опции фильтрации и сортировки.
    :return: Список публикаций.
    """
    q = query_with_stat(info)
    reader_followed_authors: Select = select(AuthorFollower.following).where(AuthorFollower.follower == follower_id)
    reader_followed_topics: Select = select(TopicFollower.topic).where(TopicFollower.follower == follower_id)
    reader_followed_shouts: Select = select(ShoutReactionsFollower.shout).where(
        ShoutReactionsFollower.follower == follower_id
    )
    followed_subquery = (
        select(Shout.id)
        .join(ShoutAuthor, ShoutAuthor.shout == Shout.id)
        .join(ShoutTopic, ShoutTopic.shout == Shout.id)
        .where(
            ShoutAuthor.author.in_(reader_followed_authors)
            | ShoutTopic.topic.in_(reader_followed_topics)
            | Shout.id.in_(reader_followed_shouts)
        )
        .scalar_subquery()
    )
    q = q.where(Shout.id.in_(followed_subquery))
    q, limit, offset = apply_options(q, options)
    return get_shouts_with_links(info, q, limit, offset=offset)


@query.field("load_shouts_followed_by")
async def load_shouts_followed_by(_: None, info: GraphQLResolveInfo, slug: str, options: dict) -> list[Shout]:
    """
    Загружает публикации, на которые подписан автор по slug.

    :param info: Информация о контексте GraphQL.
    :param slug: Slug автора.
    :param options: Опции фильтрации и сортировки.
    :return: Список публикаций.
    """
    with local_session() as session:
        author = session.query(Author).where(Author.slug == slug).first()
        if author:
            follower_id = author.dict()["id"]
            return shouts_by_follower(info, follower_id, options)
    return []


@query.field("load_shouts_feed")
@login_required
async def load_shouts_feed(_: None, info: GraphQLResolveInfo, options: dict) -> list[Shout]:
    """
    Загружает публикации, на которые подписан авторизованный пользователь.

    :param info: Информация о контексте GraphQL.
    :param options: Опции фильтрации и сортировки.
    :return: Список публикаций.
    """
    author_id = info.context.get("author", {}).get("id")
    return shouts_by_follower(info, author_id, options) if author_id else []


@query.field("load_shouts_authored_by")
async def load_shouts_authored_by(_: None, info: GraphQLResolveInfo, slug: str, options: dict[str, Any]) -> list[Shout]:
    """
    Загружает публикации, написанные автором по slug.

    :param info: Информация о контексте GraphQL.
    :param slug: Slug автора.
    :param options: Опции фильтрации и сортировки.
    :return: Список публикаций.
    """
    with local_session() as session:
        author = session.query(Author).where(Author.slug == slug).first()
        if author:
            try:
                author_id: int = author.dict()["id"]
                q: Select = (
                    query_with_stat(info)
                    if has_field(info, "stat")
                    else select(Shout).where(and_(Shout.published_at.is_not(None), Shout.deleted_at.is_(None)))
                )
                q = q.where(Shout.authors.any(id=author_id))
                q, limit, offset = apply_options(q, options, author_id)
                return get_shouts_with_links(info, q, limit, offset=offset)
            except Exception as error:
                logger.debug(error)
    return []


@query.field("load_shouts_with_topic")
async def load_shouts_with_topic(_: None, info: GraphQLResolveInfo, slug: str, options: dict[str, Any]) -> list[Shout]:
    """
    Загружает публикации, связанные с темой по slug.

    :param info: Информация о контексте GraphQL.
    :param slug: Slug темы.
    :param options: Опции фильтрации и сортировки.
    :return: Список публикаций.
    """
    with local_session() as session:
        topic = session.query(Topic).where(Topic.slug == slug).first()
        if topic:
            try:
                topic_id: int = topic.dict()["id"]
                q: Select = (
                    query_with_stat(info)
                    if has_field(info, "stat")
                    else select(Shout).where(and_(Shout.published_at.is_not(None), Shout.deleted_at.is_(None)))
                )
                q = q.where(Shout.topics.any(id=topic_id))
                q, limit, offset = apply_options(q, options)
                return get_shouts_with_links(info, q, limit, offset=offset)
            except Exception as error:
                logger.debug(error)
    return []