diff --git a/README.md b/README.md index 6096eef..0f8660a 100644 --- a/README.md +++ b/README.md @@ -42,8 +42,13 @@ requirements, audit detail, and retention behavior. ## Module Contract -The module registers the `organizations.directory` capability from -`govoplan_core.core.organizations`. +The module registers two capabilities from +`govoplan_core.core.organizations`: + +- `organizations.directory` for backward-compatible direct unit/function + lookup; +- `organizations.hierarchyDirectory` for typed, tenant-safe, explicitly + structure-scoped hierarchy and path resolution. Feature modules should consume the capability instead of importing organization ORM models. diff --git a/docs/ORGANIZATION_MODEL.md b/docs/ORGANIZATION_MODEL.md index edf4e9d..a1fe56d 100644 --- a/docs/ORGANIZATION_MODEL.md +++ b/docs/ORGANIZATION_MODEL.md @@ -66,6 +66,36 @@ The close-out condition is that Access role resolution works with canonical Organizations plus IDM installed and still works through projection fallback for transition deployments. +## Directory Contracts + +`organizations.directory` remains the small compatibility contract for direct +unit and function lookup. It deliberately retains the legacy `parent_id` +projection needed by existing Access and IDM integrations. + +`organizations.hierarchyDirectory` is the structure-aware contract for new +consumers. It provides: + +- tenant-scoped unit-type and function-type references; +- function resolution by type and an explicit set of units; +- unit resolution by type, optionally bounded to one named structure; +- batched ancestor, descendant, and path resolution with a required structure, + optional relation-type filter, and bounded depth; +- the exact structure, relation type, and relation edge for every path step; +- explicit missing, inactive, unreachable, invalid-filter, cycle, and + depth-limit state. + +An edge is traversed from `source_unit_id` to `target_unit_id` for descendant +lookups and in reverse for ancestor lookups. Consumers must select a structure; +the provider never treats `parent_id` or one arbitrary structure as the +institution's universal hierarchy. + +Committed changes emit versioned `organizations...v1` +platform events for units, functions, their types, structures, relation types, +and relation edges. The event payload contains schema version `1`, tenant and +resource references, status, changed fields, and routing-relevant IDs so +directory consumers can invalidate derived addresses without importing this +module's models. + ## UI Boundary The organization workspace at `/organizations` is the primary module UI for diff --git a/src/govoplan_organizations/backend/api/v1/routes.py b/src/govoplan_organizations/backend/api/v1/routes.py index 27877a3..2459e07 100644 --- a/src/govoplan_organizations/backend/api/v1/routes.py +++ b/src/govoplan_organizations/backend/api/v1/routes.py @@ -1,9 +1,11 @@ from __future__ import annotations +from datetime import datetime import re from typing import Any, TypeVar from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy import inspect as sqlalchemy_inspect from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -15,6 +17,16 @@ from govoplan_core.core.configuration_control import ( record_configuration_change_applied, ) from govoplan_core.core.principal_cache import invalidate_auth_principals +from govoplan_core.core.organizations import ( + ORGANIZATION_LIFECYCLE_EVENT_SCHEMA_VERSION, + organization_lifecycle_event_type, +) +from govoplan_core.core.events import ( + EventObjectRef, + EventTenantRef, + PlatformEvent, + emit_platform_event, +) from govoplan_core.db.session import get_session from govoplan_organizations.backend.db.models import ( OrganizationFunction, @@ -131,14 +143,29 @@ def _ensure_unique_slug(session: Session, model: type, tenant_id: str, slug: str def _commit(session: Session, item: ModelT) -> ModelT: - invalidate_auth_principals( - session, - tenant_id=getattr(item, "tenant_id", None), - source_module="organizations", - resource_type=item.__class__.__name__, - resource_id=str(getattr(item, "id", getattr(item, "tenant_id", "system"))), - ) + lifecycle = _organization_lifecycle_change(item) try: + session.flush() + invalidate_auth_principals( + session, + tenant_id=getattr(item, "tenant_id", None), + source_module="organizations", + resource_type=item.__class__.__name__, + resource_id=str( + getattr( + item, + "id", + getattr(item, "tenant_id", "system"), + ) + ), + ) + if lifecycle is not None: + _emit_organization_lifecycle_event( + session, + item, + action=lifecycle[0], + changes=lifecycle[1], + ) session.commit() except IntegrityError as exc: session.rollback() @@ -147,6 +174,121 @@ def _commit(session: Session, item: ModelT) -> ModelT: return item +_LIFECYCLE_RESOURCES = { + OrganizationUnitType: "unit_type", + OrganizationStructure: "structure", + OrganizationRelationType: "relation_type", + OrganizationUnit: "unit", + OrganizationRelation: "relation", + OrganizationFunctionType: "function_type", + OrganizationFunction: "function", +} + + +def _organization_lifecycle_change( + item: object, +) -> tuple[str, dict[str, dict[str, object | None]]] | None: + resource_type = _LIFECYCLE_RESOURCES.get(type(item)) + if resource_type is None: + return None + state = sqlalchemy_inspect(item) + changes: dict[str, dict[str, object | None]] = {} + for attribute in state.mapper.column_attrs: + history = state.attrs[attribute.key].history + if not history.has_changes(): + continue + changes[attribute.key] = { + "before": ( + _organization_event_value(history.deleted[0]) + if history.deleted + else None + ), + "after": ( + _organization_event_value(history.added[0]) + if history.added + else None + ), + } + if state.pending: + action = "created" + elif ( + "is_active" in changes + and changes["is_active"]["after"] is False + ): + action = "deactivated" + elif resource_type == "unit" and "parent_id" in changes: + action = "moved" + else: + action = "updated" + return action, changes + + +def _organization_event_value(value: object) -> object: + if isinstance(value, datetime): + return value.isoformat() + return value + + +def _emit_organization_lifecycle_event( + session: Session, + item: object, + *, + action: str, + changes: dict[str, dict[str, object | None]], +) -> None: + resource_type = _LIFECYCLE_RESOURCES[type(item)] + tenant_id = str(getattr(item, "tenant_id")) + item_id = str(getattr(item, "id")) + payload: dict[str, object] = { + "schema_version": ORGANIZATION_LIFECYCLE_EVENT_SCHEMA_VERSION, + "tenant_id": tenant_id, + "resource_type": resource_type, + "resource_id": item_id, + "status": ( + "active" + if bool(getattr(item, "is_active", True)) + else "inactive" + ), + "changed_fields": sorted(changes), + "changes": changes, + } + for field in ( + "slug", + "unit_type_id", + "organization_unit_id", + "function_type_id", + "structure_id", + "relation_type_id", + "source_unit_id", + "target_unit_id", + ): + value = getattr(item, field, None) + if value is not None: + payload[field] = str(value) + emit_platform_event( + session, + PlatformEvent( + type=organization_lifecycle_event_type( + resource_type, # type: ignore[arg-type] + action, # type: ignore[arg-type] + ), + module_id="organizations", + tenant=EventTenantRef(id=tenant_id), + subject=EventObjectRef( + type=f"organization_{resource_type}", + id=item_id, + label=str(getattr(item, "name", None) or "") or None, + ), + resource=EventObjectRef( + type=f"organization_{resource_type}", + id=item_id, + ), + payload=payload, + classification="internal", + ), + ) + + def _requires_organization_change_request(session: Session, tenant_id: str) -> bool: item = session.query(OrganizationTenantSettings).filter(OrganizationTenantSettings.tenant_id == tenant_id).one_or_none() return bool(item and item.require_model_change_requests) diff --git a/src/govoplan_organizations/backend/directory.py b/src/govoplan_organizations/backend/directory.py index c5fcace..51e754d 100644 --- a/src/govoplan_organizations/backend/directory.py +++ b/src/govoplan_organizations/backend/directory.py @@ -1,17 +1,48 @@ from __future__ import annotations +from collections import deque +from collections.abc import Callable, Iterator, Sequence +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import datetime, timezone + +from sqlalchemy import or_ +from sqlalchemy.orm import Session + from govoplan_core.core.organizations import ( OrganizationDirectory, OrganizationFunctionRef, + OrganizationFunctionTypeRef, + OrganizationFunctionTypeResolution, + OrganizationHierarchyDirection, + OrganizationHierarchyDirectory, + OrganizationHierarchyEdgeRef, + OrganizationHierarchyMatchRef, + OrganizationHierarchyPathResolution, + OrganizationHierarchyResolution, + OrganizationRelationTypeRef, + OrganizationResolutionStatus, + OrganizationStructureRef, OrganizationUnitRef, + OrganizationUnitTypeRef, + OrganizationUnitTypeResolution, ) from govoplan_core.db.session import get_database from govoplan_organizations.backend.db.models import ( OrganizationFunction, + OrganizationFunctionType, + OrganizationRelation, + OrganizationRelationType, + OrganizationStructure, OrganizationUnit, + OrganizationUnitType, ) +MAX_DIRECTORY_BATCH = 500 +MAX_HIERARCHY_DEPTH = 100 + + def _status(active: bool) -> str: return "active" if active else "inactive" @@ -29,6 +60,17 @@ def _unit_ref(item: OrganizationUnit) -> OrganizationUnitRef: ) +def _unit_type_ref(item: OrganizationUnitType) -> OrganizationUnitTypeRef: + return OrganizationUnitTypeRef( + id=item.id, + tenant_id=item.tenant_id, + slug=item.slug, + name=item.name, + description=item.description, + status=_status(item.is_active), # type: ignore[arg-type] + ) + + def _function_ref(item: OrganizationFunction) -> OrganizationFunctionRef: return OrganizationFunctionRef( id=item.id, @@ -44,24 +86,143 @@ def _function_ref(item: OrganizationFunction) -> OrganizationFunctionRef: ) -class SqlOrganizationDirectory(OrganizationDirectory): - def get_organization_unit(self, organization_unit_id: str) -> OrganizationUnitRef | None: - with get_database().session() as session: +def _function_type_ref( + item: OrganizationFunctionType, +) -> OrganizationFunctionTypeRef: + return OrganizationFunctionTypeRef( + id=item.id, + tenant_id=item.tenant_id, + slug=item.slug, + name=item.name, + organization_unit_type_id=item.organization_unit_type_id, + description=item.description, + delegable=item.delegable, + act_in_place_allowed=item.act_in_place_allowed, + status=_status(item.is_active), # type: ignore[arg-type] + ) + + +def _structure_ref(item: OrganizationStructure) -> OrganizationStructureRef: + return OrganizationStructureRef( + id=item.id, + tenant_id=item.tenant_id, + slug=item.slug, + name=item.name, + structure_kind=item.structure_kind, + description=item.description, + status=_status(item.is_active), # type: ignore[arg-type] + ) + + +def _relation_type_ref( + item: OrganizationRelationType, +) -> OrganizationRelationTypeRef: + return OrganizationRelationTypeRef( + id=item.id, + tenant_id=item.tenant_id, + slug=item.slug, + name=item.name, + structure_id=item.structure_id, + source_unit_type_id=item.source_unit_type_id, + target_unit_type_id=item.target_unit_type_id, + is_hierarchical=item.is_hierarchical, + allow_cycles=item.allow_cycles, + description=item.description, + status=_status(item.is_active), # type: ignore[arg-type] + ) + + +def _unique_ids(values: Sequence[str], *, label: str) -> tuple[str, ...]: + result = tuple(dict.fromkeys(str(value).strip() for value in values)) + if any(not value for value in result): + raise ValueError(f"{label} must not contain empty IDs.") + if len(result) > MAX_DIRECTORY_BATCH: + raise ValueError( + f"{label} is limited to {MAX_DIRECTORY_BATCH} entries." + ) + return result + + +def _validated_direction( + value: OrganizationHierarchyDirection, +) -> OrganizationHierarchyDirection: + if value not in {"ancestors", "descendants"}: + raise ValueError( + "Hierarchy direction must be ancestors or descendants." + ) + return value + + +def _validated_depth(value: int) -> int: + depth = int(value) + if depth < 1 or depth > MAX_HIERARCHY_DEPTH: + raise ValueError( + f"Hierarchy depth must be between 1 and {MAX_HIERARCHY_DEPTH}." + ) + return depth + + +@dataclass(slots=True) +class _HierarchyGraph: + status: OrganizationResolutionStatus + structure_id: str + structure: OrganizationStructureRef | None + relation_type_ids: tuple[str, ...] + units: dict[str, OrganizationUnitRef] + outgoing: dict[str, tuple[OrganizationHierarchyEdgeRef, ...]] + incoming: dict[str, tuple[OrganizationHierarchyEdgeRef, ...]] + diagnostics: tuple[str, ...] + + +class SqlOrganizationDirectory( + OrganizationDirectory, + OrganizationHierarchyDirectory, +): + def __init__( + self, + *, + session_factory: Callable[[], Session] | None = None, + ) -> None: + self._session_factory = session_factory + + @contextmanager + def _session(self) -> Iterator[Session]: + if self._session_factory is None: + with get_database().session() as session: + yield session + return + session = self._session_factory() + try: + yield session + finally: + session.close() + + def get_organization_unit( + self, + organization_unit_id: str, + ) -> OrganizationUnitRef | None: + with self._session() as session: item = session.get(OrganizationUnit, organization_unit_id) return _unit_ref(item) if item is not None else None - def organization_units_for_tenant(self, tenant_id: str) -> tuple[OrganizationUnitRef, ...]: - with get_database().session() as session: + def organization_units_for_tenant( + self, + tenant_id: str, + ) -> tuple[OrganizationUnitRef, ...]: + with self._session() as session: rows = ( session.query(OrganizationUnit) .filter(OrganizationUnit.tenant_id == tenant_id) - .order_by(OrganizationUnit.name.asc()) + .order_by(OrganizationUnit.name.asc(), OrganizationUnit.id) .all() ) return tuple(_unit_ref(item) for item in rows) - def get_function(self, function_id: str) -> OrganizationFunctionRef | None: - with get_database().session() as session: + def get_function( + self, + function_id: str, + ) -> OrganizationFunctionRef | None: + with self._session() as session: item = session.get(OrganizationFunction, function_id) return _function_ref(item) if item is not None else None @@ -71,7 +232,10 @@ class SqlOrganizationDirectory(OrganizationDirectory): *, include_subunits: bool = False, ) -> tuple[OrganizationFunctionRef, ...]: - with get_database().session() as session: + with self._session() as session: + unit = session.get(OrganizationUnit, organization_unit_id) + if unit is None: + return () unit_ids = {organization_unit_id} if include_subunits: pending = [organization_unit_id] @@ -80,7 +244,10 @@ class SqlOrganizationDirectory(OrganizationDirectory): child_ids = [ row[0] for row in session.query(OrganizationUnit.id) - .filter(OrganizationUnit.parent_id == parent_id) + .filter( + OrganizationUnit.tenant_id == unit.tenant_id, + OrganizationUnit.parent_id == parent_id, + ) .all() ] for child_id in child_ids: @@ -89,8 +256,659 @@ class SqlOrganizationDirectory(OrganizationDirectory): pending.append(child_id) rows = ( session.query(OrganizationFunction) - .filter(OrganizationFunction.organization_unit_id.in_(unit_ids)) - .order_by(OrganizationFunction.name.asc()) + .filter( + OrganizationFunction.tenant_id == unit.tenant_id, + OrganizationFunction.organization_unit_id.in_(unit_ids), + ) + .order_by( + OrganizationFunction.name.asc(), + OrganizationFunction.id, + ) .all() ) return tuple(_function_ref(item) for item in rows) + + def get_unit_type( + self, + tenant_id: str, + unit_type_id: str, + ) -> OrganizationUnitTypeRef | None: + with self._session() as session: + item = session.get(OrganizationUnitType, unit_type_id) + if item is None or item.tenant_id != tenant_id: + return None + return _unit_type_ref(item) + + def get_function_type( + self, + tenant_id: str, + function_type_id: str, + ) -> OrganizationFunctionTypeRef | None: + with self._session() as session: + item = session.get(OrganizationFunctionType, function_type_id) + if item is None or item.tenant_id != tenant_id: + return None + return _function_type_ref(item) + + def resolve_functions_by_type( + self, + tenant_id: str, + function_type_id: str, + *, + organization_unit_ids: Sequence[str] = (), + ) -> OrganizationFunctionTypeResolution: + requested_ids = _unique_ids( + organization_unit_ids, + label="Organization unit scope", + ) + with self._session() as session: + function_type = session.get( + OrganizationFunctionType, + function_type_id, + ) + if ( + function_type is None + or function_type.tenant_id != tenant_id + ): + return OrganizationFunctionTypeResolution( + tenant_id=tenant_id, + function_type_id=function_type_id, + requested_unit_ids=requested_ids, + status="missing", + diagnostics=("function_type_missing",), + ) + units: dict[str, OrganizationUnit] = {} + if requested_ids: + units = { + item.id: item + for item in session.query(OrganizationUnit) + .filter( + OrganizationUnit.tenant_id == tenant_id, + OrganizationUnit.id.in_(requested_ids), + ) + .all() + } + query = session.query(OrganizationFunction).filter( + OrganizationFunction.tenant_id == tenant_id, + OrganizationFunction.function_type_id == function_type_id, + ) + if requested_ids: + query = query.filter( + OrganizationFunction.organization_unit_id.in_( + tuple(units) + ) + ) + rows = query.order_by( + OrganizationFunction.organization_unit_id, + OrganizationFunction.name, + OrganizationFunction.id, + ).all() + relevant_unit_ids = ( + requested_ids + if requested_ids + else tuple( + dict.fromkeys( + item.organization_unit_id for item in rows + ) + ) + ) + if not requested_ids and relevant_unit_ids: + units = { + item.id: item + for item in session.query(OrganizationUnit) + .filter( + OrganizationUnit.tenant_id == tenant_id, + OrganizationUnit.id.in_(relevant_unit_ids), + ) + .all() + } + missing = tuple( + item_id + for item_id in relevant_unit_ids + if item_id not in units + ) + inactive = tuple( + item_id + for item_id in relevant_unit_ids + if item_id in units and not units[item_id].is_active + ) + diagnostics = [] + if missing: + diagnostics.append("organization_units_missing") + if inactive: + diagnostics.append("organization_units_inactive") + if not rows: + diagnostics.append("no_matching_functions") + return OrganizationFunctionTypeResolution( + tenant_id=tenant_id, + function_type_id=function_type_id, + requested_unit_ids=requested_ids, + status=( + "active" if function_type.is_active else "inactive" + ), + function_type=_function_type_ref(function_type), + matches=tuple(_function_ref(item) for item in rows), + missing_unit_ids=missing, + inactive_unit_ids=inactive, + diagnostics=tuple(diagnostics), + ) + + def resolve_units_by_type( + self, + tenant_id: str, + unit_type_id: str, + *, + structure_id: str | None = None, + root_unit_id: str | None = None, + relation_type_ids: Sequence[str] = (), + direction: OrganizationHierarchyDirection = "descendants", + max_depth: int = 10, + ) -> OrganizationUnitTypeResolution: + relation_ids = _unique_ids( + relation_type_ids, + label="Relation type filter", + ) + direction = _validated_direction(direction) + depth = _validated_depth(max_depth) + if bool(structure_id) != bool(root_unit_id): + return OrganizationUnitTypeResolution( + tenant_id=tenant_id, + unit_type_id=unit_type_id, + status="invalid", + structure_id=structure_id, + root_unit_id=root_unit_id, + diagnostics=( + "structure_and_root_must_be_supplied_together", + ), + ) + with self._session() as session: + unit_type = session.get(OrganizationUnitType, unit_type_id) + if unit_type is None or unit_type.tenant_id != tenant_id: + return OrganizationUnitTypeResolution( + tenant_id=tenant_id, + unit_type_id=unit_type_id, + status="missing", + structure_id=structure_id, + root_unit_id=root_unit_id, + diagnostics=("unit_type_missing",), + ) + rows = ( + session.query(OrganizationUnit) + .filter( + OrganizationUnit.tenant_id == tenant_id, + OrganizationUnit.unit_type_id == unit_type_id, + ) + .order_by(OrganizationUnit.name, OrganizationUnit.id) + .all() + ) + diagnostics: tuple[str, ...] = () + status: OrganizationResolutionStatus = ( + "active" if unit_type.is_active else "inactive" + ) + if structure_id and root_unit_id: + hierarchy = self.resolve_hierarchy_relatives( + tenant_id, + (root_unit_id,), + structure_id=structure_id, + relation_type_ids=relation_ids, + direction=direction, + max_depth=depth, + )[0] + scope_ids = { + root_unit_id, + *(match.unit.id for match in hierarchy.matches), + } + rows = [item for item in rows if item.id in scope_ids] + diagnostics = hierarchy.diagnostics + if hierarchy.status in {"missing", "invalid"}: + status = hierarchy.status + elif hierarchy.status == "inactive" and status == "active": + status = "inactive" + return OrganizationUnitTypeResolution( + tenant_id=tenant_id, + unit_type_id=unit_type_id, + status=status, + unit_type=_unit_type_ref(unit_type), + structure_id=structure_id, + root_unit_id=root_unit_id, + matches=tuple(_unit_ref(item) for item in rows), + diagnostics=diagnostics, + ) + + def resolve_hierarchy_relatives( + self, + tenant_id: str, + organization_unit_ids: Sequence[str], + *, + structure_id: str, + relation_type_ids: Sequence[str] = (), + direction: OrganizationHierarchyDirection = "ancestors", + max_depth: int = 10, + ) -> tuple[OrganizationHierarchyResolution, ...]: + root_ids = _unique_ids( + organization_unit_ids, + label="Organization hierarchy roots", + ) + direction = _validated_direction(direction) + depth = _validated_depth(max_depth) + relation_ids = _unique_ids( + relation_type_ids, + label="Relation type filter", + ) + with self._session() as session: + graph = self._hierarchy_graph( + session, + tenant_id=tenant_id, + structure_id=structure_id, + relation_type_ids=relation_ids, + required_unit_ids=root_ids, + ) + return tuple( + self._relative_resolution( + graph, + tenant_id=tenant_id, + root_unit_id=root_id, + direction=direction, + max_depth=depth, + ) + for root_id in root_ids + ) + + def resolve_hierarchy_paths( + self, + tenant_id: str, + unit_pairs: Sequence[tuple[str, str]], + *, + structure_id: str, + relation_type_ids: Sequence[str] = (), + direction: OrganizationHierarchyDirection = "descendants", + max_depth: int = 10, + ) -> tuple[OrganizationHierarchyPathResolution, ...]: + if len(unit_pairs) > MAX_DIRECTORY_BATCH: + raise ValueError( + f"Organization path batches are limited to " + f"{MAX_DIRECTORY_BATCH} entries." + ) + pairs = tuple( + (str(source).strip(), str(target).strip()) + for source, target in unit_pairs + ) + if any(not source or not target for source, target in pairs): + raise ValueError("Organization paths require source and target IDs.") + direction = _validated_direction(direction) + depth = _validated_depth(max_depth) + relation_ids = _unique_ids( + relation_type_ids, + label="Relation type filter", + ) + required_ids = tuple( + dict.fromkeys( + item + for pair in pairs + for item in pair + ) + ) + with self._session() as session: + graph = self._hierarchy_graph( + session, + tenant_id=tenant_id, + structure_id=structure_id, + relation_type_ids=relation_ids, + required_unit_ids=required_ids, + ) + cache: dict[str, OrganizationHierarchyResolution] = {} + results: list[OrganizationHierarchyPathResolution] = [] + for source_id, target_id in pairs: + if source_id not in cache: + cache[source_id] = self._relative_resolution( + graph, + tenant_id=tenant_id, + root_unit_id=source_id, + direction=direction, + max_depth=depth, + ) + relative = cache[source_id] + target = graph.units.get(target_id) + match = next( + ( + item + for item in relative.matches + if item.unit.id == target_id + ), + None, + ) + if source_id == target_id and relative.root is not None: + status: OrganizationResolutionStatus = ( + "active" + if relative.root.status == "active" + else "inactive" + ) + path: tuple[OrganizationHierarchyEdgeRef, ...] = () + elif relative.root is None or target is None: + status = "missing" + path = () + elif match is None: + status = ( + "invalid" + if relative.status == "invalid" + else "unreachable" + ) + path = () + else: + status = ( + "active" + if ( + relative.root.status == "active" + and target.status == "active" + ) + else "inactive" + ) + path = match.path + results.append( + OrganizationHierarchyPathResolution( + tenant_id=tenant_id, + source_unit_id=source_id, + target_unit_id=target_id, + direction=direction, + structure_id=structure_id, + relation_type_ids=graph.relation_type_ids, + max_depth=depth, + status=status, + source=relative.root, + target=target, + path=path, + cycle_detected=relative.cycle_detected, + depth_limited=relative.depth_limited, + diagnostics=relative.diagnostics, + ) + ) + return tuple(results) + + def _hierarchy_graph( + self, + session: Session, + *, + tenant_id: str, + structure_id: str, + relation_type_ids: tuple[str, ...], + required_unit_ids: tuple[str, ...], + ) -> _HierarchyGraph: + structure = session.get(OrganizationStructure, structure_id) + if structure is None or structure.tenant_id != tenant_id: + return _HierarchyGraph( + status="missing", + structure_id=structure_id, + structure=None, + relation_type_ids=relation_type_ids, + units={}, + outgoing={}, + incoming={}, + diagnostics=("structure_missing",), + ) + relation_query = session.query(OrganizationRelationType).filter( + OrganizationRelationType.tenant_id == tenant_id, + OrganizationRelationType.is_hierarchical.is_(True), + or_( + OrganizationRelationType.structure_id.is_(None), + OrganizationRelationType.structure_id == structure_id, + ), + ) + if relation_type_ids: + relation_query = relation_query.filter( + OrganizationRelationType.id.in_(relation_type_ids) + ) + relation_types = relation_query.all() + relation_type_by_id = { + item.id: item + for item in relation_types + if item.is_active + } + diagnostics: list[str] = [] + invalid_relation_filter = False + if relation_type_ids: + missing_relation_ids = tuple( + item + for item in relation_type_ids + if item not in {row.id for row in relation_types} + ) + inactive_relation_ids = tuple( + item.id for item in relation_types if not item.is_active + ) + if missing_relation_ids: + diagnostics.append("relation_types_missing_or_invalid") + invalid_relation_filter = True + if inactive_relation_ids: + diagnostics.append("relation_types_inactive") + invalid_relation_filter = True + selected_ids = tuple(sorted(relation_type_by_id)) + now = datetime.now(timezone.utc) + relations = [] + if selected_ids: + relations = ( + session.query(OrganizationRelation) + .filter( + OrganizationRelation.tenant_id == tenant_id, + OrganizationRelation.structure_id == structure_id, + OrganizationRelation.relation_type_id.in_(selected_ids), + OrganizationRelation.is_active.is_(True), + ) + .order_by(OrganizationRelation.id) + .all() + ) + relations = [ + item + for item in relations + if ( + ( + item.valid_from is None + or _utc_datetime(item.valid_from) <= now + ) + and ( + item.valid_until is None + or _utc_datetime(item.valid_until) > now + ) + ) + ] + unit_ids = { + *required_unit_ids, + *(item.source_unit_id for item in relations), + *(item.target_unit_id for item in relations), + } + units = { + item.id: _unit_ref(item) + for item in session.query(OrganizationUnit) + .filter( + OrganizationUnit.tenant_id == tenant_id, + OrganizationUnit.id.in_(unit_ids), + ) + .all() + } if unit_ids else {} + structure_ref = _structure_ref(structure) + outgoing_lists: dict[ + str, + list[OrganizationHierarchyEdgeRef], + ] = {} + incoming_lists: dict[ + str, + list[OrganizationHierarchyEdgeRef], + ] = {} + for relation in relations: + relation_type = relation_type_by_id[relation.relation_type_id] + if ( + relation.source_unit_id not in units + or relation.target_unit_id not in units + ): + diagnostics.append( + f"relation_unit_missing:{relation.id}" + ) + continue + edge = OrganizationHierarchyEdgeRef( + id=relation.id, + tenant_id=tenant_id, + structure=structure_ref, + relation_type=_relation_type_ref(relation_type), + source_unit_id=relation.source_unit_id, + target_unit_id=relation.target_unit_id, + valid_from=relation.valid_from, + valid_until=relation.valid_until, + ) + outgoing_lists.setdefault(edge.source_unit_id, []).append(edge) + incoming_lists.setdefault(edge.target_unit_id, []).append(edge) + return _HierarchyGraph( + status=( + "inactive" + if not structure.is_active + else "invalid" + if invalid_relation_filter + else "active" + ), + structure_id=structure_id, + structure=structure_ref, + relation_type_ids=selected_ids, + units=units, + outgoing={ + key: tuple(sorted(value, key=lambda item: item.id)) + for key, value in outgoing_lists.items() + }, + incoming={ + key: tuple(sorted(value, key=lambda item: item.id)) + for key, value in incoming_lists.items() + }, + diagnostics=tuple(dict.fromkeys(diagnostics)), + ) + + def _relative_resolution( + self, + graph: _HierarchyGraph, + *, + tenant_id: str, + root_unit_id: str, + direction: OrganizationHierarchyDirection, + max_depth: int, + ) -> OrganizationHierarchyResolution: + root = graph.units.get(root_unit_id) + if root is None: + return OrganizationHierarchyResolution( + tenant_id=tenant_id, + root_unit_id=root_unit_id, + direction=direction, + structure_id=( + graph.structure_id + ), + relation_type_ids=graph.relation_type_ids, + max_depth=max_depth, + status="missing", + diagnostics=tuple( + dict.fromkeys( + (*graph.diagnostics, "root_unit_missing") + ) + ), + ) + if graph.status in {"missing", "invalid"}: + return OrganizationHierarchyResolution( + tenant_id=tenant_id, + root_unit_id=root_unit_id, + direction=direction, + structure_id=( + graph.structure_id + ), + relation_type_ids=graph.relation_type_ids, + max_depth=max_depth, + status=graph.status, + root=root, + diagnostics=graph.diagnostics, + ) + adjacency = ( + graph.incoming + if direction == "ancestors" + else graph.outgoing + ) + pending = deque( + [(root_unit_id, 0, (), frozenset({root_unit_id}))] + ) + best_depth = {root_unit_id: 0} + matches: dict[str, OrganizationHierarchyMatchRef] = {} + diagnostics = list(graph.diagnostics) + cycle_detected = False + depth_limited = False + while pending: + current_id, current_depth, path, path_units = pending.popleft() + edges = adjacency.get(current_id, ()) + if current_depth >= max_depth: + if edges: + depth_limited = True + continue + for edge in edges: + next_id = ( + edge.source_unit_id + if direction == "ancestors" + else edge.target_unit_id + ) + if next_id in path_units: + cycle_detected = True + diagnostics.append(f"cycle_detected:{edge.id}") + continue + unit = graph.units.get(next_id) + if unit is None: + diagnostics.append(f"unit_missing:{next_id}") + continue + next_depth = current_depth + 1 + previous_depth = best_depth.get(next_id) + if ( + previous_depth is not None + and previous_depth <= next_depth + ): + continue + next_path = (*path, edge) + best_depth[next_id] = next_depth + matches[next_id] = OrganizationHierarchyMatchRef( + unit=unit, + depth=next_depth, + path=next_path, + ) + pending.append( + ( + next_id, + next_depth, + next_path, + path_units | {next_id}, + ) + ) + if depth_limited: + diagnostics.append("maximum_depth_reached") + return OrganizationHierarchyResolution( + tenant_id=tenant_id, + root_unit_id=root_unit_id, + direction=direction, + structure_id=graph.structure_id, + relation_type_ids=graph.relation_type_ids, + max_depth=max_depth, + status=( + "active" + if root.status == "active" and graph.status == "active" + else "inactive" + ), + root=root, + matches=tuple( + sorted( + matches.values(), + key=lambda item: ( + item.depth, + item.unit.name.casefold(), + item.unit.id, + ), + ) + ), + cycle_detected=cycle_detected, + depth_limited=depth_limited, + diagnostics=tuple(dict.fromkeys(diagnostics)), + ) + + +def _utc_datetime(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +__all__ = ["SqlOrganizationDirectory"] diff --git a/src/govoplan_organizations/backend/manifest.py b/src/govoplan_organizations/backend/manifest.py index 00afbbe..e5941b4 100644 --- a/src/govoplan_organizations/backend/manifest.py +++ b/src/govoplan_organizations/backend/manifest.py @@ -10,12 +10,16 @@ from govoplan_core.core.modules import ( FrontendRoute, MigrationSpec, ModuleContext, + ModuleInterfaceProvider, ModuleManifest, NavItem, PermissionDefinition, RoleTemplate, ) -from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY +from govoplan_core.core.organizations import ( + CAPABILITY_ORGANIZATION_DIRECTORY, + CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY, +) from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base from govoplan_organizations.backend.db import models as organization_models # noqa: F401 - populate metadata @@ -90,6 +94,16 @@ manifest = ModuleManifest( version="0.1.8", required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR), optional_dependencies=("tenancy", "access", "audit", "policy"), + provides_interfaces=( + ModuleInterfaceProvider( + name="organizations.directory", + version="0.1.0", + ), + ModuleInterfaceProvider( + name="organizations.hierarchy_directory", + version="0.1.0", + ), + ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, route_factory=_route_factory, @@ -129,6 +143,9 @@ manifest = ModuleManifest( ), capability_factories={ CAPABILITY_ORGANIZATION_DIRECTORY: _organization_directory, + CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY: ( + _organization_directory + ), }, documentation=( DocumentationTopic( diff --git a/tests/test_hierarchy_directory.py b/tests/test_hierarchy_directory.py new file mode 100644 index 0000000..a7374dc --- /dev/null +++ b/tests/test_hierarchy_directory.py @@ -0,0 +1,421 @@ +from __future__ import annotations + +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, sessionmaker + +from govoplan_core.core.events import EventBus, event_bus_context +from govoplan_core.db.base import Base +from govoplan_organizations.backend.api.v1.routes import _commit +from govoplan_organizations.backend.db.models import ( + OrganizationFunction, + OrganizationFunctionType, + OrganizationRelation, + OrganizationRelationType, + OrganizationStructure, + OrganizationUnit, + OrganizationUnitType, +) +from govoplan_organizations.backend.directory import ( + SqlOrganizationDirectory, +) + + +class OrganizationHierarchyDirectoryTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all( + self.engine, + tables=[ + OrganizationUnitType.__table__, + OrganizationStructure.__table__, + OrganizationRelationType.__table__, + OrganizationUnit.__table__, + OrganizationRelation.__table__, + OrganizationFunctionType.__table__, + OrganizationFunction.__table__, + ], + ) + self.Session = sessionmaker( + bind=self.engine, + expire_on_commit=False, + ) + self.session: Session = self.Session() + self._seed() + self.directory = SqlOrganizationDirectory( + session_factory=self.Session, + ) + + def tearDown(self) -> None: + self.session.close() + Base.metadata.drop_all( + self.engine, + tables=[ + OrganizationFunction.__table__, + OrganizationFunctionType.__table__, + OrganizationRelation.__table__, + OrganizationUnit.__table__, + OrganizationRelationType.__table__, + OrganizationStructure.__table__, + OrganizationUnitType.__table__, + ], + ) + self.engine.dispose() + + def _seed(self) -> None: + self.unit_type = OrganizationUnitType( + id="type-department", + tenant_id="tenant-1", + slug="department", + name="Department", + is_active=True, + settings={}, + ) + self.function_type = OrganizationFunctionType( + id="type-intake", + tenant_id="tenant-1", + slug="intake", + name="Intake", + organization_unit_type_id=self.unit_type.id, + is_active=True, + settings={}, + ) + self.employer = OrganizationStructure( + id="structure-employer", + tenant_id="tenant-1", + slug="employer", + name="Employer hierarchy", + structure_kind="hierarchy", + is_active=True, + settings={}, + ) + self.project = OrganizationStructure( + id="structure-project", + tenant_id="tenant-1", + slug="project", + name="Project hierarchy", + structure_kind="hierarchy", + is_active=True, + settings={}, + ) + self.employer_parent = OrganizationRelationType( + id="relation-type-employer", + tenant_id="tenant-1", + structure_id=self.employer.id, + slug="contains", + name="Contains", + is_hierarchical=True, + allow_cycles=False, + is_active=True, + settings={}, + ) + self.project_parent = OrganizationRelationType( + id="relation-type-project", + tenant_id="tenant-1", + structure_id=self.project.id, + slug="project-contains", + name="Contains", + is_hierarchical=True, + allow_cycles=False, + is_active=True, + settings={}, + ) + self.root = self._unit("unit-root", "Root") + self.child = self._unit("unit-child", "Child") + self.grandchild = self._unit("unit-grandchild", "Grandchild") + self.project_root = self._unit("unit-project", "Project root") + self.other_tenant = self._unit( + "unit-other-tenant", + "Other tenant", + tenant_id="tenant-2", + ) + self.employer_child = OrganizationRelation( + id="edge-employer-child", + tenant_id="tenant-1", + structure_id=self.employer.id, + relation_type_id=self.employer_parent.id, + source_unit_id=self.root.id, + target_unit_id=self.child.id, + is_active=True, + settings={}, + ) + self.employer_grandchild = OrganizationRelation( + id="edge-employer-grandchild", + tenant_id="tenant-1", + structure_id=self.employer.id, + relation_type_id=self.employer_parent.id, + source_unit_id=self.child.id, + target_unit_id=self.grandchild.id, + is_active=True, + settings={}, + ) + self.project_child = OrganizationRelation( + id="edge-project-child", + tenant_id="tenant-1", + structure_id=self.project.id, + relation_type_id=self.project_parent.id, + source_unit_id=self.project_root.id, + target_unit_id=self.child.id, + is_active=True, + settings={}, + ) + self.function = OrganizationFunction( + id="function-child-intake", + tenant_id="tenant-1", + function_type_id=self.function_type.id, + organization_unit_id=self.child.id, + slug="intake", + name="Child intake", + is_active=True, + settings={}, + ) + self.session.add_all( + [ + self.unit_type, + self.function_type, + self.employer, + self.project, + self.employer_parent, + self.project_parent, + self.root, + self.child, + self.grandchild, + self.project_root, + self.other_tenant, + self.employer_child, + self.employer_grandchild, + self.project_child, + self.function, + ] + ) + self.session.commit() + + def _unit( + self, + item_id: str, + name: str, + *, + tenant_id: str = "tenant-1", + ) -> OrganizationUnit: + return OrganizationUnit( + id=item_id, + tenant_id=tenant_id, + unit_type_id=( + self.unit_type.id if tenant_id == "tenant-1" else None + ), + slug=item_id, + name=name, + is_active=True, + settings={}, + ) + + def test_parallel_structures_preserve_distinct_ancestor_provenance( + self, + ) -> None: + employer = self.directory.resolve_hierarchy_relatives( + "tenant-1", + (self.child.id,), + structure_id=self.employer.id, + direction="ancestors", + )[0] + project = self.directory.resolve_hierarchy_relatives( + "tenant-1", + (self.child.id,), + structure_id=self.project.id, + direction="ancestors", + )[0] + + self.assertEqual([self.root.id], [item.unit.id for item in employer.matches]) + self.assertEqual( + [self.project_root.id], + [item.unit.id for item in project.matches], + ) + self.assertEqual( + self.employer.id, + employer.matches[0].path[0].structure.id, + ) + self.assertEqual( + self.project_parent.id, + project.matches[0].path[0].relation_type.id, + ) + + def test_bounded_paths_report_depth_and_cycles(self) -> None: + bounded = self.directory.resolve_hierarchy_relatives( + "tenant-1", + (self.root.id,), + structure_id=self.employer.id, + direction="descendants", + max_depth=1, + )[0] + path = self.directory.resolve_hierarchy_paths( + "tenant-1", + ((self.child.id, self.root.id),), + structure_id=self.employer.id, + direction="ancestors", + )[0] + + self.assertEqual([self.child.id], [item.unit.id for item in bounded.matches]) + self.assertTrue(bounded.depth_limited) + self.assertEqual("active", path.status) + self.assertEqual( + [self.employer_child.id], + [edge.id for edge in path.path], + ) + + self.employer_parent.allow_cycles = True + self.session.add( + OrganizationRelation( + id="edge-cycle", + tenant_id="tenant-1", + structure_id=self.employer.id, + relation_type_id=self.employer_parent.id, + source_unit_id=self.grandchild.id, + target_unit_id=self.root.id, + is_active=True, + settings={}, + ) + ) + self.session.commit() + cycle = self.directory.resolve_hierarchy_relatives( + "tenant-1", + (self.root.id,), + structure_id=self.employer.id, + direction="descendants", + )[0] + self.assertTrue(cycle.cycle_detected) + self.assertTrue( + any(item.startswith("cycle_detected:") for item in cycle.diagnostics) + ) + + def test_function_and_unit_type_resolution_explains_missing_state( + self, + ) -> None: + functions = self.directory.resolve_functions_by_type( + "tenant-1", + self.function_type.id, + organization_unit_ids=(self.child.id, "missing-unit"), + ) + missing_type = self.directory.resolve_functions_by_type( + "tenant-1", + "missing-type", + organization_unit_ids=(self.child.id,), + ) + units = self.directory.resolve_units_by_type( + "tenant-1", + self.unit_type.id, + structure_id=self.employer.id, + root_unit_id=self.root.id, + direction="descendants", + ) + + self.assertEqual([self.function.id], [item.id for item in functions.matches]) + self.assertEqual(("missing-unit",), functions.missing_unit_ids) + self.assertEqual("missing", missing_type.status) + self.assertEqual( + {self.root.id, self.child.id, self.grandchild.id}, + {item.id for item in units.matches}, + ) + + def test_moved_deactivated_and_cross_tenant_units_are_current( + self, + ) -> None: + self.employer_child.source_unit_id = self.project_root.id + self.child.is_active = False + self.session.commit() + + moved = self.directory.resolve_hierarchy_relatives( + "tenant-1", + (self.child.id,), + structure_id=self.employer.id, + direction="ancestors", + )[0] + isolated = self.directory.resolve_hierarchy_relatives( + "tenant-1", + (self.other_tenant.id,), + structure_id=self.employer.id, + direction="ancestors", + )[0] + functions = self.directory.resolve_functions_by_type( + "tenant-1", + self.function_type.id, + organization_unit_ids=(self.child.id,), + ) + + self.assertEqual("inactive", moved.status) + self.assertEqual( + [self.project_root.id], + [item.unit.id for item in moved.matches], + ) + self.assertEqual("missing", isolated.status) + self.assertEqual((self.child.id,), functions.inactive_unit_ids) + + +class OrganizationLifecycleEventTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(self.engine) + self.Session = sessionmaker( + bind=self.engine, + expire_on_commit=False, + ) + self.session: Session = self.Session() + + def tearDown(self) -> None: + self.session.close() + Base.metadata.drop_all(self.engine) + self.engine.dispose() + + def test_changes_emit_versioned_commit_coupled_events(self) -> None: + events = [] + bus = EventBus() + bus.subscribe("*", events.append) + with event_bus_context(bus): + unit_type = OrganizationUnitType( + tenant_id="tenant-1", + slug="department", + name="Department", + is_active=True, + settings={}, + ) + self.session.add(unit_type) + _commit(self.session, unit_type) + unit_type.name = "Office" + _commit(self.session, unit_type) + unit_type.is_active = False + _commit(self.session, unit_type) + + organization_events = [ + event + for event in events + if event.module_id == "organizations" + ] + self.assertEqual( + [ + "organizations.unit_type.created.v1", + "organizations.unit_type.updated.v1", + "organizations.unit_type.deactivated.v1", + ], + [event.type for event in organization_events], + ) + self.assertTrue(unit_type.id) + self.assertEqual( + 1, + organization_events[0].payload["schema_version"], + ) + self.assertEqual( + unit_type.id, + organization_events[0].payload["resource_id"], + ) + self.assertIn( + "name", + organization_events[1].payload["changed_fields"], + ) + self.assertEqual( + "inactive", + organization_events[2].payload["status"], + ) + + +if __name__ == "__main__": + unittest.main()