add-ideas/govoplan-organizations
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

Add organizations module surface

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-10 17:33:53 +02:00
parent 34133ba9cd
commit ad2561b50f
22 changed files with 3284 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/govoplan_organizations/backend/api/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Organizations API package."""

1
src/govoplan_organizations/backend/api/v1/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Organizations API v1 package."""

720
src/govoplan_organizations/backend/api/v1/routes.py Normal file
View File

@@ -0,0 +1,720 @@
from __future__ import annotations
import re
from typing import Any, TypeVar
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal, require_any_scope
from govoplan_core.db.session import get_session
from govoplan_organizations.backend.db.models import (
OrganizationFunction,
OrganizationFunctionAssignment,
OrganizationFunctionType,
OrganizationRelation,
OrganizationRelationType,
OrganizationStructure,
OrganizationUnit,
OrganizationUnitType,
)
from .schemas import (
FunctionAssignmentCreateRequest,
FunctionAssignmentUpdateRequest,
FunctionCreateRequest,
FunctionTypeCreateRequest,
FunctionTypeUpdateRequest,
FunctionUpdateRequest,
OrganizationFunctionAssignmentItem,
OrganizationFunctionItem,
OrganizationFunctionTypeItem,
OrganizationModelResponse,
OrganizationRelationItem,
OrganizationRelationTypeItem,
OrganizationStructureItem,
OrganizationUnitItem,
OrganizationUnitTypeItem,
RelationCreateRequest,
RelationTypeCreateRequest,
RelationTypeUpdateRequest,
RelationUpdateRequest,
StructureCreateRequest,
StructureUpdateRequest,
UnitCreateRequest,
UnitTypeCreateRequest,
UnitTypeUpdateRequest,
UnitUpdateRequest,
)
router = APIRouter(prefix="/organizations", tags=["organizations"])
ORG_READ_SCOPES = (
"organizations:model:read",
"organizations:unit:read",
"organizations:function:read",
"admin:settings:read",
)
ORG_MODEL_WRITE_SCOPES = ("organizations:model:write",)
ORG_UNIT_WRITE_SCOPES = ("organizations:unit:write",)
ORG_FUNCTION_WRITE_SCOPES = ("organizations:function:write",)
ORG_ASSIGN_SCOPES = ("organizations:function:assign",)
SLUG_RE = re.compile(r"[^a-z0-9]+")
ModelT = TypeVar("ModelT")
def _slug(value: str | None, fallback: str) -> str:
source = value or fallback
normalized = SLUG_RE.sub("-", source.strip().casefold()).strip("-")
return normalized[:100] or "item"
def _tenant_id(principal: ApiPrincipal) -> str:
return principal.tenant_id
def _not_found(label: str) -> HTTPException:
return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"{label} not found")
def _conflict(message: str) -> HTTPException:
return HTTPException(status_code=status.HTTP_409_CONFLICT, detail=message)
def _invalid(message: str) -> HTTPException:
return HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=message)
def _get_tenant_row(session: Session, model: type[ModelT], item_id: str, tenant_id: str, label: str) -> ModelT:
item = session.get(model, item_id)
if item is None or getattr(item, "tenant_id") != tenant_id:
raise _not_found(label)
return item
def _ensure_optional_unit_type(session: Session, tenant_id: str, unit_type_id: str | None) -> None:
if unit_type_id is None:
return
_get_tenant_row(session, OrganizationUnitType, unit_type_id, tenant_id, "Organization unit type")
def _ensure_optional_structure(session: Session, tenant_id: str, structure_id: str | None) -> None:
if structure_id is None:
return
_get_tenant_row(session, OrganizationStructure, structure_id, tenant_id, "Organization structure")
def _ensure_unique_slug(session: Session, model: type, tenant_id: str, slug: str, *, exclude_id: str | None = None) -> None:
query = session.query(model).filter(model.tenant_id == tenant_id, model.slug == slug)
if exclude_id is not None:
query = query.filter(model.id != exclude_id)
if query.count():
raise _conflict(f"Slug already exists in this tenant: {slug}")
def _commit(session: Session, item: ModelT) -> ModelT:
try:
session.commit()
except IntegrityError as exc:
session.rollback()
raise _conflict("The organization model change conflicts with existing data.") from exc
session.refresh(item)
return item
def _item_unit_type(item: OrganizationUnitType) -> OrganizationUnitTypeItem:
return OrganizationUnitTypeItem(**_row_fields(item))
def _item_structure(item: OrganizationStructure) -> OrganizationStructureItem:
return OrganizationStructureItem(**_row_fields(item))
def _item_relation_type(item: OrganizationRelationType) -> OrganizationRelationTypeItem:
return OrganizationRelationTypeItem(**_row_fields(item))
def _item_unit(item: OrganizationUnit) -> OrganizationUnitItem:
return OrganizationUnitItem(**_row_fields(item))
def _item_relation(item: OrganizationRelation) -> OrganizationRelationItem:
return OrganizationRelationItem(**_row_fields(item))
def _item_function_type(item: OrganizationFunctionType) -> OrganizationFunctionTypeItem:
return OrganizationFunctionTypeItem(**_row_fields(item))
def _item_function(item: OrganizationFunction) -> OrganizationFunctionItem:
return OrganizationFunctionItem(**_row_fields(item))
def _item_assignment(item: OrganizationFunctionAssignment) -> OrganizationFunctionAssignmentItem:
return OrganizationFunctionAssignmentItem(**_row_fields(item))
def _row_fields(item: object) -> dict[str, Any]:
keys = [column.name for column in item.__table__.columns] # type: ignore[attr-defined]
return {key: getattr(item, key) for key in keys}
def _apply_slugged_update(session: Session, item: object, payload: object, tenant_id: str, model: type) -> None:
fields = payload.model_fields_set # type: ignore[attr-defined]
if "slug" in fields:
slug = _slug(getattr(payload, "slug"), getattr(item, "name", "item"))
_ensure_unique_slug(session, model, tenant_id, slug, exclude_id=getattr(item, "id"))
setattr(item, "slug", slug)
if "name" in fields:
value = getattr(payload, "name")
if value is None:
raise _invalid("Name cannot be empty.")
setattr(item, "name", value.strip())
if "description" in fields:
setattr(item, "description", getattr(payload, "description"))
if "is_active" in fields:
value = getattr(payload, "is_active")
if value is None:
raise _invalid("Active state cannot be empty.")
setattr(item, "is_active", value)
if "settings" in fields:
value = getattr(payload, "settings")
if value is None:
raise _invalid("Settings cannot be empty.")
setattr(item, "settings", value)
@router.get("/model", response_model=OrganizationModelResponse)
def get_organization_model(
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_READ_SCOPES)),
) -> OrganizationModelResponse:
tenant_id = _tenant_id(principal)
return OrganizationModelResponse(
unit_types=[_item_unit_type(item) for item in session.query(OrganizationUnitType).filter(OrganizationUnitType.tenant_id == tenant_id).order_by(OrganizationUnitType.name.asc()).all()],
structures=[_item_structure(item) for item in session.query(OrganizationStructure).filter(OrganizationStructure.tenant_id == tenant_id).order_by(OrganizationStructure.name.asc()).all()],
relation_types=[_item_relation_type(item) for item in session.query(OrganizationRelationType).filter(OrganizationRelationType.tenant_id == tenant_id).order_by(OrganizationRelationType.name.asc()).all()],
units=[_item_unit(item) for item in session.query(OrganizationUnit).filter(OrganizationUnit.tenant_id == tenant_id).order_by(OrganizationUnit.name.asc()).all()],
relations=[_item_relation(item) for item in session.query(OrganizationRelation).filter(OrganizationRelation.tenant_id == tenant_id).order_by(OrganizationRelation.created_at.asc()).all()],
function_types=[_item_function_type(item) for item in session.query(OrganizationFunctionType).filter(OrganizationFunctionType.tenant_id == tenant_id).order_by(OrganizationFunctionType.name.asc()).all()],
functions=[_item_function(item) for item in session.query(OrganizationFunction).filter(OrganizationFunction.tenant_id == tenant_id).order_by(OrganizationFunction.name.asc()).all()],
function_assignments=[_item_assignment(item) for item in session.query(OrganizationFunctionAssignment).filter(OrganizationFunctionAssignment.tenant_id == tenant_id).order_by(OrganizationFunctionAssignment.created_at.asc()).all()],
)
@router.post("/unit-types", response_model=OrganizationUnitTypeItem, status_code=status.HTTP_201_CREATED)
def create_unit_type(
payload: UnitTypeCreateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_MODEL_WRITE_SCOPES)),
) -> OrganizationUnitTypeItem:
tenant_id = _tenant_id(principal)
slug = _slug(payload.slug, payload.name)
_ensure_unique_slug(session, OrganizationUnitType, tenant_id, slug)
item = OrganizationUnitType(tenant_id=tenant_id, slug=slug, name=payload.name.strip(), description=payload.description, is_active=payload.is_active, settings=payload.settings)
session.add(item)
return _item_unit_type(_commit(session, item))
@router.patch("/unit-types/{item_id}", response_model=OrganizationUnitTypeItem)
def update_unit_type(
item_id: str,
payload: UnitTypeUpdateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_MODEL_WRITE_SCOPES)),
) -> OrganizationUnitTypeItem:
tenant_id = _tenant_id(principal)
item = _get_tenant_row(session, OrganizationUnitType, item_id, tenant_id, "Organization unit type")
_apply_slugged_update(session, item, payload, tenant_id, OrganizationUnitType)
return _item_unit_type(_commit(session, item))
@router.post("/structures", response_model=OrganizationStructureItem, status_code=status.HTTP_201_CREATED)
def create_structure(
payload: StructureCreateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_MODEL_WRITE_SCOPES)),
) -> OrganizationStructureItem:
tenant_id = _tenant_id(principal)
slug = _slug(payload.slug, payload.name)
_ensure_unique_slug(session, OrganizationStructure, tenant_id, slug)
item = OrganizationStructure(tenant_id=tenant_id, slug=slug, name=payload.name.strip(), description=payload.description, structure_kind=payload.structure_kind, is_active=payload.is_active, settings=payload.settings)
session.add(item)
return _item_structure(_commit(session, item))
@router.patch("/structures/{item_id}", response_model=OrganizationStructureItem)
def update_structure(
item_id: str,
payload: StructureUpdateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_MODEL_WRITE_SCOPES)),
) -> OrganizationStructureItem:
tenant_id = _tenant_id(principal)
item = _get_tenant_row(session, OrganizationStructure, item_id, tenant_id, "Organization structure")
_apply_slugged_update(session, item, payload, tenant_id, OrganizationStructure)
if "structure_kind" in payload.model_fields_set:
if payload.structure_kind is None:
raise _invalid("Structure kind cannot be empty.")
item.structure_kind = payload.structure_kind
return _item_structure(_commit(session, item))
@router.post("/relation-types", response_model=OrganizationRelationTypeItem, status_code=status.HTTP_201_CREATED)
def create_relation_type(
payload: RelationTypeCreateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_MODEL_WRITE_SCOPES)),
) -> OrganizationRelationTypeItem:
tenant_id = _tenant_id(principal)
_ensure_optional_structure(session, tenant_id, payload.structure_id)
_ensure_optional_unit_type(session, tenant_id, payload.source_unit_type_id)
_ensure_optional_unit_type(session, tenant_id, payload.target_unit_type_id)
slug = _slug(payload.slug, payload.name)
_ensure_unique_slug(session, OrganizationRelationType, tenant_id, slug)
item = OrganizationRelationType(
tenant_id=tenant_id,
structure_id=payload.structure_id,
slug=slug,
name=payload.name.strip(),
description=payload.description,
source_unit_type_id=payload.source_unit_type_id,
target_unit_type_id=payload.target_unit_type_id,
is_hierarchical=payload.is_hierarchical,
allow_cycles=payload.allow_cycles,
is_active=payload.is_active,
settings=payload.settings,
)
session.add(item)
return _item_relation_type(_commit(session, item))
@router.patch("/relation-types/{item_id}", response_model=OrganizationRelationTypeItem)
def update_relation_type(
item_id: str,
payload: RelationTypeUpdateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_MODEL_WRITE_SCOPES)),
) -> OrganizationRelationTypeItem:
tenant_id = _tenant_id(principal)
item = _get_tenant_row(session, OrganizationRelationType, item_id, tenant_id, "Organization relation type")
_apply_slugged_update(session, item, payload, tenant_id, OrganizationRelationType)
fields = payload.model_fields_set
if "structure_id" in fields:
_ensure_optional_structure(session, tenant_id, payload.structure_id)
item.structure_id = payload.structure_id
if "source_unit_type_id" in fields:
_ensure_optional_unit_type(session, tenant_id, payload.source_unit_type_id)
item.source_unit_type_id = payload.source_unit_type_id
if "target_unit_type_id" in fields:
_ensure_optional_unit_type(session, tenant_id, payload.target_unit_type_id)
item.target_unit_type_id = payload.target_unit_type_id
for field in ("is_hierarchical", "allow_cycles"):
if field in fields:
value = getattr(payload, field)
if value is None:
raise _invalid(f"{field} cannot be empty.")
setattr(item, field, value)
return _item_relation_type(_commit(session, item))
@router.post("/units", response_model=OrganizationUnitItem, status_code=status.HTTP_201_CREATED)
def create_unit(
payload: UnitCreateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_UNIT_WRITE_SCOPES)),
) -> OrganizationUnitItem:
tenant_id = _tenant_id(principal)
_ensure_optional_unit_type(session, tenant_id, payload.unit_type_id)
if payload.parent_id is not None:
_get_tenant_row(session, OrganizationUnit, payload.parent_id, tenant_id, "Parent organization unit")
slug = _slug(payload.slug, payload.name)
_ensure_unique_slug(session, OrganizationUnit, tenant_id, slug)
item = OrganizationUnit(tenant_id=tenant_id, unit_type_id=payload.unit_type_id, parent_id=payload.parent_id, slug=slug, name=payload.name.strip(), description=payload.description, is_active=payload.is_active, settings=payload.settings)
session.add(item)
return _item_unit(_commit(session, item))
@router.patch("/units/{item_id}", response_model=OrganizationUnitItem)
def update_unit(
item_id: str,
payload: UnitUpdateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_UNIT_WRITE_SCOPES)),
) -> OrganizationUnitItem:
tenant_id = _tenant_id(principal)
item = _get_tenant_row(session, OrganizationUnit, item_id, tenant_id, "Organization unit")
_apply_slugged_update(session, item, payload, tenant_id, OrganizationUnit)
if "unit_type_id" in payload.model_fields_set:
_ensure_optional_unit_type(session, tenant_id, payload.unit_type_id)
item.unit_type_id = payload.unit_type_id
if "parent_id" in payload.model_fields_set:
if payload.parent_id == item.id:
raise _invalid("An organization unit cannot be its own parent.")
if payload.parent_id is not None:
_get_tenant_row(session, OrganizationUnit, payload.parent_id, tenant_id, "Parent organization unit")
item.parent_id = payload.parent_id
return _item_unit(_commit(session, item))
@router.post("/relations", response_model=OrganizationRelationItem, status_code=status.HTTP_201_CREATED)
def create_relation(
payload: RelationCreateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_UNIT_WRITE_SCOPES)),
) -> OrganizationRelationItem:
tenant_id = _tenant_id(principal)
structure, relation_type, source, target = _validated_relation_parts(
session,
tenant_id,
structure_id=payload.structure_id,
relation_type_id=payload.relation_type_id,
source_unit_id=payload.source_unit_id,
target_unit_id=payload.target_unit_id,
)
_validate_relation_edge(session, tenant_id, structure, relation_type, source, target)
item = OrganizationRelation(
tenant_id=tenant_id,
structure_id=structure.id,
relation_type_id=relation_type.id,
source_unit_id=source.id,
target_unit_id=target.id,
valid_from=payload.valid_from,
valid_until=payload.valid_until,
is_active=payload.is_active,
settings=payload.settings,
)
session.add(item)
return _item_relation(_commit(session, item))
@router.patch("/relations/{item_id}", response_model=OrganizationRelationItem)
def update_relation(
item_id: str,
payload: RelationUpdateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_UNIT_WRITE_SCOPES)),
) -> OrganizationRelationItem:
tenant_id = _tenant_id(principal)
item = _get_tenant_row(session, OrganizationRelation, item_id, tenant_id, "Organization relation")
structure_id = payload.structure_id if "structure_id" in payload.model_fields_set else item.structure_id
relation_type_id = payload.relation_type_id if "relation_type_id" in payload.model_fields_set else item.relation_type_id
source_unit_id = payload.source_unit_id if "source_unit_id" in payload.model_fields_set else item.source_unit_id
target_unit_id = payload.target_unit_id if "target_unit_id" in payload.model_fields_set else item.target_unit_id
if structure_id is None or relation_type_id is None or source_unit_id is None or target_unit_id is None:
raise _invalid("Relation structure, type, source, and target are required.")
structure, relation_type, source, target = _validated_relation_parts(
session,
tenant_id,
structure_id=structure_id,
relation_type_id=relation_type_id,
source_unit_id=source_unit_id,
target_unit_id=target_unit_id,
)
_validate_relation_edge(session, tenant_id, structure, relation_type, source, target, exclude_relation_id=item.id)
item.structure_id = structure.id
item.relation_type_id = relation_type.id
item.source_unit_id = source.id
item.target_unit_id = target.id
for field in ("valid_from", "valid_until", "is_active", "settings"):
if field in payload.model_fields_set:
setattr(item, field, getattr(payload, field))
return _item_relation(_commit(session, item))
@router.post("/function-types", response_model=OrganizationFunctionTypeItem, status_code=status.HTTP_201_CREATED)
def create_function_type(
payload: FunctionTypeCreateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_MODEL_WRITE_SCOPES)),
) -> OrganizationFunctionTypeItem:
tenant_id = _tenant_id(principal)
_ensure_optional_unit_type(session, tenant_id, payload.organization_unit_type_id)
slug = _slug(payload.slug, payload.name)
_ensure_unique_slug(session, OrganizationFunctionType, tenant_id, slug)
item = OrganizationFunctionType(
tenant_id=tenant_id,
slug=slug,
name=payload.name.strip(),
description=payload.description,
organization_unit_type_id=payload.organization_unit_type_id,
delegable=payload.delegable,
act_in_place_allowed=payload.act_in_place_allowed,
is_active=payload.is_active,
settings=payload.settings,
)
session.add(item)
return _item_function_type(_commit(session, item))
@router.patch("/function-types/{item_id}", response_model=OrganizationFunctionTypeItem)
def update_function_type(
item_id: str,
payload: FunctionTypeUpdateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_MODEL_WRITE_SCOPES)),
) -> OrganizationFunctionTypeItem:
tenant_id = _tenant_id(principal)
item = _get_tenant_row(session, OrganizationFunctionType, item_id, tenant_id, "Organization function type")
_apply_slugged_update(session, item, payload, tenant_id, OrganizationFunctionType)
if "organization_unit_type_id" in payload.model_fields_set:
_ensure_optional_unit_type(session, tenant_id, payload.organization_unit_type_id)
item.organization_unit_type_id = payload.organization_unit_type_id
for field in ("delegable", "act_in_place_allowed"):
if field in payload.model_fields_set:
value = getattr(payload, field)
if value is None:
raise _invalid(f"{field} cannot be empty.")
setattr(item, field, value)
return _item_function_type(_commit(session, item))
@router.post("/functions", response_model=OrganizationFunctionItem, status_code=status.HTTP_201_CREATED)
def create_function(
payload: FunctionCreateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_FUNCTION_WRITE_SCOPES)),
) -> OrganizationFunctionItem:
tenant_id = _tenant_id(principal)
unit = _get_tenant_row(session, OrganizationUnit, payload.organization_unit_id, tenant_id, "Organization unit")
function_type = _optional_function_type(session, tenant_id, payload.function_type_id)
slug = _slug(payload.slug, payload.name)
_ensure_function_slug(session, tenant_id, unit.id, slug)
delegable = payload.delegable if payload.delegable is not None else (function_type.delegable if function_type else False)
act_in_place_allowed = payload.act_in_place_allowed if payload.act_in_place_allowed is not None else (function_type.act_in_place_allowed if function_type else False)
item = OrganizationFunction(
tenant_id=tenant_id,
function_type_id=payload.function_type_id,
organization_unit_id=unit.id,
slug=slug,
name=payload.name.strip(),
description=payload.description,
delegable=delegable,
act_in_place_allowed=act_in_place_allowed,
is_active=payload.is_active,
settings=payload.settings,
)
session.add(item)
return _item_function(_commit(session, item))
@router.patch("/functions/{item_id}", response_model=OrganizationFunctionItem)
def update_function(
item_id: str,
payload: FunctionUpdateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_FUNCTION_WRITE_SCOPES)),
) -> OrganizationFunctionItem:
tenant_id = _tenant_id(principal)
item = _get_tenant_row(session, OrganizationFunction, item_id, tenant_id, "Organization function")
_apply_slugged_update_for_function(session, item, payload, tenant_id)
if "organization_unit_id" in payload.model_fields_set:
if payload.organization_unit_id is None:
raise _invalid("Organization unit is required.")
unit = _get_tenant_row(session, OrganizationUnit, payload.organization_unit_id, tenant_id, "Organization unit")
_ensure_function_slug(session, tenant_id, unit.id, item.slug, exclude_id=item.id)
item.organization_unit_id = unit.id
if "function_type_id" in payload.model_fields_set:
_optional_function_type(session, tenant_id, payload.function_type_id)
item.function_type_id = payload.function_type_id
for field in ("delegable", "act_in_place_allowed"):
if field in payload.model_fields_set:
value = getattr(payload, field)
if value is None:
raise _invalid(f"{field} cannot be empty.")
setattr(item, field, value)
return _item_function(_commit(session, item))
@router.post("/function-assignments", response_model=OrganizationFunctionAssignmentItem, status_code=status.HTTP_201_CREATED)
def create_function_assignment(
payload: FunctionAssignmentCreateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_ASSIGN_SCOPES)),
) -> OrganizationFunctionAssignmentItem:
tenant_id = _tenant_id(principal)
function = _get_tenant_row(session, OrganizationFunction, payload.function_id, tenant_id, "Organization function")
item = OrganizationFunctionAssignment(
tenant_id=tenant_id,
identity_id=payload.identity_id,
account_id=payload.account_id,
function_id=function.id,
organization_unit_id=function.organization_unit_id,
applies_to_subunits=payload.applies_to_subunits,
source=payload.source,
delegated_from_assignment_id=payload.delegated_from_assignment_id,
acting_for_account_id=payload.acting_for_account_id,
valid_from=payload.valid_from,
valid_until=payload.valid_until,
is_active=payload.is_active,
settings=payload.settings,
)
if item.delegated_from_assignment_id is not None:
_get_tenant_row(session, OrganizationFunctionAssignment, item.delegated_from_assignment_id, tenant_id, "Delegated function assignment")
session.add(item)
return _item_assignment(_commit(session, item))
@router.patch("/function-assignments/{item_id}", response_model=OrganizationFunctionAssignmentItem)
def update_function_assignment(
item_id: str,
payload: FunctionAssignmentUpdateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope(*ORG_ASSIGN_SCOPES)),
) -> OrganizationFunctionAssignmentItem:
tenant_id = _tenant_id(principal)
item = _get_tenant_row(session, OrganizationFunctionAssignment, item_id, tenant_id, "Organization function assignment")
if "function_id" in payload.model_fields_set:
if payload.function_id is None:
raise _invalid("Function is required.")
function = _get_tenant_row(session, OrganizationFunction, payload.function_id, tenant_id, "Organization function")
item.function_id = function.id
item.organization_unit_id = function.organization_unit_id
if "identity_id" in payload.model_fields_set and payload.identity_id is None:
raise _invalid("Identity is required.")
if "source" in payload.model_fields_set and payload.source is None:
raise _invalid("Assignment source is required.")
if "applies_to_subunits" in payload.model_fields_set and payload.applies_to_subunits is None:
raise _invalid("Subunit applicability cannot be empty.")
if "is_active" in payload.model_fields_set and payload.is_active is None:
raise _invalid("Active state cannot be empty.")
if "settings" in payload.model_fields_set and payload.settings is None:
raise _invalid("Settings cannot be empty.")
for field in (
"identity_id",
"account_id",
"applies_to_subunits",
"source",
"delegated_from_assignment_id",
"acting_for_account_id",
"valid_from",
"valid_until",
"is_active",
"settings",
):
if field in payload.model_fields_set:
setattr(item, field, getattr(payload, field))
if item.delegated_from_assignment_id == item.id:
raise _invalid("A function assignment cannot delegate from itself.")
if item.delegated_from_assignment_id is not None:
_get_tenant_row(session, OrganizationFunctionAssignment, item.delegated_from_assignment_id, tenant_id, "Delegated function assignment")
return _item_assignment(_commit(session, item))
def _optional_function_type(session: Session, tenant_id: str, function_type_id: str | None) -> OrganizationFunctionType | None:
if function_type_id is None:
return None
return _get_tenant_row(session, OrganizationFunctionType, function_type_id, tenant_id, "Organization function type")
def _ensure_function_slug(session: Session, tenant_id: str, organization_unit_id: str, slug: str, *, exclude_id: str | None = None) -> None:
query = session.query(OrganizationFunction).filter(
OrganizationFunction.tenant_id == tenant_id,
OrganizationFunction.organization_unit_id == organization_unit_id,
OrganizationFunction.slug == slug,
)
if exclude_id is not None:
query = query.filter(OrganizationFunction.id != exclude_id)
if query.count():
raise _conflict(f"Function slug already exists in this organization unit: {slug}")
def _apply_slugged_update_for_function(session: Session, item: OrganizationFunction, payload: FunctionUpdateRequest, tenant_id: str) -> None:
fields = payload.model_fields_set
if "slug" in fields:
slug = _slug(payload.slug, item.name)
_ensure_function_slug(session, tenant_id, item.organization_unit_id, slug, exclude_id=item.id)
item.slug = slug
if "name" in fields:
if payload.name is None:
raise _invalid("Name cannot be empty.")
item.name = payload.name.strip()
if "description" in fields:
item.description = payload.description
if "is_active" in fields:
if payload.is_active is None:
raise _invalid("Active state cannot be empty.")
item.is_active = payload.is_active
if "settings" in fields:
if payload.settings is None:
raise _invalid("Settings cannot be empty.")
item.settings = payload.settings
def _validated_relation_parts(
session: Session,
tenant_id: str,
*,
structure_id: str,
relation_type_id: str,
source_unit_id: str,
target_unit_id: str,
) -> tuple[OrganizationStructure, OrganizationRelationType, OrganizationUnit, OrganizationUnit]:
structure = _get_tenant_row(session, OrganizationStructure, structure_id, tenant_id, "Organization structure")
relation_type = _get_tenant_row(session, OrganizationRelationType, relation_type_id, tenant_id, "Organization relation type")
source = _get_tenant_row(session, OrganizationUnit, source_unit_id, tenant_id, "Source organization unit")
target = _get_tenant_row(session, OrganizationUnit, target_unit_id, tenant_id, "Target organization unit")
if relation_type.structure_id is not None and relation_type.structure_id != structure.id:
raise _invalid("Relation type is bound to another structure.")
if source.id == target.id:
raise _invalid("A relation cannot connect an organization unit to itself.")
if relation_type.source_unit_type_id is not None and source.unit_type_id != relation_type.source_unit_type_id:
raise _invalid("Source organization unit type is not valid for this relation type.")
if relation_type.target_unit_type_id is not None and target.unit_type_id != relation_type.target_unit_type_id:
raise _invalid("Target organization unit type is not valid for this relation type.")
return structure, relation_type, source, target
def _validate_relation_edge(
session: Session,
tenant_id: str,
structure: OrganizationStructure,
relation_type: OrganizationRelationType,
source: OrganizationUnit,
target: OrganizationUnit,
*,
exclude_relation_id: str | None = None,
) -> None:
if relation_type.is_hierarchical and not relation_type.allow_cycles:
if _would_create_cycle(
session,
tenant_id=tenant_id,
structure_id=structure.id,
source_unit_id=source.id,
target_unit_id=target.id,
exclude_relation_id=exclude_relation_id,
):
raise _invalid("This hierarchical relation would create a cycle.")
def _would_create_cycle(
session: Session,
*,
tenant_id: str,
structure_id: str,
source_unit_id: str,
target_unit_id: str,
exclude_relation_id: str | None = None,
) -> bool:
pending = [target_unit_id]
seen: set[str] = set()
while pending:
current = pending.pop()
if current == source_unit_id:
return True
if current in seen:
continue
seen.add(current)
query = session.query(OrganizationRelation.target_unit_id).filter(
OrganizationRelation.tenant_id == tenant_id,
OrganizationRelation.structure_id == structure_id,
OrganizationRelation.source_unit_id == current,
OrganizationRelation.is_active.is_(True),
)
if exclude_relation_id is not None:
query = query.filter(OrganizationRelation.id != exclude_relation_id)
pending.extend(row[0] for row in query.all())
return False

