Files
state-corp-backend/src/apps/user/services.py
Aleksandr Meshchriakov 697ecb7d1c
Some checks failed
CI/CD Pipeline / Code Quality Checks (push) Failing after 3m50s
CI/CD Pipeline / Run Tests (push) Successful in 3m57s
CI/CD Pipeline / Build Docker Images (push) Has been skipped
CI/CD Pipeline / Push to Gitea Registry (push) Has been skipped
CI/CD Pipeline / Deploy to Server (push) Has been skipped
Implement exchange imports and frontend reporting APIs
2026-04-07 16:31:04 +02:00

243 lines
7.3 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from typing import Any
from django.contrib.auth import get_user_model
from django.db import transaction
from rest_framework_simplejwt.tokens import RefreshToken
from apps.core.exceptions import NotFoundError
from .models import Profile
User = get_user_model()
class UserService:
"""Сервисный слой для работы с пользователями"""
@classmethod
def create_user(
cls, *, email: str, username: str, password: str, **extra_fields
) -> User:
"""
Создает нового пользователя
Args:
email: Email пользователя
username: Username пользователя
password: Пароль
**extra_fields: Дополнительные поля
Returns:
User: Созданный пользователь
Raises:
ValidationError: При некорректных данных
"""
with transaction.atomic():
user = User.objects.create_user(
email=email, username=username, password=password, **extra_fields
)
return user
@classmethod
def get_user_by_email(cls, email: str) -> User:
"""Получает пользователя по email
Raises:
NotFoundError: Если пользователь не найден
"""
try:
return User.objects.get(email=email)
except User.DoesNotExist as e:
raise NotFoundError(
message=f"User with email={email} not found",
details={"email": email},
) from e
@classmethod
def get_user_by_email_or_none(cls, email: str) -> User | None:
"""Получает пользователя по email или None"""
try:
return User.objects.get(email=email)
except User.DoesNotExist:
return None
@classmethod
def get_user_by_id(cls, user_id: int) -> User:
"""Получает пользователя по ID
Raises:
NotFoundError: Если пользователь не найден
"""
try:
return User.objects.get(id=user_id)
except User.DoesNotExist as e:
raise NotFoundError(
message=f"User with id={user_id} not found",
details={"user_id": user_id},
) from e
@classmethod
def get_user_by_id_or_none(cls, user_id: int) -> User | None:
"""Получает пользователя по ID или None"""
try:
return User.objects.get(id=user_id)
except User.DoesNotExist:
return None
@classmethod
def update_user(cls, user_id: int, **fields) -> User:
"""
Обновляет данные пользователя
Args:
user_id: ID пользователя
**fields: Поля для обновления
Returns:
User: Обновленный пользователь
Raises:
NotFoundError: Если пользователь не найден
"""
user = cls.get_user_by_id(user_id)
for field, value in fields.items():
setattr(user, field, value)
user.save()
return user
@classmethod
def delete_user(cls, user_id: int) -> None:
"""
Удаляет пользователя
Args:
user_id: ID пользователя
Raises:
NotFoundError: Если пользователь не найден
"""
user = cls.get_user_by_id(user_id)
user.delete()
@classmethod
def get_tokens_for_user(cls, user: User) -> dict[str, str]:
"""
Генерирует JWT токены для пользователя
Args:
user: Пользователь
Returns:
Dict[str, str]: refresh и access токены
"""
refresh = RefreshToken.for_user(user)
return {
"refresh": str(refresh),
"access": str(refresh.access_token),
}
@classmethod
def verify_email(cls, user_id: int) -> User:
"""
Подтверждает email пользователя
Args:
user_id: ID пользователя
Returns:
User: Обновленный пользователь
Raises:
NotFoundError: Если пользователь не найден
"""
user = cls.get_user_by_id(user_id)
user.is_verified = True
user.save()
return user
class ProfileService:
"""Сервисный слой для работы с профилями"""
@classmethod
def get_profile_by_user_id(cls, user_id: int) -> Profile:
"""Получает профиль по ID пользователя
Raises:
NotFoundError: Если профиль не найден
"""
try:
return Profile.objects.select_related("user").get(user_id=user_id)
except Profile.DoesNotExist as e:
raise NotFoundError(
message=f"Profile for user_id={user_id} not found",
details={"user_id": user_id},
) from e
@classmethod
def get_profile_by_user_id_or_none(cls, user_id: int) -> Profile | None:
"""Получает профиль по ID пользователя или None"""
try:
return Profile.objects.select_related("user").get(user_id=user_id)
except Profile.DoesNotExist:
return None
@classmethod
def update_profile(cls, user_id: int, **fields) -> Profile:
"""
Обновляет профиль пользователя
Args:
user_id: ID пользователя
**fields: Поля для обновления
Returns:
Profile: Обновленный профиль
Raises:
NotFoundError: Если профиль не найден
"""
profile = cls.get_profile_by_user_id(user_id)
for field, value in fields.items():
setattr(profile, field, value)
profile.save()
return profile
@classmethod
def get_full_profile_data(cls, user_id: int) -> dict[str, Any]:
"""
Получает полные данные пользователя и профиля
Args:
user_id: ID пользователя
Returns:
Dict: Полные данные
Raises:
NotFoundError: Если профиль не найден
"""
profile = cls.get_profile_by_user_id(user_id)
user = profile.user
return {
"id": user.id,
"email": user.email,
"username": user.username,
"is_verified": user.is_verified,
"phone": user.phone,
"first_name": profile.first_name,
"mid_name": profile.mid_name,
"last_name": profile.last_name,
"full_name": profile.full_name,
"bio": profile.bio,
"avatar": profile.avatar.url if profile.avatar else None,
"date_of_birth": profile.date_of_birth,
"created_at": user.created_at,
"updated_at": user.updated_at,
}