first commit
Some checks failed
CI/CD Pipeline / Run Tests (push) Failing after 0s
CI/CD Pipeline / Code Quality Checks (push) Failing after 1m43s
CI/CD Pipeline / Build Docker Images (push) Has been skipped
CI/CD Pipeline / Push to Gitea Registry (push) Has been skipped

This commit is contained in:
2026-01-21 12:07:35 +01:00
commit e9d7f24aaa
102 changed files with 13890 additions and 0 deletions

240
src/apps/user/services.py Normal file
View File

@@ -0,0 +1,240 @@
from typing import Any
from apps.core.exceptions import NotFoundError
from django.contrib.auth import get_user_model
from django.db import transaction
from rest_framework_simplejwt.tokens import RefreshToken
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,
"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,
}