Sync GovOPlaN module state

src/govoplan_organizations/backend/directory.py

@@ -1,19 +1,13 @@
 from __future__ import annotations
 
-from sqlalchemy import or_
-
 from govoplan_core.core.organizations import (
     OrganizationDirectory,
-    OrganizationFunctionAssignmentRef,
     OrganizationFunctionRef,
     OrganizationUnitRef,
 )
-from govoplan_core.core.identity import IdentityDirectory
 from govoplan_core.db.session import get_database
-from govoplan_core.security.time import utc_now
 from govoplan_organizations.backend.db.models import (
     OrganizationFunction,
-    OrganizationFunctionAssignment,
     OrganizationUnit,
 )
 
@@ -50,28 +44,7 @@ def _function_ref(item: OrganizationFunction) -> OrganizationFunctionRef:
     )
 
 
-def _assignment_ref(item: OrganizationFunctionAssignment) -> OrganizationFunctionAssignmentRef:
-    return OrganizationFunctionAssignmentRef(
-        id=item.id,
-        tenant_id=item.tenant_id,
-        identity_id=item.identity_id,
-        account_id=item.account_id,
-        function_id=item.function_id,
-        organization_unit_id=item.organization_unit_id,
-        applies_to_subunits=item.applies_to_subunits,
-        source=item.source,  # type: ignore[arg-type]
-        delegated_from_assignment_id=item.delegated_from_assignment_id,
-        acting_for_account_id=item.acting_for_account_id,
-        valid_from=item.valid_from,
-        valid_until=item.valid_until,
-        status=_status(item.is_active),  # type: ignore[arg-type]
-    )
-
-
 class SqlOrganizationDirectory(OrganizationDirectory):
-    def __init__(self, *, identity_directory: IdentityDirectory | None = None) -> None:
-        self._identity_directory = identity_directory
-
     def get_organization_unit(self, organization_unit_id: str) -> OrganizationUnitRef | None:
         with get_database().session() as session:
             item = session.get(OrganizationUnit, organization_unit_id)
@@ -121,70 +94,3 @@ class SqlOrganizationDirectory(OrganizationDirectory):
                 .all()
             )
             return tuple(_function_ref(item) for item in rows)
-
-    def get_function_assignment(self, assignment_id: str) -> OrganizationFunctionAssignmentRef | None:
-        with get_database().session() as session:
-            item = session.get(OrganizationFunctionAssignment, assignment_id)
-            return _assignment_ref(item) if item is not None else None
-
-    def function_assignments_for_account(
-        self,
-        account_id: str,
-        *,
-        tenant_id: str | None = None,
-    ) -> tuple[OrganizationFunctionAssignmentRef, ...]:
-        identity_ids: set[str] = set()
-        if self._identity_directory is not None:
-            identity = self._identity_directory.identity_for_account(account_id)
-            if identity is not None:
-                identity_ids.add(identity.id)
-        if not identity_ids:
-            return self._function_assignments_for_identity_ids(
-                identity_ids=(),
-                tenant_id=tenant_id,
-                fallback_account_id=account_id,
-            )
-        return self._function_assignments_for_identity_ids(
-            identity_ids=tuple(identity_ids),
-            tenant_id=tenant_id,
-            fallback_account_id=account_id,
-        )
-
-    def function_assignments_for_identity(
-        self,
-        identity_id: str,
-        *,
-        tenant_id: str | None = None,
-    ) -> tuple[OrganizationFunctionAssignmentRef, ...]:
-        return self._function_assignments_for_identity_ids(identity_ids=(identity_id,), tenant_id=tenant_id)
-
-    def _function_assignments_for_identity_ids(
-        self,
-        *,
-        identity_ids: tuple[str, ...],
-        tenant_id: str | None = None,
-        fallback_account_id: str | None = None,
-    ) -> tuple[OrganizationFunctionAssignmentRef, ...]:
-        now = utc_now()
-        with get_database().session() as session:
-            identity_or_account_filter = OrganizationFunctionAssignment.identity_id.in_(identity_ids)
-            if fallback_account_id:
-                identity_or_account_filter = or_(
-                    identity_or_account_filter,
-                    OrganizationFunctionAssignment.account_id == fallback_account_id,
-                )
-            query = (
-                session.query(OrganizationFunctionAssignment)
-                .join(OrganizationFunction, OrganizationFunction.id == OrganizationFunctionAssignment.function_id)
-                .filter(
-                    identity_or_account_filter,
-                    OrganizationFunctionAssignment.is_active.is_(True),
-                    OrganizationFunction.is_active.is_(True),
-                    or_(OrganizationFunctionAssignment.valid_from.is_(None), OrganizationFunctionAssignment.valid_from <= now),
-                    or_(OrganizationFunctionAssignment.valid_until.is_(None), OrganizationFunctionAssignment.valid_until > now),
-                )
-                .order_by(OrganizationFunctionAssignment.created_at.asc())
-            )
-            if tenant_id is not None:
-                query = query.filter(OrganizationFunctionAssignment.tenant_id == tenant_id)
-            return tuple(_assignment_ref(item) for item in query.all())