71 lines
2.1 KiB
Python
71 lines
2.1 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.repository.summary_repository import SummaryRepository
|
|
from src.repository.user_repository import UserRepository
|
|
|
|
|
|
class SummaryService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
self.summary_repo = SummaryRepository(db)
|
|
self.user_repo = UserRepository(db)
|
|
|
|
async def get_svod1_rows(
|
|
self,
|
|
year: int,
|
|
sheet: str,
|
|
user: AppUser,
|
|
org_ids: list[int] | None = None,
|
|
) -> list[tuple]:
|
|
org_ids = await self._allowed_org_ids(user, org_ids)
|
|
if org_ids == []:
|
|
return []
|
|
return await self.summary_repo.get_svod1_rows(
|
|
year=year,
|
|
sheet=sheet,
|
|
org_ids=org_ids,
|
|
)
|
|
|
|
async def get_svod2_rows(
|
|
self,
|
|
year: int,
|
|
sheet: str,
|
|
user: AppUser,
|
|
org_ids: list[int] | None = None,
|
|
offset: int | None = None,
|
|
limit: int | None = None,
|
|
sort_by: str = "org_name",
|
|
sort_direction: str = "asc",
|
|
) -> list[tuple]:
|
|
org_ids = await self._allowed_org_ids(user, org_ids)
|
|
if org_ids == []:
|
|
return []
|
|
return await self.summary_repo.get_svod2_rows(
|
|
year=year,
|
|
sheet=sheet,
|
|
org_ids=org_ids,
|
|
offset=offset,
|
|
limit=limit,
|
|
sort_by=sort_by,
|
|
sort_direction=sort_direction,
|
|
)
|
|
|
|
async def _allowed_org_ids(
|
|
self,
|
|
user: AppUser,
|
|
requested_org_ids: list[int] | None,
|
|
) -> list[int] | None:
|
|
if user.role_id == UserRoleEnum.ADMIN:
|
|
return requested_org_ids
|
|
|
|
db_user = await self.user_repo.get(user_id=user.id, load_orgs=True)
|
|
if not db_user:
|
|
return []
|
|
|
|
allowed_org_ids = {org.id for org in db_user.org_units}
|
|
if requested_org_ids is None:
|
|
return list(allowed_org_ids)
|
|
return [org_id for org_id in requested_org_ids if org_id in allowed_org_ids]
|