Sync GovOPlaN module state

src/govoplan_idm/backend/directory.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+from sqlalchemy import or_
+
+from govoplan_core.core.idm import IdmDirectory, OrganizationFunctionAssignmentRef
+from govoplan_core.db.session import get_database
+from govoplan_core.security.time import utc_now
+from govoplan_identity.backend.db.models import IdentityAccountLink
+from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment
+from govoplan_organizations.backend.db.models import OrganizationFunction
+
+
+def _status(active: bool) -> str:
+    return "active" if active else "inactive"
+
+
+def _assignment_ref(item: IdmOrganizationFunctionAssignment) -> 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 SqlIdmDirectory(IdmDirectory):
+    def get_organization_function_assignment(self, assignment_id: str) -> OrganizationFunctionAssignmentRef | None:
+        with get_database().session() as session:
+            item = session.get(IdmOrganizationFunctionAssignment, assignment_id)
+            return _assignment_ref(item) if item is not None else None
+
+    def organization_function_assignments_for_identity(
+        self,
+        identity_id: str,
+        *,
+        tenant_id: str | None = None,
+    ) -> tuple[OrganizationFunctionAssignmentRef, ...]:
+        return self._assignments_for_identity_ids(identity_ids=(identity_id,), tenant_id=tenant_id)
+
+    def organization_function_assignments_for_account(
+        self,
+        account_id: str,
+        *,
+        tenant_id: str | None = None,
+    ) -> tuple[OrganizationFunctionAssignmentRef, ...]:
+        with get_database().session() as session:
+            identity_ids = [
+                row[0]
+                for row in session.query(IdentityAccountLink.identity_id)
+                .filter(IdentityAccountLink.account_id == account_id)
+                .all()
+            ]
+        if not identity_ids:
+            return ()
+        return self._assignments_for_identity_ids(identity_ids=tuple(identity_ids), tenant_id=tenant_id, account_id=account_id)
+
+    def _assignments_for_identity_ids(
+        self,
+        *,
+        identity_ids: tuple[str, ...],
+        tenant_id: str | None = None,
+        account_id: str | None = None,
+    ) -> tuple[OrganizationFunctionAssignmentRef, ...]:
+        if not identity_ids:
+            return ()
+        now = utc_now()
+        with get_database().session() as session:
+            query = (
+                session.query(IdmOrganizationFunctionAssignment)
+                .join(OrganizationFunction, OrganizationFunction.id == IdmOrganizationFunctionAssignment.function_id)
+                .filter(
+                    IdmOrganizationFunctionAssignment.identity_id.in_(identity_ids),
+                    IdmOrganizationFunctionAssignment.is_active.is_(True),
+                    OrganizationFunction.is_active.is_(True),
+                    or_(IdmOrganizationFunctionAssignment.valid_from.is_(None), IdmOrganizationFunctionAssignment.valid_from <= now),
+                    or_(IdmOrganizationFunctionAssignment.valid_until.is_(None), IdmOrganizationFunctionAssignment.valid_until > now),
+                )
+                .order_by(IdmOrganizationFunctionAssignment.created_at.asc())
+            )
+            if account_id is not None:
+                query = query.filter(or_(IdmOrganizationFunctionAssignment.account_id.is_(None), IdmOrganizationFunctionAssignment.account_id == account_id))
+            if tenant_id is not None:
+                query = query.filter(IdmOrganizationFunctionAssignment.tenant_id == tenant_id)
+            return tuple(_assignment_ref(item) for item in query.all())