feat: Add comprehensive Django user app with tests using model-bakery
- Implemented user authentication with JWT tokens - Added user and profile models with OneToOne relationship - Created service layer for business logic separation - Implemented DRF serializers and views - Added comprehensive test suite with model-bakery factories - Fixed ipdb/pdbpp dependency conflicts with custom test runner - Configured development and production environments - Added deployment configurations for Apache, systemd, and Docker
This commit is contained in:
194
src/apps/user/services.py
Normal file
194
src/apps/user/services.py
Normal file
@@ -0,0 +1,194 @@
|
||||
from typing import Dict, Any, Optional
|
||||
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) -> Optional[User]:
|
||||
"""Получает пользователя по email"""
|
||||
try:
|
||||
return User.objects.get(email=email)
|
||||
except User.DoesNotExist:
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def get_user_by_id(cls, user_id: int) -> Optional[User]:
|
||||
"""Получает пользователя по ID"""
|
||||
try:
|
||||
return User.objects.get(id=user_id)
|
||||
except User.DoesNotExist:
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def update_user(cls, user_id: int, **fields) -> Optional[User]:
|
||||
"""
|
||||
Обновляет данные пользователя
|
||||
|
||||
Args:
|
||||
user_id: ID пользователя
|
||||
**fields: Поля для обновления
|
||||
|
||||
Returns:
|
||||
User: Обновленный пользователь или None
|
||||
"""
|
||||
user = cls.get_user_by_id(user_id)
|
||||
if not user:
|
||||
return None
|
||||
|
||||
for field, value in fields.items():
|
||||
setattr(user, field, value)
|
||||
|
||||
user.save()
|
||||
return user
|
||||
|
||||
@classmethod
|
||||
def delete_user(cls, user_id: int) -> bool:
|
||||
"""
|
||||
Удаляет пользователя
|
||||
|
||||
Args:
|
||||
user_id: ID пользователя
|
||||
|
||||
Returns:
|
||||
bool: True если успешно удален
|
||||
"""
|
||||
user = cls.get_user_by_id(user_id)
|
||||
if user:
|
||||
user.delete()
|
||||
return True
|
||||
return False
|
||||
|
||||
@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) -> bool:
|
||||
"""
|
||||
Подтверждает email пользователя
|
||||
|
||||
Args:
|
||||
user_id: ID пользователя
|
||||
|
||||
Returns:
|
||||
bool: True если успешно подтвержден
|
||||
"""
|
||||
user = cls.get_user_by_id(user_id)
|
||||
if user:
|
||||
user.is_verified = True
|
||||
user.save()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class ProfileService:
|
||||
"""Сервисный слой для работы с профилями"""
|
||||
|
||||
@classmethod
|
||||
def get_profile_by_user_id(cls, user_id: int) -> Optional[Profile]:
|
||||
"""Получает профиль по ID пользователя"""
|
||||
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) -> Optional[Profile]:
|
||||
"""
|
||||
Обновляет профиль пользователя
|
||||
|
||||
Args:
|
||||
user_id: ID пользователя
|
||||
**fields: Поля для обновления
|
||||
|
||||
Returns:
|
||||
Profile: Обновленный профиль или None
|
||||
"""
|
||||
profile = cls.get_profile_by_user_id(user_id)
|
||||
if not profile:
|
||||
return None
|
||||
|
||||
for field, value in fields.items():
|
||||
setattr(profile, field, value)
|
||||
|
||||
profile.save()
|
||||
return profile
|
||||
|
||||
@classmethod
|
||||
def get_full_profile_data(cls, user_id: int) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Получает полные данные пользователя и профиля
|
||||
|
||||
Args:
|
||||
user_id: ID пользователя
|
||||
|
||||
Returns:
|
||||
Dict: Полные данные или None
|
||||
"""
|
||||
profile = cls.get_profile_by_user_id(user_id)
|
||||
if not profile:
|
||||
return None
|
||||
|
||||
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,
|
||||
}
|
||||
Reference in New Issue
Block a user