169 lines
6.8 KiB
Python
169 lines
6.8 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.db.models.app_user import AppUser
|
||
from src.db.models.role import UserRoleEnum
|
||
from src.core.errors import (
|
||
AccessDeniedException,
|
||
UserNotFoundException,
|
||
ValidationException,
|
||
UsernameConflictException,
|
||
)
|
||
from src.domain.schemas import UserCreate, UserUpdate
|
||
from src.repository.user_repository import UserRepository
|
||
|
||
|
||
class UserService:
|
||
def __init__(self, db: AsyncSession):
|
||
self.db = db
|
||
self.user_repo = UserRepository(db)
|
||
|
||
async def get_all(self, current_user: AppUser, skip: int = 0, limit: int = 100):
|
||
if current_user.role_id != UserRoleEnum.ADMIN:
|
||
raise AccessDeniedException()
|
||
return await self.user_repo.get_list(skip=skip, limit=limit)
|
||
|
||
async def get(self, user_id: int, current_user: AppUser):
|
||
if current_user.role_id != UserRoleEnum.ADMIN:
|
||
raise AccessDeniedException()
|
||
return await self.user_repo.get(user_id)
|
||
|
||
async def create_user(self, user_data: UserCreate, creator: AppUser):
|
||
if creator.role_id != UserRoleEnum.ADMIN:
|
||
raise AccessDeniedException()
|
||
|
||
role_id = user_data.role_id or UserRoleEnum.EXECUTOR_RF.value
|
||
try:
|
||
role = UserRoleEnum(role_id)
|
||
except ValueError:
|
||
raise ValidationException("Роль не найдена", field="role_id")
|
||
user_data.role_id = role.value
|
||
|
||
if await self.user_repo.get_by_username(user_data.username):
|
||
raise UsernameConflictException()
|
||
|
||
payload = user_data.model_dump(mode="json")
|
||
return await self.user_repo.create(payload)
|
||
|
||
async def update_user(self, user_id: int, user_data: UserUpdate, user: AppUser):
|
||
if user.role_id != UserRoleEnum.ADMIN:
|
||
raise AccessDeniedException()
|
||
|
||
target = await self.user_repo.get(user_id)
|
||
if not target:
|
||
raise UserNotFoundException()
|
||
|
||
update_payload = user_data.model_dump(mode="json", exclude_unset=True)
|
||
previous_role_id = self._normalize_role_id(target.role_id)
|
||
role_id = update_payload.get("role_id")
|
||
role = None
|
||
if role_id is not None:
|
||
try:
|
||
role = UserRoleEnum(role_id)
|
||
except ValueError:
|
||
raise ValidationException("Роль не найдена", field="role_id")
|
||
|
||
if role_id != self._normalize_role_id(user.role_id) and user.id == user_id:
|
||
raise ValidationException(
|
||
"Нельзя изменить роль самому себе", field="role_id"
|
||
)
|
||
update_payload["role_id"] = role.value
|
||
|
||
if "username" in update_payload and update_payload["username"] != target.username:
|
||
conflict = await self.user_repo.get_by_username(update_payload["username"])
|
||
if conflict:
|
||
raise UsernameConflictException()
|
||
|
||
updated = await self.user_repo.update(user_id, update_payload)
|
||
|
||
if role_id is not None and role is not None:
|
||
had_many_ssp_role = previous_role_id in (
|
||
UserRoleEnum.ADMIN.value,
|
||
UserRoleEnum.EXECUTOR_DFIP.value,
|
||
)
|
||
has_many_ssp_role_now = role.value in (
|
||
UserRoleEnum.ADMIN.value,
|
||
UserRoleEnum.EXECUTOR_DFIP.value,
|
||
)
|
||
is_executor_to_dfip_transition = (
|
||
previous_role_id == UserRoleEnum.EXECUTOR_RF.value
|
||
and role.value == UserRoleEnum.EXECUTOR_DFIP.value
|
||
)
|
||
if (had_many_ssp_role and not has_many_ssp_role_now) or is_executor_to_dfip_transition:
|
||
await self.user_repo.clear_many_ssp(user_id)
|
||
|
||
return updated
|
||
|
||
async def delete_user(self, user_id: int, user: AppUser):
|
||
if user.role_id != UserRoleEnum.ADMIN:
|
||
raise AccessDeniedException()
|
||
deleted = await self.user_repo.logical_delete(user_id)
|
||
if not deleted:
|
||
raise UserNotFoundException()
|
||
|
||
async def get_roles(self, current_user: AppUser):
|
||
if current_user.role_id != UserRoleEnum.ADMIN:
|
||
raise AccessDeniedException()
|
||
return await self.user_repo.get_roles()
|
||
|
||
async def get_many_ssp(self, user_id: int, current_user: AppUser):
|
||
if current_user.role_id == UserRoleEnum.ADMIN:
|
||
return await self.user_repo.get_many_ssp(user_id)
|
||
if current_user.id != user_id:
|
||
raise AccessDeniedException()
|
||
return await self.user_repo.get_many_ssp(user_id)
|
||
|
||
async def set_many_ssp(self, user_id: int, ssp_ids: list[int], current_user: AppUser):
|
||
if current_user.role_id != UserRoleEnum.ADMIN:
|
||
raise AccessDeniedException()
|
||
|
||
user = await self.user_repo.get(user_id)
|
||
if not user:
|
||
raise UserNotFoundException()
|
||
|
||
target_role_id = self._normalize_role_id(user.role_id)
|
||
if target_role_id not in (
|
||
UserRoleEnum.ADMIN.value,
|
||
UserRoleEnum.EXECUTOR_DFIP.value,
|
||
UserRoleEnum.EXECUTOR_RF.value,
|
||
):
|
||
err = AccessDeniedException()
|
||
err.description = (
|
||
"Несколько ССП/РФ может быть только у пользовтелей с ролями EXECUTOR и ADMIN"
|
||
)
|
||
raise err
|
||
|
||
curr_ssp_ids = await self.user_repo.get_many_ssp_ids(user_id)
|
||
if target_role_id == UserRoleEnum.EXECUTOR_RF.value:
|
||
merged_ssp_ids = set(curr_ssp_ids) | set(ssp_ids)
|
||
if len(merged_ssp_ids) > 1:
|
||
raise ValidationException(
|
||
"Для пользователя с ролью EXECUTOR_RF можно назначить только один ССП/РФ"
|
||
)
|
||
new_ssp_ids = [ssp_id for ssp_id in ssp_ids if ssp_id not in curr_ssp_ids]
|
||
return await self.user_repo.set_many_ssp(user.id, new_ssp_ids)
|
||
|
||
async def unset_many_ssp(self, user_id: int, ssp_ids: list[int], current_user: AppUser):
|
||
if current_user.role_id != UserRoleEnum.ADMIN:
|
||
raise AccessDeniedException()
|
||
|
||
user = await self.user_repo.get(user_id)
|
||
if not user:
|
||
raise UserNotFoundException()
|
||
|
||
target_role_id = self._normalize_role_id(user.role_id)
|
||
if target_role_id not in (
|
||
UserRoleEnum.ADMIN.value,
|
||
UserRoleEnum.EXECUTOR_DFIP.value,
|
||
UserRoleEnum.EXECUTOR_RF.value,
|
||
):
|
||
err = AccessDeniedException()
|
||
err.description = (
|
||
"Несколько ССП/РФ может быть только у пользовтелей с ролями EXECUTOR и ADMIN"
|
||
)
|
||
raise err
|
||
return await self.user_repo.unset_many_ssp(user_id, ssp_ids)
|
||
|
||
@staticmethod
|
||
def _normalize_role_id(role_id: int | UserRoleEnum) -> int:
|
||
return role_id.value if isinstance(role_id, UserRoleEnum) else int(role_id)
|