275
src/govoplan_organizations/backend/api/v1/schemas.py Normal file
View File

@@ -0,0 +1,275 @@
from __future__ import annotations
from datetime import datetime
from typing import Any, Literal
from pydantic import BaseModel, Field
StructureKind = Literal["hierarchy", "network", "membership", "classification"]
class OrganizationUnitTypeItem(BaseModel):
id: str
tenant_id: str
slug: str
name: str
description: str | None = None
is_active: bool
settings: dict[str, Any]
created_at: datetime
updated_at: datetime
class OrganizationStructureItem(BaseModel):
id: str
tenant_id: str
slug: str
name: str
description: str | None = None
structure_kind: str
is_active: bool
settings: dict[str, Any]
created_at: datetime
updated_at: datetime
class OrganizationRelationTypeItem(BaseModel):
id: str
tenant_id: str
structure_id: str | None = None
slug: str
name: str
description: str | None = None
source_unit_type_id: str | None = None
target_unit_type_id: str | None = None
is_hierarchical: bool
allow_cycles: bool
is_active: bool
settings: dict[str, Any]
created_at: datetime
updated_at: datetime
class OrganizationUnitItem(BaseModel):
id: str
tenant_id: str
unit_type_id: str | None = None
parent_id: str | None = None
slug: str
name: str
description: str | None = None
is_active: bool
settings: dict[str, Any]
created_at: datetime
updated_at: datetime
class OrganizationRelationItem(BaseModel):
id: str
tenant_id: str
structure_id: str
relation_type_id: str
source_unit_id: str
target_unit_id: str
valid_from: datetime | None = None
valid_until: datetime | None = None
is_active: bool
settings: dict[str, Any]
created_at: datetime
updated_at: datetime
class OrganizationFunctionTypeItem(BaseModel):
id: str
tenant_id: str
slug: str
name: str
description: str | None = None
organization_unit_type_id: str | None = None
delegable: bool
act_in_place_allowed: bool
is_active: bool
settings: dict[str, Any]
created_at: datetime
updated_at: datetime
class OrganizationFunctionItem(BaseModel):
id: str
tenant_id: str
function_type_id: str | None = None
organization_unit_id: str
slug: str
name: str
description: str | None = None
delegable: bool
act_in_place_allowed: bool
is_active: bool
settings: dict[str, Any]
created_at: datetime
updated_at: datetime
class OrganizationFunctionAssignmentItem(BaseModel):
id: str
tenant_id: str
identity_id: str
account_id: str | None = None
function_id: str
organization_unit_id: str
applies_to_subunits: bool
source: str
delegated_from_assignment_id: str | None = None
acting_for_account_id: str | None = None
valid_from: datetime | None = None
valid_until: datetime | None = None
is_active: bool
settings: dict[str, Any]
created_at: datetime
updated_at: datetime
class OrganizationModelResponse(BaseModel):
unit_types: list[OrganizationUnitTypeItem]
structures: list[OrganizationStructureItem]
relation_types: list[OrganizationRelationTypeItem]
units: list[OrganizationUnitItem]
relations: list[OrganizationRelationItem]
function_types: list[OrganizationFunctionTypeItem]
functions: list[OrganizationFunctionItem]
function_assignments: list[OrganizationFunctionAssignmentItem]
class SluggedCreateRequest(BaseModel):
slug: str | None = Field(default=None, max_length=100)
name: str = Field(min_length=1, max_length=255)
description: str | None = None
is_active: bool = True
settings: dict[str, Any] = Field(default_factory=dict)
class SluggedUpdateRequest(BaseModel):
slug: str | None = Field(default=None, max_length=100)
name: str | None = Field(default=None, max_length=255)
description: str | None = None
is_active: bool | None = None
settings: dict[str, Any] | None = None
class UnitTypeCreateRequest(SluggedCreateRequest):
pass
class UnitTypeUpdateRequest(SluggedUpdateRequest):
pass
class StructureCreateRequest(SluggedCreateRequest):
structure_kind: StructureKind = "hierarchy"
class StructureUpdateRequest(SluggedUpdateRequest):
structure_kind: StructureKind | None = None
class RelationTypeCreateRequest(SluggedCreateRequest):
structure_id: str | None = None
source_unit_type_id: str | None = None
target_unit_type_id: str | None = None
is_hierarchical: bool = True
allow_cycles: bool = False
class RelationTypeUpdateRequest(SluggedUpdateRequest):
structure_id: str | None = None
source_unit_type_id: str | None = None
target_unit_type_id: str | None = None
is_hierarchical: bool | None = None
allow_cycles: bool | None = None
class UnitCreateRequest(SluggedCreateRequest):
unit_type_id: str | None = None
parent_id: str | None = None
class UnitUpdateRequest(SluggedUpdateRequest):
unit_type_id: str | None = None
parent_id: str | None = None
class RelationCreateRequest(BaseModel):
structure_id: str
relation_type_id: str
source_unit_id: str
target_unit_id: str
valid_from: datetime | None = None
valid_until: datetime | None = None
is_active: bool = True
settings: dict[str, Any] = Field(default_factory=dict)
class RelationUpdateRequest(BaseModel):
structure_id: str | None = None
relation_type_id: str | None = None
source_unit_id: str | None = None
target_unit_id: str | None = None
valid_from: datetime | None = None
valid_until: datetime | None = None
is_active: bool | None = None
settings: dict[str, Any] | None = None
class FunctionTypeCreateRequest(SluggedCreateRequest):
organization_unit_type_id: str | None = None
delegable: bool = False
act_in_place_allowed: bool = False
class FunctionTypeUpdateRequest(SluggedUpdateRequest):
organization_unit_type_id: str | None = None
delegable: bool | None = None
act_in_place_allowed: bool | None = None
class FunctionCreateRequest(SluggedCreateRequest):
organization_unit_id: str
function_type_id: str | None = None
delegable: bool | None = None
act_in_place_allowed: bool | None = None
class FunctionUpdateRequest(SluggedUpdateRequest):
organization_unit_id: str | None = None
function_type_id: str | None = None
delegable: bool | None = None
act_in_place_allowed: bool | None = None
class FunctionAssignmentCreateRequest(BaseModel):
identity_id: str
function_id: str
account_id: str | None = None
applies_to_subunits: bool = False
source: str = "direct"
delegated_from_assignment_id: str | None = None
acting_for_account_id: str | None = None
valid_from: datetime | None = None
valid_until: datetime | None = None
is_active: bool = True
settings: dict[str, Any] = Field(default_factory=dict)
class FunctionAssignmentUpdateRequest(BaseModel):
identity_id: str | None = None
function_id: str | None = None
account_id: str | None = None
applies_to_subunits: bool | None = None
source: str | None = None
delegated_from_assignment_id: str | None = None
acting_for_account_id: str | None = None
valid_from: datetime | None = None
valid_until: datetime | None = None
is_active: bool | None = None
settings: dict[str, Any] | None = None

