106 lines
3.2 KiB
Python
106 lines
3.2 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.core.errors import AccessDeniedException
|
|
from src.db.models.app_user import AppUser
|
|
from src.db.models.org_unit import OrgUnit
|
|
from src.db.models.role import UserRoleEnum
|
|
from src.repository.org_unit_repository import OrgUnitRepository
|
|
|
|
|
|
|
|
class OrgUnitService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
self.org_repo = OrgUnitRepository(db)
|
|
|
|
async def get_list(
|
|
self,
|
|
user: AppUser,
|
|
offset: int | None = None,
|
|
limit: int | None = None,
|
|
is_ssp: bool | None = None,
|
|
is_active: bool | None = None,
|
|
org_unit_id: int | list[int] | None = None,
|
|
with_count: bool = False,
|
|
load_users: bool = False,
|
|
) -> list[OrgUnit] | tuple[int, list[OrgUnit]]:
|
|
if user.role_id != UserRoleEnum.ADMIN:
|
|
return (0, []) if with_count else []
|
|
|
|
return await self.org_repo.get_list(
|
|
offset=offset,
|
|
limit=limit,
|
|
with_count=with_count,
|
|
is_ssp=is_ssp,
|
|
is_active=is_active,
|
|
load_users=load_users,
|
|
org_unit_id=org_unit_id,
|
|
)
|
|
|
|
async def get(
|
|
self,
|
|
org_unit_id: int,
|
|
user: AppUser,
|
|
load_users: bool = False,
|
|
) -> OrgUnit | None:
|
|
if user.role_id != UserRoleEnum.ADMIN:
|
|
return None
|
|
return await self.org_repo.get(
|
|
org_unit_id=org_unit_id,
|
|
load_users=load_users,
|
|
)
|
|
|
|
async def create(
|
|
self,
|
|
title: str,
|
|
is_ssp: bool,
|
|
utc_offset: str,
|
|
user: AppUser,
|
|
) -> OrgUnit:
|
|
"""Создание нового ССП/РФ."""
|
|
if not self._can_create_ssp(user):
|
|
raise AccessDeniedException()
|
|
|
|
return await self.org_repo.create(
|
|
title=title,
|
|
is_ssp=is_ssp,
|
|
utc_offset=utc_offset,
|
|
)
|
|
|
|
async def update(
|
|
self, org_unit_id: int, user: AppUser, **kwargs
|
|
) -> OrgUnit | None:
|
|
"""Редактирование ССП/РФ"""
|
|
#ToDo: писать логи
|
|
if not self._can_edit_ssp(user):
|
|
raise AccessDeniedException()
|
|
|
|
org_unit = await self.org_repo.get(org_unit_id=org_unit_id)
|
|
|
|
if not org_unit:
|
|
return None
|
|
|
|
return await self.org_repo.update(
|
|
org_unit=org_unit, **kwargs,
|
|
)
|
|
|
|
async def logical_delete(
|
|
self,
|
|
org_unit_id: int,
|
|
user: AppUser,
|
|
is_active: bool | None = None,
|
|
) -> bool:
|
|
"""Логическое удаление ССП/РФ"""
|
|
if not self._can_delete_ssp(user):
|
|
raise AccessDeniedException()
|
|
|
|
return bool(await self.org_repo.logical_delete(org_unit_id=org_unit_id, is_active=is_active))
|
|
|
|
def _can_create_ssp(self, user: AppUser) -> bool:
|
|
return user.role_id == UserRoleEnum.ADMIN
|
|
|
|
def _can_edit_ssp(self, user: AppUser) -> bool:
|
|
return user.role_id == UserRoleEnum.ADMIN
|
|
|
|
def _can_delete_ssp(self, user: AppUser) -> bool:
|
|
return user.role_id == UserRoleEnum.ADMIN |