/
outthevoid
/
python_microblogging_service_project
Обзор
Документация
Войти
/
outthevoid
/
python_microblogging_service_project
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
app/services/user_service.py
177 строк
8 KB
VladimirChukavin
Revert "Вернул проект в исходное состояние."
07 апр 2026, 14:01
07 апр 2026, 14:01
7531b60
Код
Авторство
О чём код?
from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import ConflictError, NotFoundError, ValidationError from app.models.user_model import User from app.repositories.user_repo import UserRepository from app.schemas.user_schema import ( UserDataResponseSchema, UserFullResponseSchema, UserShortResponseSchema, ) class UserService: """ Сервис для работы с пользователями и подписками. Инкапсулирует бизнес-логику получения профилей пользователей, подготовки данных ответа и управления подписками между пользователями. :ivar session: Асинхронная сессия базы данных. :vartype session: sqlalchemy.ext.asyncio.AsyncSession :ivar repo: Репозиторий для операций с пользователями и подписками. :vartype repo: app.repositories.user_repo.UserRepository """ def __init__(self, session: AsyncSession): """ Инициализирует сервис пользователей. :param session: Асинхронная сессия базы данных. :type session: sqlalchemy.ext.asyncio.AsyncSession """ self.session = session self.repo = UserRepository(session) @staticmethod def _user_response(user: User) -> UserDataResponseSchema: """ Преобразует ORM-объект пользователя в схему данных ответа. Формирует структуру с основной информацией о пользователе, а также списками его подписчиков и подписок. :param user: Объект пользователя. :type user: app.models.user_model.User :returns: Схема с детализированными данными пользователя. :rtype: app.schemas.user_schema.UserDataResponseSchema """ return UserDataResponseSchema( id=user.id, name=user.name, followers=[ UserShortResponseSchema(id=follower.id, name=follower.name) for follower in user.followers ], following=[ UserShortResponseSchema(id=following_user.id, name=following_user.name) for following_user in user.following ], ) async def get_current_user(self, user: User) -> UserFullResponseSchema: """ Возвращает полный ответ с данными текущего пользователя. :param user: Текущий аутентифицированный пользователь. :type user: app.models.user_model.User :returns: Полный ответ с данными текущего пользователя. :rtype: app.schemas.user_schema.UserFullResponseSchema """ return UserFullResponseSchema( result=True, user=self._user_response(user), ) async def get_user_by_id(self, user_id: int) -> UserFullResponseSchema: """ Возвращает полный ответ с данными пользователя по его идентификатору. Выполняет поиск пользователя в репозитории и выбрасывает исключение, если пользователь не найден. :param user_id: Идентификатор пользователя. :type user_id: int :raises app.core.exceptions.NotFoundError: Если пользователь не найден. :returns: Полный ответ с данными пользователя. :rtype: app.schemas.user_schema.UserFullResponseSchema """ user = await self.repo.get_by_id(user_id) if not user: raise NotFoundError("Пользователь не найден") return UserFullResponseSchema( result=True, user=self._user_response(user), ) async def subscribe_for_user(self, current_user: User, target_user_id: int) -> None: """ Оформляет подписку текущего пользователя на другого пользователя. Проверяет, что пользователь не подписывается сам на себя, целевой пользователь существует, а подписка ещё не создана. :param current_user: Текущий аутентифицированный пользователь. :type current_user: app.models.user_model.User :param target_user_id: Идентификатор пользователя, на которого оформляется подписка. :type target_user_id: int :raises app.core.exceptions.ValidationError: Если пользователь пытается подписаться сам на себя. :raises app.core.exceptions.NotFoundError: Если целевой пользователь не найден. :raises app.core.exceptions.ConflictError: Если подписка уже существует. :returns: ``None`` :rtype: None """ if current_user.id == target_user_id: raise ValidationError("Вы не можете подписаться сам на себя") target_user = await self.repo.get_by_id(target_user_id) if not target_user: raise NotFoundError("Пользователь не найден") check_following = await self.repo.subscribe_exists( current_user.id, target_user.id ) if check_following: raise ConflictError("Вы уже подписаны на этого пользователя") await self.repo.subscribe(current_user.id, target_user.id) await self.session.commit() async def unsubscribe_for_user( self, current_user: User, target_user_id: int ) -> None: """ Отменяет подписку текущего пользователя на другого пользователя. Проверяет, что пользователь не пытается отписаться от самого себя, целевой пользователь существует, а подписка действительно есть. :param current_user: Текущий аутентифицированный пользователь. :type current_user: app.models.user_model.User :param target_user_id: Идентификатор пользователя, от которого выполняется отписка. :type target_user_id: int :raises app.core.exceptions.ValidationError: Если пользователь пытается отписаться от самого себя. :raises app.core.exceptions.NotFoundError: Если целевой пользователь не найден или подписка отсутствует. :returns: ``None`` :rtype: None """ if current_user.id == target_user_id: raise ValidationError("Вы не можете отписаться от самого себя") target_user = await self.repo.get_by_id(target_user_id) if not target_user: raise NotFoundError("Пользователь не найден") subscription = await self.repo.unsubscribe(current_user.id, target_user.id) if not subscription: raise NotFoundError("Подписка не найдена") await self.session.commit()