443 lines
15 KiB
Python
443 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_organizations.backend.api.v1.schemas import (
|
|
OrganizationModelTemplateDefinition,
|
|
)
|
|
from govoplan_organizations.backend.db.models import (
|
|
OrganizationFunction,
|
|
OrganizationFunctionType,
|
|
OrganizationModelInstantiation,
|
|
OrganizationModelTemplate,
|
|
OrganizationModelTemplateVersion,
|
|
OrganizationRelation,
|
|
OrganizationRelationType,
|
|
OrganizationStructure,
|
|
OrganizationUnit,
|
|
OrganizationUnitType,
|
|
)
|
|
|
|
|
|
class OrganizationTemplateError(ValueError):
|
|
pass
|
|
|
|
|
|
TENANT_MODEL_TYPES = (
|
|
OrganizationRelation,
|
|
OrganizationFunction,
|
|
OrganizationUnit,
|
|
OrganizationRelationType,
|
|
OrganizationFunctionType,
|
|
OrganizationStructure,
|
|
OrganizationUnitType,
|
|
)
|
|
|
|
|
|
def canonical_template_definition(
|
|
definition: OrganizationModelTemplateDefinition,
|
|
) -> tuple[dict[str, Any], str]:
|
|
_validate_definition_references(definition)
|
|
payload = definition.model_dump(mode="json")
|
|
encoded = json.dumps(
|
|
payload,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
return payload, hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
def instantiate_template_version(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
template: OrganizationModelTemplate,
|
|
version: OrganizationModelTemplateVersion,
|
|
actor_account_id: str | None,
|
|
) -> OrganizationModelInstantiation:
|
|
if version.template_id != template.id or version.status != "published":
|
|
raise OrganizationTemplateError(
|
|
"Only a published version of the selected template can be instantiated"
|
|
)
|
|
if any(
|
|
session.query(model).filter(model.tenant_id == tenant_id).first()
|
|
is not None
|
|
for model in TENANT_MODEL_TYPES
|
|
):
|
|
raise OrganizationTemplateError(
|
|
"Organization templates can currently be instantiated only into an empty tenant model"
|
|
)
|
|
existing = (
|
|
session.query(OrganizationModelInstantiation)
|
|
.filter(
|
|
OrganizationModelInstantiation.tenant_id == tenant_id,
|
|
OrganizationModelInstantiation.template_version_id == version.id,
|
|
)
|
|
.one_or_none()
|
|
)
|
|
if existing is not None:
|
|
return existing
|
|
|
|
definition = OrganizationModelTemplateDefinition.model_validate(
|
|
version.definition
|
|
)
|
|
_validate_definition_references(definition)
|
|
provenance_base = {
|
|
"template_id": template.id,
|
|
"template_slug": template.slug,
|
|
"template_version_id": version.id,
|
|
"template_version": version.version,
|
|
"definition_sha256": version.definition_sha256,
|
|
}
|
|
|
|
unit_types = {
|
|
item.slug: OrganizationUnitType(
|
|
tenant_id=tenant_id,
|
|
slug=item.slug,
|
|
name=item.name,
|
|
description=item.description,
|
|
is_active=item.is_active,
|
|
settings=_settings(item.settings, provenance_base, item.slug),
|
|
)
|
|
for item in definition.unit_types
|
|
}
|
|
structures = {
|
|
item.slug: OrganizationStructure(
|
|
tenant_id=tenant_id,
|
|
slug=item.slug,
|
|
name=item.name,
|
|
description=item.description,
|
|
structure_kind=item.structure_kind,
|
|
is_active=item.is_active,
|
|
settings=_settings(item.settings, provenance_base, item.slug),
|
|
)
|
|
for item in definition.structures
|
|
}
|
|
session.add_all([*unit_types.values(), *structures.values()])
|
|
session.flush()
|
|
|
|
relation_types = {
|
|
item.slug: OrganizationRelationType(
|
|
tenant_id=tenant_id,
|
|
structure_id=_ref_id(structures, item.structure_slug),
|
|
slug=item.slug,
|
|
name=item.name,
|
|
description=item.description,
|
|
source_unit_type_id=_ref_id(
|
|
unit_types,
|
|
item.source_unit_type_slug,
|
|
),
|
|
target_unit_type_id=_ref_id(
|
|
unit_types,
|
|
item.target_unit_type_slug,
|
|
),
|
|
is_hierarchical=item.is_hierarchical,
|
|
allow_cycles=item.allow_cycles,
|
|
is_active=item.is_active,
|
|
settings=_settings(item.settings, provenance_base, item.slug),
|
|
)
|
|
for item in definition.relation_types
|
|
}
|
|
function_types = {
|
|
item.slug: OrganizationFunctionType(
|
|
tenant_id=tenant_id,
|
|
slug=item.slug,
|
|
name=item.name,
|
|
description=item.description,
|
|
organization_unit_type_id=_ref_id(
|
|
unit_types,
|
|
item.organization_unit_type_slug,
|
|
),
|
|
delegable=item.delegable,
|
|
act_in_place_allowed=item.act_in_place_allowed,
|
|
is_active=item.is_active,
|
|
settings=_settings(item.settings, provenance_base, item.slug),
|
|
)
|
|
for item in definition.function_types
|
|
}
|
|
session.add_all([*relation_types.values(), *function_types.values()])
|
|
session.flush()
|
|
|
|
units = {
|
|
item.slug: OrganizationUnit(
|
|
tenant_id=tenant_id,
|
|
unit_type_id=_ref_id(unit_types, item.unit_type_slug),
|
|
slug=item.slug,
|
|
name=item.name,
|
|
description=item.description,
|
|
is_active=item.is_active,
|
|
settings=_settings(item.settings, provenance_base, item.slug),
|
|
)
|
|
for item in definition.units
|
|
}
|
|
session.add_all(list(units.values()))
|
|
session.flush()
|
|
for item in definition.units:
|
|
units[item.slug].parent_id = _ref_id(units, item.parent_slug)
|
|
|
|
relations = [
|
|
OrganizationRelation(
|
|
tenant_id=tenant_id,
|
|
structure_id=structures[item.structure_slug].id,
|
|
relation_type_id=relation_types[item.relation_type_slug].id,
|
|
source_unit_id=units[item.source_unit_slug].id,
|
|
target_unit_id=units[item.target_unit_slug].id,
|
|
valid_from=item.valid_from,
|
|
valid_until=item.valid_until,
|
|
is_active=item.is_active,
|
|
settings=_settings(
|
|
item.settings,
|
|
provenance_base,
|
|
(
|
|
f"{item.structure_slug}:{item.relation_type_slug}:"
|
|
f"{item.source_unit_slug}:{item.target_unit_slug}"
|
|
),
|
|
),
|
|
)
|
|
for item in definition.relations
|
|
]
|
|
functions = [
|
|
OrganizationFunction(
|
|
tenant_id=tenant_id,
|
|
function_type_id=_ref_id(
|
|
function_types,
|
|
item.function_type_slug,
|
|
),
|
|
organization_unit_id=units[item.organization_unit_slug].id,
|
|
slug=item.slug,
|
|
name=item.name,
|
|
description=item.description,
|
|
delegable=item.delegable,
|
|
act_in_place_allowed=item.act_in_place_allowed,
|
|
is_active=item.is_active,
|
|
settings=_settings(item.settings, provenance_base, item.slug),
|
|
)
|
|
for item in definition.functions
|
|
]
|
|
session.add_all([*relations, *functions])
|
|
|
|
counts = {
|
|
"unit_types": len(unit_types),
|
|
"structures": len(structures),
|
|
"relation_types": len(relation_types),
|
|
"units": len(units),
|
|
"relations": len(relations),
|
|
"function_types": len(function_types),
|
|
"functions": len(functions),
|
|
}
|
|
instantiation = OrganizationModelInstantiation(
|
|
tenant_id=tenant_id,
|
|
template_id=template.id,
|
|
template_version_id=version.id,
|
|
source_definition_sha256=version.definition_sha256,
|
|
instantiated_by_account_id=actor_account_id,
|
|
object_counts=counts,
|
|
provenance={
|
|
**provenance_base,
|
|
"copy_semantics": "tenant_owned_no_live_inheritance",
|
|
},
|
|
)
|
|
session.add(instantiation)
|
|
session.flush()
|
|
return instantiation
|
|
|
|
|
|
def _validate_definition_references(
|
|
definition: OrganizationModelTemplateDefinition,
|
|
) -> None:
|
|
collections = {
|
|
"unit type": [item.slug for item in definition.unit_types],
|
|
"structure": [item.slug for item in definition.structures],
|
|
"relation type": [item.slug for item in definition.relation_types],
|
|
"unit": [item.slug for item in definition.units],
|
|
"function type": [item.slug for item in definition.function_types],
|
|
"function": [item.slug for item in definition.functions],
|
|
}
|
|
for label, slugs in collections.items():
|
|
if len(slugs) != len(set(slugs)):
|
|
raise OrganizationTemplateError(
|
|
f"Template contains duplicate {label} slugs"
|
|
)
|
|
unit_types = set(collections["unit type"])
|
|
structures = set(collections["structure"])
|
|
relation_types = set(collections["relation type"])
|
|
units = set(collections["unit"])
|
|
function_types = set(collections["function type"])
|
|
unit_by_slug = {item.slug: item for item in definition.units}
|
|
relation_type_by_slug = {
|
|
item.slug: item for item in definition.relation_types
|
|
}
|
|
function_type_by_slug = {
|
|
item.slug: item for item in definition.function_types
|
|
}
|
|
for item in definition.relation_types:
|
|
_require_ref(structures, item.structure_slug, "structure")
|
|
_require_ref(unit_types, item.source_unit_type_slug, "source unit type")
|
|
_require_ref(unit_types, item.target_unit_type_slug, "target unit type")
|
|
for item in definition.units:
|
|
_require_ref(unit_types, item.unit_type_slug, "unit type")
|
|
_require_ref(units, item.parent_slug, "parent unit")
|
|
if item.parent_slug == item.slug:
|
|
raise OrganizationTemplateError("A unit cannot be its own parent")
|
|
_require_acyclic_graph(
|
|
{
|
|
item.slug: {item.parent_slug}
|
|
for item in definition.units
|
|
if item.parent_slug is not None
|
|
},
|
|
label="unit parent hierarchy",
|
|
)
|
|
relation_edges: dict[str, dict[str, set[str]]] = {}
|
|
relation_keys: set[tuple[str, str, str, str]] = set()
|
|
for item in definition.relations:
|
|
_require_ref(structures, item.structure_slug, "structure")
|
|
_require_ref(relation_types, item.relation_type_slug, "relation type")
|
|
_require_ref(units, item.source_unit_slug, "source unit")
|
|
_require_ref(units, item.target_unit_slug, "target unit")
|
|
if (
|
|
item.valid_from is not None
|
|
and item.valid_until is not None
|
|
and item.valid_until < item.valid_from
|
|
):
|
|
raise OrganizationTemplateError(
|
|
"A relation validity end cannot precede its start"
|
|
)
|
|
relation_type = relation_type_by_slug[item.relation_type_slug]
|
|
if (
|
|
relation_type.structure_slug is not None
|
|
and relation_type.structure_slug != item.structure_slug
|
|
):
|
|
raise OrganizationTemplateError(
|
|
f"Relation {item.relation_type_slug!r} belongs to structure "
|
|
f"{relation_type.structure_slug!r}, not {item.structure_slug!r}"
|
|
)
|
|
source_unit_type = unit_by_slug[item.source_unit_slug].unit_type_slug
|
|
target_unit_type = unit_by_slug[item.target_unit_slug].unit_type_slug
|
|
if (
|
|
relation_type.source_unit_type_slug is not None
|
|
and relation_type.source_unit_type_slug != source_unit_type
|
|
):
|
|
raise OrganizationTemplateError(
|
|
f"Relation {item.relation_type_slug!r} does not allow source "
|
|
f"unit {item.source_unit_slug!r}"
|
|
)
|
|
if (
|
|
relation_type.target_unit_type_slug is not None
|
|
and relation_type.target_unit_type_slug != target_unit_type
|
|
):
|
|
raise OrganizationTemplateError(
|
|
f"Relation {item.relation_type_slug!r} does not allow target "
|
|
f"unit {item.target_unit_slug!r}"
|
|
)
|
|
relation_key = (
|
|
item.structure_slug,
|
|
item.relation_type_slug,
|
|
item.source_unit_slug,
|
|
item.target_unit_slug,
|
|
)
|
|
if relation_key in relation_keys:
|
|
raise OrganizationTemplateError(
|
|
"Template contains a duplicate organization relation"
|
|
)
|
|
relation_keys.add(relation_key)
|
|
if relation_type.is_hierarchical and not relation_type.allow_cycles:
|
|
targets = relation_edges.setdefault(item.relation_type_slug, {})
|
|
targets.setdefault(item.source_unit_slug, set()).add(
|
|
item.target_unit_slug
|
|
)
|
|
for relation_type_slug, edges in relation_edges.items():
|
|
_require_acyclic_graph(
|
|
edges,
|
|
label=f"relation hierarchy {relation_type_slug!r}",
|
|
)
|
|
for item in definition.function_types:
|
|
_require_ref(
|
|
unit_types,
|
|
item.organization_unit_type_slug,
|
|
"organization unit type",
|
|
)
|
|
for item in definition.functions:
|
|
_require_ref(function_types, item.function_type_slug, "function type")
|
|
_require_ref(units, item.organization_unit_slug, "organization unit")
|
|
if item.function_type_slug is None:
|
|
continue
|
|
expected_unit_type = function_type_by_slug[
|
|
item.function_type_slug
|
|
].organization_unit_type_slug
|
|
actual_unit_type = unit_by_slug[
|
|
item.organization_unit_slug
|
|
].unit_type_slug
|
|
if (
|
|
expected_unit_type is not None
|
|
and expected_unit_type != actual_unit_type
|
|
):
|
|
raise OrganizationTemplateError(
|
|
f"Function type {item.function_type_slug!r} does not apply to "
|
|
f"unit {item.organization_unit_slug!r}"
|
|
)
|
|
|
|
|
|
def _require_ref(
|
|
values: set[str],
|
|
value: str | None,
|
|
label: str,
|
|
) -> None:
|
|
if value is not None and value not in values:
|
|
raise OrganizationTemplateError(
|
|
f"Template references an unknown {label}: {value}"
|
|
)
|
|
|
|
|
|
def _ref_id(rows: dict[str, Any], slug: str | None) -> str | None:
|
|
return rows[slug].id if slug is not None else None
|
|
|
|
|
|
def _require_acyclic_graph(
|
|
edges: dict[str, set[str]],
|
|
*,
|
|
label: str,
|
|
) -> None:
|
|
complete: set[str] = set()
|
|
active: set[str] = set()
|
|
|
|
def visit(node: str) -> None:
|
|
if node in complete:
|
|
return
|
|
if node in active:
|
|
raise OrganizationTemplateError(
|
|
f"Template contains a cycle in {label}"
|
|
)
|
|
active.add(node)
|
|
for target in edges.get(node, ()):
|
|
visit(target)
|
|
active.remove(node)
|
|
complete.add(node)
|
|
|
|
for start in edges:
|
|
visit(start)
|
|
|
|
|
|
def _settings(
|
|
settings: dict[str, Any],
|
|
provenance_base: dict[str, Any],
|
|
source_key: str,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
**dict(settings),
|
|
"template_provenance": {
|
|
**provenance_base,
|
|
"source_key": source_key,
|
|
},
|
|
}
|
|
|
|
|
|
__all__ = [
|
|
"OrganizationTemplateError",
|
|
"canonical_template_definition",
|
|
"instantiate_template_version",
|
|
]
|