52 lines
2.2 KiB
Python
52 lines
2.2 KiB
Python
"""Чтение правил элиминации для версии консолидации."""
|
||
|
||
import uuid
|
||
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.common.exceptions import NotFoundError
|
||
from src.domain.policy import VisibilityScope
|
||
from src.repositories import consolidations as consolidations_repo
|
||
from src.repositories import elimination_rules as rules_repo
|
||
from src.schemas.consolidations import EliminationRuleOut, IntercompanyMatrixRuleSideOut
|
||
|
||
VERSION_NOT_FOUND = "версия консолидации не найдена"
|
||
|
||
|
||
class EliminationRuleService:
|
||
"""Правила выбранной версии из БД с учетом видимости консолидации."""
|
||
|
||
def __init__(self, session: AsyncSession, scope: VisibilityScope) -> None:
|
||
self._session = session
|
||
self._scope = scope
|
||
|
||
async def list_rules(self, consolidation_version_id: uuid.UUID) -> list[EliminationRuleOut]:
|
||
found = await consolidations_repo.get_version(self._session, self._scope, consolidation_version_id)
|
||
if found is None:
|
||
raise NotFoundError(VERSION_NOT_FOUND)
|
||
|
||
rows = await rules_repo.list_rules_with_accounts(self._session, consolidation_version_id)
|
||
by_id: dict[uuid.UUID, EliminationRuleOut] = {}
|
||
for rule, account in rows:
|
||
result = by_id.get(rule.id)
|
||
if result is None:
|
||
result = EliminationRuleOut.model_validate(
|
||
{
|
||
"rule_no": rule.rule_no,
|
||
"version_no": rule.version_no,
|
||
"statement_type": rule.statement_type,
|
||
"side_1": rule.side_1,
|
||
"measure": rule.measure,
|
||
"company_1": IntercompanyMatrixRuleSideOut(accounts=[], excluded=[]),
|
||
"company_2": IntercompanyMatrixRuleSideOut(accounts=[], excluded=[]),
|
||
}
|
||
)
|
||
by_id[rule.id] = result
|
||
|
||
if account is None:
|
||
continue
|
||
side = result.company_1 if account.company_no == 1 else result.company_2
|
||
(side.excluded if account.is_excluded else side.accounts).append(account.account_id)
|
||
|
||
return list(by_id.values())
|