114
src/govoplan_organizations/backend/db/models.py
View File

@@ -4,7 +4,7 @@ import uuid
from datetime import datetime
from typing import Any
from sqlalchemy import Boolean, DateTime, ForeignKey, JSON, String, Text, UniqueConstraint
from sqlalchemy import Boolean, DateTime, ForeignKey, Index, JSON, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from govoplan_core.db.base import Base, TimestampMixin
@@ -19,7 +19,8 @@ class OrganizationUnit(Base, TimestampMixin):
__table_args__ = (UniqueConstraint("tenant_id", "slug", name="uq_organizations_units_tenant_slug"),)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(ForeignKey("tenancy_tenants.id", ondelete="CASCADE"), nullable=False, index=True)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
unit_type_id: Mapped[str | None] = mapped_column(ForeignKey("organizations_unit_types.id", ondelete="SET NULL"), nullable=True, index=True)
parent_id: Mapped[str | None] = mapped_column(ForeignKey("organizations_units.id", ondelete="SET NULL"), nullable=True, index=True)
slug: Mapped[str] = mapped_column(String(100), nullable=False)
name: Mapped[str] = mapped_column(String(255), nullable=False)
@@ -28,12 +29,104 @@ class OrganizationUnit(Base, TimestampMixin):
settings: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
class OrganizationUnitType(Base, TimestampMixin):
__tablename__ = "organizations_unit_types"
__table_args__ = (UniqueConstraint("tenant_id", "slug", name="uq_organizations_unit_types_tenant_slug"),)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
slug: Mapped[str] = mapped_column(String(100), nullable=False)
name: Mapped[str] = mapped_column(String(255), nullable=False)
description: Mapped[str | None] = mapped_column(Text)
is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
settings: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
class OrganizationStructure(Base, TimestampMixin):
__tablename__ = "organizations_structures"
__table_args__ = (UniqueConstraint("tenant_id", "slug", name="uq_organizations_structures_tenant_slug"),)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
slug: Mapped[str] = mapped_column(String(100), nullable=False)
name: Mapped[str] = mapped_column(String(255), nullable=False)
description: Mapped[str | None] = mapped_column(Text)
structure_kind: Mapped[str] = mapped_column(String(30), default="hierarchy", nullable=False)
is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
settings: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
class OrganizationRelationType(Base, TimestampMixin):
__tablename__ = "organizations_relation_types"
__table_args__ = (
UniqueConstraint("tenant_id", "slug", name="uq_organizations_relation_types_tenant_slug"),
Index("ix_organizations_relation_types_structure", "tenant_id", "structure_id"),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
structure_id: Mapped[str | None] = mapped_column(ForeignKey("organizations_structures.id", ondelete="SET NULL"), nullable=True, index=True)
slug: Mapped[str] = mapped_column(String(100), nullable=False)
name: Mapped[str] = mapped_column(String(255), nullable=False)
description: Mapped[str | None] = mapped_column(Text)
source_unit_type_id: Mapped[str | None] = mapped_column(ForeignKey("organizations_unit_types.id", ondelete="SET NULL"), nullable=True, index=True)
target_unit_type_id: Mapped[str | None] = mapped_column(ForeignKey("organizations_unit_types.id", ondelete="SET NULL"), nullable=True, index=True)
is_hierarchical: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
allow_cycles: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
settings: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
class OrganizationRelation(Base, TimestampMixin):
__tablename__ = "organizations_relations"
__table_args__ = (
UniqueConstraint(
"tenant_id",
"structure_id",
"relation_type_id",
"source_unit_id",
"target_unit_id",
name="uq_organizations_relations_edge",
),
Index("ix_organizations_relations_structure_source", "tenant_id", "structure_id", "source_unit_id"),
Index("ix_organizations_relations_structure_target", "tenant_id", "structure_id", "target_unit_id"),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
structure_id: Mapped[str] = mapped_column(ForeignKey("organizations_structures.id", ondelete="CASCADE"), nullable=False, index=True)
relation_type_id: Mapped[str] = mapped_column(ForeignKey("organizations_relation_types.id", ondelete="CASCADE"), nullable=False, index=True)
source_unit_id: Mapped[str] = mapped_column(ForeignKey("organizations_units.id", ondelete="CASCADE"), nullable=False, index=True)
target_unit_id: Mapped[str] = mapped_column(ForeignKey("organizations_units.id", ondelete="CASCADE"), nullable=False, index=True)
valid_from: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
valid_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
settings: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
class OrganizationFunctionType(Base, TimestampMixin):
__tablename__ = "organizations_function_types"
__table_args__ = (UniqueConstraint("tenant_id", "slug", name="uq_organizations_function_types_tenant_slug"),)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
slug: Mapped[str] = mapped_column(String(100), nullable=False)
name: Mapped[str] = mapped_column(String(255), nullable=False)
description: Mapped[str | None] = mapped_column(Text)
organization_unit_type_id: Mapped[str | None] = mapped_column(ForeignKey("organizations_unit_types.id", ondelete="SET NULL"), nullable=True, index=True)
delegable: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
act_in_place_allowed: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
settings: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
class OrganizationFunction(Base, TimestampMixin):
__tablename__ = "organizations_functions"
__table_args__ = (UniqueConstraint("tenant_id", "organization_unit_id", "slug", name="uq_organizations_functions_tenant_unit_slug"),)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(ForeignKey("tenancy_tenants.id", ondelete="CASCADE"), nullable=False, index=True)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
function_type_id: Mapped[str | None] = mapped_column(ForeignKey("organizations_function_types.id", ondelete="SET NULL"), nullable=True, index=True)
organization_unit_id: Mapped[str] = mapped_column(ForeignKey("organizations_units.id", ondelete="CASCADE"), nullable=False, index=True)
slug: Mapped[str] = mapped_column(String(100), nullable=False)
name: Mapped[str] = mapped_column(String(255), nullable=False)
@@ -49,17 +142,17 @@ class OrganizationFunctionAssignment(Base, TimestampMixin):
__table_args__ = (
UniqueConstraint(
"tenant_id",
"account_id",
"identity_id",
"function_id",
"organization_unit_id",
name="uq_organizations_function_assignments_account_scope",
name="uq_organizations_function_assignments_identity_scope",
),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(ForeignKey("tenancy_tenants.id", ondelete="CASCADE"), nullable=False, index=True)
account_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
identity_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
identity_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
account_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
function_id: Mapped[str] = mapped_column(ForeignKey("organizations_functions.id", ondelete="CASCADE"), nullable=False, index=True)
organization_unit_id: Mapped[str] = mapped_column(ForeignKey("organizations_units.id", ondelete="CASCADE"), nullable=False, index=True)
applies_to_subunits: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
@@ -75,6 +168,11 @@ class OrganizationFunctionAssignment(Base, TimestampMixin):
__all__ = [
"OrganizationFunction",
"OrganizationFunctionAssignment",
"OrganizationFunctionType",
"OrganizationRelation",
"OrganizationRelationType",
"OrganizationStructure",
"OrganizationUnit",
"OrganizationUnitType",
"new_uuid",
]

48
src/govoplan_organizations/backend/directory.py
View File

@@ -8,6 +8,7 @@ from govoplan_core.core.organizations import (
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 (
@@ -27,6 +28,7 @@ def _unit_ref(item: OrganizationUnit) -> OrganizationUnitRef:
tenant_id=item.tenant_id,
slug=item.slug,
name=item.name,
unit_type_id=item.unit_type_id,
parent_id=item.parent_id,
description=item.description,
status=_status(item.is_active), # type: ignore[arg-type]
@@ -40,6 +42,7 @@ def _function_ref(item: OrganizationFunction) -> OrganizationFunctionRef:
organization_unit_id=item.organization_unit_id,
slug=item.slug,
name=item.name,
function_type_id=item.function_type_id,
description=item.description,
delegable=item.delegable,
act_in_place_allowed=item.act_in_place_allowed,
@@ -51,8 +54,8 @@ def _assignment_ref(item: OrganizationFunctionAssignment) -> OrganizationFunctio
return OrganizationFunctionAssignmentRef(
id=item.id,
tenant_id=item.tenant_id,
account_id=item.account_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,
@@ -66,6 +69,9 @@ def _assignment_ref(item: OrganizationFunctionAssignment) -> OrganizationFunctio
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)
@@ -126,14 +132,52 @@ class SqlOrganizationDirectory(OrganizationDirectory):
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(
OrganizationFunctionAssignment.account_id == account_id,
identity_or_account_filter,
OrganizationFunctionAssignment.is_active.is_(True),
OrganizationFunction.is_active.is_(True),
or_(OrganizationFunctionAssignment.valid_from.is_(None), OrganizationFunctionAssignment.valid_from <= now),

109
src/govoplan_organizations/backend/manifest.py
View File

@@ -1,28 +1,122 @@
from __future__ import annotations
from pathlib import Path
from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY, IdentityDirectory
from govoplan_core.core.module_guards import persistent_table_uninstall_guard
from govoplan_core.core.modules import DocumentationTopic, MigrationSpec, ModuleContext, ModuleManifest
from govoplan_core.core.modules import (
DocumentationTopic,
FrontendModule,
FrontendRoute,
MigrationSpec,
ModuleContext,
ModuleManifest,
NavItem,
PermissionDefinition,
RoleTemplate,
)
from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY
from govoplan_core.db.base import Base
from govoplan_organizations.backend.db import models as organization_models # noqa: F401 - populate metadata
def _organization_directory(context: ModuleContext) -> object:
ORGANIZATIONS_READ_SCOPES = (
"organizations:model:read",
"organizations:unit:read",
"organizations:function:read",
"admin:settings:read",
)
def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
module_id, resource, action = scope.split(":", 2)
return PermissionDefinition(
scope=scope,
label=label,
description=description,
category="Organizations",
level="tenant",
module_id=module_id,
resource=resource,
action=action,
)
PERMISSIONS = (
_permission("organizations:model:read", "View organization model", "Read organization meta-model definitions such as unit types, structures, and relation types."),
_permission("organizations:model:write", "Manage organization model", "Create and edit organization meta-model definitions."),
_permission("organizations:unit:read", "View organization units", "Read concrete organization units and relations."),
_permission("organizations:unit:write", "Manage organization units", "Create and edit concrete organization units and relations."),
_permission("organizations:function:read", "View organization functions", "Read function definitions and identity-held assignments."),
_permission("organizations:function:write", "Manage organization functions", "Create and edit function definitions."),
_permission("organizations:function:assign", "Assign organization functions", "Assign identity-held functions in organization units."),
)
ROLE_TEMPLATES = (
RoleTemplate(
slug="organization_modeler",
name="Organization modeler",
description="Manage organization meta-model, concrete units, structures, functions, and assignments.",
permissions=tuple(permission.scope for permission in PERMISSIONS),
),
RoleTemplate(
slug="organization_viewer",
name="Organization viewer",
description="Read organization model, organization units, and function assignments.",
permissions=("organizations:model:read", "organizations:unit:read", "organizations:function:read"),
),
)
def _route_factory(context: ModuleContext):
del context
from govoplan_organizations.backend.api.v1.routes import router
return router
def _organization_directory(context: ModuleContext) -> object:
identity_directory: IdentityDirectory | None = None
registry = context.registry
if hasattr(registry, "has_capability") and registry.has_capability(CAPABILITY_IDENTITY_DIRECTORY):
capability = registry.require_capability(CAPABILITY_IDENTITY_DIRECTORY)
if isinstance(capability, IdentityDirectory):
identity_directory = capability
from govoplan_organizations.backend.directory import SqlOrganizationDirectory
return SqlOrganizationDirectory()
return SqlOrganizationDirectory(identity_directory=identity_directory)
manifest = ModuleManifest(
id="organizations",
name="Organizations",
version="0.1.6",
dependencies=("tenancy",),
migration_spec=MigrationSpec(module_id="organizations", metadata=Base.metadata),
required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
optional_dependencies=("tenancy", "identity", "access", "audit", "policy"),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
route_factory=_route_factory,
nav_items=(NavItem(path="/organizations", label="Organizations", icon="users", required_any=ORGANIZATIONS_READ_SCOPES, order=70),),
frontend=FrontendModule(
module_id="organizations",
package_name="@govoplan/organizations-webui",
routes=(FrontendRoute(path="/organizations", component="OrganizationsPage", required_any=ORGANIZATIONS_READ_SCOPES, order=70),),
nav_items=(NavItem(path="/organizations", label="Organizations", icon="users", required_any=ORGANIZATIONS_READ_SCOPES, order=70),),
),
migration_spec=MigrationSpec(
module_id="organizations",
metadata=Base.metadata,
script_location=str(Path(__file__).with_name("migrations") / "versions"),
),
uninstall_guard_providers=(
persistent_table_uninstall_guard(
organization_models.OrganizationUnitType,
organization_models.OrganizationStructure,
organization_models.OrganizationRelationType,
organization_models.OrganizationRelation,
organization_models.OrganizationUnit,
organization_models.OrganizationFunctionType,
organization_models.OrganizationFunction,
organization_models.OrganizationFunctionAssignment,
label="Organizations",
@@ -37,8 +131,9 @@ manifest = ModuleManifest(
title="Organization model",
summary="Organizations owns units, hierarchy, functions, and function assignments. Access maps those facts to roles and rights.",
body=(
"Use organization functions to model responsibilities that outlive people. "
"A function assignment can apply to one organization unit or to that unit and all subunits."
"Use organization unit types, structures, and relation types to model how the institution describes itself. "
"A concrete organization unit can participate in several structures at the same time, such as an employer hierarchy and an academic structure. "
"Functions are held by identities in organization units. Accounts only exercise those functions through identity and access policy."
),
layer="configured",
documentation_types=("admin", "user"),

1
src/govoplan_organizations/backend/migrations/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Organizations module migrations."""

299
src/govoplan_organizations/backend/migrations/versions/6d7e8f9a0b1c_organization_meta_model.py Normal file
View File

@@ -0,0 +1,299 @@
"""organization meta-model
Revision ID: 6d7e8f9a0b1c
Revises: 2e3f4a5b6c7d
Create Date: 2026-07-10 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "6d7e8f9a0b1c"
down_revision = "2e3f4a5b6c7d"
branch_labels = None
depends_on = None
def _tables() -> set[str]:
return set(sa.inspect(op.get_bind()).get_table_names())
def _columns(table_name: str) -> set[str]:
if table_name not in _tables():
return set()
return {item["name"] for item in sa.inspect(op.get_bind()).get_columns(table_name)}
def _indexes(table_name: str) -> set[str]:
if table_name not in _tables():
return set()
return {item["name"] for item in sa.inspect(op.get_bind()).get_indexes(table_name)}
def _create_index_if_missing(name: str, table_name: str, columns: list[str], *, unique: bool = False) -> None:
if table_name in _tables() and name not in _indexes(table_name):
op.create_index(name, table_name, columns, unique=unique)
def _drop_index_if_exists(name: str, table_name: str) -> None:
if table_name in _tables() and name in _indexes(table_name):
op.drop_index(name, table_name=table_name)
def _json_column(name: str) -> sa.Column:
return sa.Column(name, sa.JSON(), nullable=False)
def _timestamp_columns() -> tuple[sa.Column, sa.Column]:
return (
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
def _create_unit_types() -> None:
if "organizations_unit_types" not in _tables():
op.create_table(
"organizations_unit_types",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("slug", sa.String(length=100), nullable=False),
sa.Column("name", sa.String(length=255), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("is_active", sa.Boolean(), nullable=False),
_json_column("settings"),
*_timestamp_columns(),
sa.PrimaryKeyConstraint("id", name=op.f("pk_organizations_unit_types")),
sa.UniqueConstraint("tenant_id", "slug", name="uq_organizations_unit_types_tenant_slug"),
)
_create_index_if_missing(op.f("ix_organizations_unit_types_tenant_id"), "organizations_unit_types", ["tenant_id"])
def _create_structures() -> None:
if "organizations_structures" not in _tables():
op.create_table(
"organizations_structures",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("slug", sa.String(length=100), nullable=False),
sa.Column("name", sa.String(length=255), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("structure_kind", sa.String(length=30), nullable=False),
sa.Column("is_active", sa.Boolean(), nullable=False),
_json_column("settings"),
*_timestamp_columns(),
sa.PrimaryKeyConstraint("id", name=op.f("pk_organizations_structures")),
sa.UniqueConstraint("tenant_id", "slug", name="uq_organizations_structures_tenant_slug"),
)
_create_index_if_missing(op.f("ix_organizations_structures_tenant_id"), "organizations_structures", ["tenant_id"])
def _create_relation_types() -> None:
if "organizations_relation_types" not in _tables():
op.create_table(
"organizations_relation_types",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("structure_id", sa.String(length=36), nullable=True),
sa.Column("slug", sa.String(length=100), nullable=False),
sa.Column("name", sa.String(length=255), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("source_unit_type_id", sa.String(length=36), nullable=True),
sa.Column("target_unit_type_id", sa.String(length=36), nullable=True),
sa.Column("is_hierarchical", sa.Boolean(), nullable=False),
sa.Column("allow_cycles", sa.Boolean(), nullable=False),
sa.Column("is_active", sa.Boolean(), nullable=False),
_json_column("settings"),
*_timestamp_columns(),
sa.ForeignKeyConstraint(["source_unit_type_id"], ["organizations_unit_types.id"], name=op.f("fk_organizations_relation_types_source_unit_type_id_organizations_unit_types"), ondelete="SET NULL"),
sa.ForeignKeyConstraint(["structure_id"], ["organizations_structures.id"], name=op.f("fk_organizations_relation_types_structure_id_organizations_structures"), ondelete="SET NULL"),
sa.ForeignKeyConstraint(["target_unit_type_id"], ["organizations_unit_types.id"], name=op.f("fk_organizations_relation_types_target_unit_type_id_organizations_unit_types"), ondelete="SET NULL"),
sa.PrimaryKeyConstraint("id", name=op.f("pk_organizations_relation_types")),
sa.UniqueConstraint("tenant_id", "slug", name="uq_organizations_relation_types_tenant_slug"),
)
for column in ("tenant_id", "source_unit_type_id", "structure_id", "target_unit_type_id"):
_create_index_if_missing(op.f(f"ix_organizations_relation_types_{column}"), "organizations_relation_types", [column])
_create_index_if_missing("ix_organizations_relation_types_structure", "organizations_relation_types", ["tenant_id", "structure_id"])
def _create_units() -> None:
if "organizations_units" not in _tables():
op.create_table(
"organizations_units",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("unit_type_id", sa.String(length=36), nullable=True),
sa.Column("parent_id", sa.String(length=36), nullable=True),
sa.Column("slug", sa.String(length=100), nullable=False),
sa.Column("name", sa.String(length=255), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("is_active", sa.Boolean(), nullable=False),
_json_column("settings"),
*_timestamp_columns(),
sa.ForeignKeyConstraint(["parent_id"], ["organizations_units.id"], name=op.f("fk_organizations_units_parent_id_organizations_units"), ondelete="SET NULL"),
sa.ForeignKeyConstraint(["unit_type_id"], ["organizations_unit_types.id"], name=op.f("fk_organizations_units_unit_type_id_organizations_unit_types"), ondelete="SET NULL"),
sa.PrimaryKeyConstraint("id", name=op.f("pk_organizations_units")),
sa.UniqueConstraint("tenant_id", "slug", name="uq_organizations_units_tenant_slug"),
)
elif "unit_type_id" not in _columns("organizations_units"):
op.add_column("organizations_units", sa.Column("unit_type_id", sa.String(length=36), nullable=True))
for column in ("parent_id", "tenant_id", "unit_type_id"):
_create_index_if_missing(op.f(f"ix_organizations_units_{column}"), "organizations_units", [column])
def _create_relations() -> None:
if "organizations_relations" not in _tables():
op.create_table(
"organizations_relations",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("structure_id", sa.String(length=36), nullable=False),
sa.Column("relation_type_id", sa.String(length=36), nullable=False),
sa.Column("source_unit_id", sa.String(length=36), nullable=False),
sa.Column("target_unit_id", sa.String(length=36), nullable=False),
sa.Column("valid_from", sa.DateTime(timezone=True), nullable=True),
sa.Column("valid_until", sa.DateTime(timezone=True), nullable=True),
sa.Column("is_active", sa.Boolean(), nullable=False),
_json_column("settings"),
*_timestamp_columns(),
sa.ForeignKeyConstraint(["relation_type_id"], ["organizations_relation_types.id"], name=op.f("fk_organizations_relations_relation_type_id_organizations_relation_types"), ondelete="CASCADE"),
sa.ForeignKeyConstraint(["source_unit_id"], ["organizations_units.id"], name=op.f("fk_organizations_relations_source_unit_id_organizations_units"), ondelete="CASCADE"),
sa.ForeignKeyConstraint(["structure_id"], ["organizations_structures.id"], name=op.f("fk_organizations_relations_structure_id_organizations_structures"), ondelete="CASCADE"),
sa.ForeignKeyConstraint(["target_unit_id"], ["organizations_units.id"], name=op.f("fk_organizations_relations_target_unit_id_organizations_units"), ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id", name=op.f("pk_organizations_relations")),
sa.UniqueConstraint("tenant_id", "structure_id", "relation_type_id", "source_unit_id", "target_unit_id", name="uq_organizations_relations_edge"),
)
for column in ("relation_type_id", "source_unit_id", "structure_id", "target_unit_id", "tenant_id"):
_create_index_if_missing(op.f(f"ix_organizations_relations_{column}"), "organizations_relations", [column])
_create_index_if_missing("ix_organizations_relations_structure_source", "organizations_relations", ["tenant_id", "structure_id", "source_unit_id"])
_create_index_if_missing("ix_organizations_relations_structure_target", "organizations_relations", ["tenant_id", "structure_id", "target_unit_id"])
def _create_function_types() -> None:
if "organizations_function_types" not in _tables():
op.create_table(
"organizations_function_types",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("slug", sa.String(length=100), nullable=False),
sa.Column("name", sa.String(length=255), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("organization_unit_type_id", sa.String(length=36), nullable=True),
sa.Column("delegable", sa.Boolean(), nullable=False),
sa.Column("act_in_place_allowed", sa.Boolean(), nullable=False),
sa.Column("is_active", sa.Boolean(), nullable=False),
_json_column("settings"),
*_timestamp_columns(),
sa.ForeignKeyConstraint(["organization_unit_type_id"], ["organizations_unit_types.id"], name=op.f("fk_organizations_function_types_organization_unit_type_id_organizations_unit_types"), ondelete="SET NULL"),
sa.PrimaryKeyConstraint("id", name=op.f("pk_organizations_function_types")),
sa.UniqueConstraint("tenant_id", "slug", name="uq_organizations_function_types_tenant_slug"),
)
for column in ("organization_unit_type_id", "tenant_id"):
_create_index_if_missing(op.f(f"ix_organizations_function_types_{column}"), "organizations_function_types", [column])
def _create_functions() -> None:
if "organizations_functions" not in _tables():
op.create_table(
"organizations_functions",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("function_type_id", sa.String(length=36), nullable=True),
sa.Column("organization_unit_id", sa.String(length=36), nullable=False),
sa.Column("slug", sa.String(length=100), nullable=False),
sa.Column("name", sa.String(length=255), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("delegable", sa.Boolean(), nullable=False),
sa.Column("act_in_place_allowed", sa.Boolean(), nullable=False),
sa.Column("is_active", sa.Boolean(), nullable=False),
_json_column("settings"),
*_timestamp_columns(),
sa.ForeignKeyConstraint(["function_type_id"], ["organizations_function_types.id"], name=op.f("fk_organizations_functions_function_type_id_organizations_function_types"), ondelete="SET NULL"),
sa.ForeignKeyConstraint(["organization_unit_id"], ["organizations_units.id"], name=op.f("fk_organizations_functions_organization_unit_id_organizations_units"), ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id", name=op.f("pk_organizations_functions")),
sa.UniqueConstraint("tenant_id", "organization_unit_id", "slug", name="uq_organizations_functions_tenant_unit_slug"),
)
elif "function_type_id" not in _columns("organizations_functions"):
op.add_column("organizations_functions", sa.Column("function_type_id", sa.String(length=36), nullable=True))
for column in ("function_type_id", "organization_unit_id", "tenant_id"):
_create_index_if_missing(op.f(f"ix_organizations_functions_{column}"), "organizations_functions", [column])
def _create_function_assignments() -> None:
if "organizations_function_assignments" not in _tables():
op.create_table(
"organizations_function_assignments",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("identity_id", sa.String(length=36), nullable=False),
sa.Column("account_id", sa.String(length=36), nullable=True),
sa.Column("function_id", sa.String(length=36), nullable=False),
sa.Column("organization_unit_id", sa.String(length=36), nullable=False),
sa.Column("applies_to_subunits", sa.Boolean(), nullable=False),
sa.Column("source", sa.String(length=50), nullable=False),
sa.Column("delegated_from_assignment_id", sa.String(length=36), nullable=True),
sa.Column("acting_for_account_id", sa.String(length=36), nullable=True),
sa.Column("valid_from", sa.DateTime(timezone=True), nullable=True),
sa.Column("valid_until", sa.DateTime(timezone=True), nullable=True),
sa.Column("is_active", sa.Boolean(), nullable=False),
_json_column("settings"),
*_timestamp_columns(),
sa.ForeignKeyConstraint(["delegated_from_assignment_id"], ["organizations_function_assignments.id"], name=op.f("fk_organizations_function_assignments_delegated_from_assignment_id_organizations_function_assignments"), ondelete="SET NULL"),
sa.ForeignKeyConstraint(["function_id"], ["organizations_functions.id"], name=op.f("fk_organizations_function_assignments_function_id_organizations_functions"), ondelete="CASCADE"),
sa.ForeignKeyConstraint(["organization_unit_id"], ["organizations_units.id"], name=op.f("fk_organizations_function_assignments_organization_unit_id_organizations_units"), ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id", name=op.f("pk_organizations_function_assignments")),
sa.UniqueConstraint("tenant_id", "identity_id", "function_id", "organization_unit_id", name="uq_organizations_function_assignments_identity_scope"),
)
else:
columns = _columns("organizations_function_assignments")
if "identity_id" not in columns:
op.add_column("organizations_function_assignments", sa.Column("identity_id", sa.String(length=36), nullable=True))
op.execute(sa.text("UPDATE organizations_function_assignments SET identity_id = account_id WHERE identity_id IS NULL"))
for column in (
"account_id",
"acting_for_account_id",
"delegated_from_assignment_id",
"function_id",
"identity_id",
"organization_unit_id",
"tenant_id",
):
_create_index_if_missing(op.f(f"ix_organizations_function_assignments_{column}"), "organizations_function_assignments", [column])
_create_index_if_missing(
"uq_organizations_function_assignments_identity_scope",
"organizations_function_assignments",
["tenant_id", "identity_id", "function_id", "organization_unit_id"],
unique=True,
)
def upgrade() -> None:
_create_unit_types()
_create_structures()
_create_relation_types()
_create_units()
_create_relations()
_create_function_types()
_create_functions()
_create_function_assignments()
def downgrade() -> None:
for table_name, indexes in (
("organizations_function_assignments", ("uq_organizations_function_assignments_identity_scope",)),
("organizations_relations", ("ix_organizations_relations_structure_source", "ix_organizations_relations_structure_target")),
("organizations_relation_types", ("ix_organizations_relation_types_structure",)),
):
for index_name in indexes:
_drop_index_if_exists(index_name, table_name)
for table_name in (
"organizations_relations",
"organizations_relation_types",
"organizations_structures",
"organizations_function_types",
"organizations_unit_types",
):
if table_name in _tables():
op.drop_table(table_name)

1
src/govoplan_organizations/backend/migrations/versions/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Organizations Alembic revisions."""