170 lines
6.1 KiB
Python
170 lines
6.1 KiB
Python
import datetime
|
|
from typing import Optional
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.application.roles_repository import RolesRepository
|
|
from app.application.users_repository import UsersRepository
|
|
from app.domain.entities.register_entity import RegisterEntity
|
|
from app.domain.entities.users_entity import UserEntity
|
|
from app.domain.models.users import User, UserGenderEnum
|
|
|
|
|
|
class UsersService:
|
|
def __init__(self, db: Session):
|
|
self.users_repository = UsersRepository(db)
|
|
self.roles_repository = RolesRepository(db)
|
|
|
|
def get_all(self) -> list[UserEntity]:
|
|
users = self.users_repository.get_all()
|
|
return [
|
|
UserEntity(
|
|
id=user.id,
|
|
first_name=user.first_name,
|
|
last_name=user.last_name,
|
|
patronymic=user.patronymic,
|
|
gender=user.gender,
|
|
birthday=user.birthday,
|
|
registration_date=user.registration_date,
|
|
login=user.login,
|
|
email=user.email,
|
|
role_id=user.role_id,
|
|
) for user in users
|
|
]
|
|
|
|
def get_by_id(self, user_id: int) -> Optional[UserEntity]:
|
|
user = self.users_repository.get_by_id(user_id)
|
|
if user:
|
|
return UserEntity(
|
|
id=user.id,
|
|
first_name=user.first_name,
|
|
last_name=user.last_name,
|
|
patronymic=user.patronymic,
|
|
gender=user.gender,
|
|
birthday=user.birthday,
|
|
registration_date=user.registration_date,
|
|
login=user.login,
|
|
email=user.email,
|
|
role_id=user.role_id,
|
|
)
|
|
|
|
return None
|
|
|
|
def get_by_login(self, login: str) -> Optional[UserEntity]:
|
|
user = self.users_repository.get_by_login(login)
|
|
if user:
|
|
return UserEntity(
|
|
id=user.id,
|
|
first_name=user.first_name,
|
|
last_name=user.last_name,
|
|
patronymic=user.patronymic,
|
|
gender=user.gender,
|
|
birthday=user.birthday,
|
|
registration_date=user.registration_date,
|
|
login=user.login,
|
|
email=user.email,
|
|
role_id=user.role_id,
|
|
)
|
|
|
|
return None
|
|
|
|
def create(self, user: RegisterEntity) -> UserEntity:
|
|
|
|
if not user.gender in ['мужской', 'женский', 'не указан']:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Gender must be 'мужской', 'женский' or 'не указан'"
|
|
)
|
|
|
|
gender = UserGenderEnum(user.gender)
|
|
if user.birthday > datetime.date.today():
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Birthday must be in the past"
|
|
)
|
|
|
|
role = self.roles_repository.get_by_id(user.role_id)
|
|
if not role:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Role not found"
|
|
)
|
|
|
|
user_model = User(
|
|
first_name=user.first_name,
|
|
last_name=user.last_name,
|
|
patronymic=user.patronymic,
|
|
gender=gender,
|
|
birthday=user.birthday,
|
|
registration_date=datetime.datetime.now(datetime.UTC),
|
|
login=user.login,
|
|
email=user.email,
|
|
role_id=user.role_id,
|
|
)
|
|
user_model.set_password(user.password)
|
|
created_user = self.users_repository.create(user_model)
|
|
return UserEntity(
|
|
id=created_user.id,
|
|
first_name=created_user.first_name,
|
|
last_name=created_user.last_name,
|
|
patronymic=created_user.patronymic,
|
|
gender=created_user.gender,
|
|
birthday=created_user.birthday,
|
|
registration_date=created_user.registration_date,
|
|
login=created_user.login,
|
|
email=created_user.email,
|
|
role_id=created_user.role_id,
|
|
)
|
|
|
|
def update(self, user: UserEntity) -> Optional[UserEntity]:
|
|
user_model = self.users_repository.get_by_id(user.id)
|
|
if user_model:
|
|
user_model.first_name = user.first_name
|
|
user_model.last_name = user.last_name
|
|
user_model.patronymic = user.patronymic
|
|
user_model.gender = user.gender
|
|
user_model.birthday = user.birthday
|
|
user_model.registration_date = user.registration_date
|
|
user_model.login = user.login
|
|
user_model.email = user.email
|
|
user_model.role_id = user.role_id
|
|
updated_user = self.users_repository.update(user_model)
|
|
return UserEntity(
|
|
id=updated_user.id,
|
|
first_name=updated_user.first_name,
|
|
last_name=updated_user.last_name,
|
|
patronymic=updated_user.patronymic,
|
|
gender=updated_user.gender,
|
|
birthday=updated_user.birthday,
|
|
registration_date=updated_user.registration_date,
|
|
login=updated_user.login,
|
|
email=updated_user.email,
|
|
role_id=updated_user.role_id,
|
|
)
|
|
|
|
return None
|
|
|
|
def update_password(self, user_id: int, old_password: str, new_password: str) -> Optional[UserEntity]:
|
|
user = self.users_repository.get_by_id(user_id)
|
|
if user and user.check_password(old_password):
|
|
user.set_password(new_password)
|
|
updated_user = self.users_repository.update(user)
|
|
return UserEntity(
|
|
id=updated_user.id,
|
|
first_name=updated_user.first_name,
|
|
last_name=updated_user.last_name,
|
|
patronymic=updated_user.patronymic,
|
|
gender=updated_user.gender,
|
|
birthday=updated_user.birthday,
|
|
registration_date=updated_user.registration_date,
|
|
login=updated_user.login,
|
|
email=updated_user.email,
|
|
role_id=updated_user.role_id,
|
|
)
|
|
|
|
return None
|
|
|
|
def delete(self, user_id: int) -> bool:
|
|
return self.users_repository.delete(user_id) is not None
|