365 lines
13 KiB
Python
365 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from govoplan_core.core.institutional import (
|
|
EvidenceReference,
|
|
GovernedContextEnvelope,
|
|
InstitutionalContextError,
|
|
InstitutionalReference,
|
|
)
|
|
|
|
|
|
CASE_ACCESS_SUBJECT_KINDS = frozenset(
|
|
{
|
|
"account",
|
|
"identity",
|
|
"group",
|
|
"role",
|
|
"function",
|
|
"function_assignment",
|
|
"organization_unit",
|
|
"service_account",
|
|
}
|
|
)
|
|
CASE_ACCESS_PERMISSIONS = frozenset({"read", "update", "share", "admin"})
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CaseGrant:
|
|
subject_kind: str
|
|
subject_id: str
|
|
permissions: tuple[str, ...] = ("read",)
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.subject_kind not in CASE_ACCESS_SUBJECT_KINDS:
|
|
raise InstitutionalContextError("Case access subject kind is invalid.")
|
|
if not self.subject_id.strip():
|
|
raise InstitutionalContextError("Case access subject id is required.")
|
|
normalized = tuple(dict.fromkeys(self.permissions))
|
|
if not normalized or set(normalized) - CASE_ACCESS_PERMISSIONS:
|
|
raise InstitutionalContextError("Case access permissions are invalid.")
|
|
object.__setattr__(self, "permissions", normalized)
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"subject_kind": self.subject_kind,
|
|
"subject_id": self.subject_id,
|
|
"permissions": list(self.permissions),
|
|
}
|
|
|
|
@classmethod
|
|
def from_mapping(cls, value: Mapping[str, object]) -> "CaseGrant":
|
|
raw_permissions = value.get("permissions", ["read"])
|
|
if not isinstance(raw_permissions, (list, tuple)):
|
|
raise InstitutionalContextError(
|
|
"Case access permissions must be a list."
|
|
)
|
|
return cls(
|
|
subject_kind=_text(value, "subject_kind"),
|
|
subject_id=_text(value, "subject_id"),
|
|
permissions=tuple(str(item) for item in raw_permissions),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CaseRecord:
|
|
reference: InstitutionalReference
|
|
case_number: str
|
|
case_type_key: str
|
|
status_key: str
|
|
title: str
|
|
context: GovernedContextEnvelope
|
|
opened_at: datetime
|
|
recorded_at: datetime
|
|
change_reason: str
|
|
access_mode: str = "tenant"
|
|
access_grants: tuple[CaseGrant, ...] = ()
|
|
service_ref: InstitutionalReference | None = None
|
|
party_refs: tuple[InstitutionalReference, ...] = ()
|
|
assignment_refs: tuple[InstitutionalReference, ...] = ()
|
|
evidence_refs: tuple[EvidenceReference, ...] = ()
|
|
decision_refs: tuple[InstitutionalReference, ...] = ()
|
|
record_refs: tuple[InstitutionalReference, ...] = ()
|
|
deadline_at: datetime | None = None
|
|
closed_at: datetime | None = None
|
|
metadata: Mapping[str, Any] = field(default_factory=dict)
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.reference.kind != "case" or self.reference.owner_module != "cases":
|
|
raise InstitutionalContextError(
|
|
"Case records require a Cases-owned case reference."
|
|
)
|
|
if not self.reference.version or not self.reference.version.isdigit():
|
|
raise InstitutionalContextError(
|
|
"Case references require a positive integer revision."
|
|
)
|
|
if int(self.reference.version) < 1:
|
|
raise InstitutionalContextError(
|
|
"Case references require a positive integer revision."
|
|
)
|
|
for value, label in (
|
|
(self.case_number, "Case number"),
|
|
(self.case_type_key, "Case type"),
|
|
(self.status_key, "Case status"),
|
|
(self.title, "Case title"),
|
|
(self.change_reason, "Case change reason"),
|
|
):
|
|
if not value.strip():
|
|
raise InstitutionalContextError(f"{label} is required.")
|
|
if self.access_mode not in {"tenant", "restricted"}:
|
|
raise InstitutionalContextError("Unsupported case access mode.")
|
|
grant_keys = {
|
|
(item.subject_kind, item.subject_id) for item in self.access_grants
|
|
}
|
|
if len(grant_keys) != len(self.access_grants):
|
|
raise InstitutionalContextError("Case access grants must be unique.")
|
|
for value, label in (
|
|
(self.opened_at, "Case opened_at"),
|
|
(self.recorded_at, "Case recorded_at"),
|
|
(self.deadline_at, "Case deadline_at"),
|
|
(self.closed_at, "Case closed_at"),
|
|
):
|
|
_require_aware(value, label)
|
|
if self.closed_at is not None and self.closed_at < self.opened_at:
|
|
raise InstitutionalContextError(
|
|
"Case closed_at cannot precede opened_at."
|
|
)
|
|
tenant_id = self.reference.tenant_id
|
|
if self.context.tenant_id != tenant_id:
|
|
raise InstitutionalContextError(
|
|
"Case institutional context belongs to another tenant."
|
|
)
|
|
if self.context.case_ref is not None and not _same_object(
|
|
self.context.case_ref,
|
|
self.reference,
|
|
):
|
|
raise InstitutionalContextError(
|
|
"Case institutional context references another case."
|
|
)
|
|
if self.service_ref is not None and self.service_ref.kind != "service":
|
|
raise InstitutionalContextError(
|
|
"Case service_ref must identify an institutional Service."
|
|
)
|
|
_validate_references(
|
|
tenant_id,
|
|
self.party_refs,
|
|
expected_kinds={"party"},
|
|
label="Case parties",
|
|
)
|
|
_validate_references(
|
|
tenant_id,
|
|
self.assignment_refs,
|
|
expected_kinds={
|
|
"function",
|
|
"function_assignment",
|
|
"organization_unit",
|
|
"work_item",
|
|
},
|
|
label="Case assignments",
|
|
)
|
|
_validate_references(
|
|
tenant_id,
|
|
self.decision_refs,
|
|
expected_kinds={"decision"},
|
|
label="Case decisions",
|
|
)
|
|
_validate_references(
|
|
tenant_id,
|
|
self.record_refs,
|
|
expected_kinds={"record"},
|
|
label="Case records",
|
|
)
|
|
references = (
|
|
(self.service_ref,) if self.service_ref is not None else ()
|
|
)
|
|
if any(item.tenant_id != tenant_id for item in references):
|
|
raise InstitutionalContextError(
|
|
"Case references cannot cross tenants."
|
|
)
|
|
if any(item.tenant_id != tenant_id for item in self.evidence_refs):
|
|
raise InstitutionalContextError(
|
|
"Case evidence cannot cross tenants."
|
|
)
|
|
if len(self.metadata) > 100:
|
|
raise InstitutionalContextError(
|
|
"Case metadata is limited to 100 entries."
|
|
)
|
|
|
|
@property
|
|
def revision(self) -> int:
|
|
return int(self.reference.version or "0")
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"reference": self.reference.to_dict(),
|
|
"revision": self.revision,
|
|
"case_number": self.case_number,
|
|
"case_type_key": self.case_type_key,
|
|
"status_key": self.status_key,
|
|
"title": self.title,
|
|
"access_mode": self.access_mode,
|
|
"access_grants": [item.to_dict() for item in self.access_grants],
|
|
"context": self.context.to_dict(),
|
|
"service_ref": self.service_ref.to_dict() if self.service_ref else None,
|
|
"party_refs": [item.to_dict() for item in self.party_refs],
|
|
"assignment_refs": [item.to_dict() for item in self.assignment_refs],
|
|
"evidence_refs": [item.to_dict() for item in self.evidence_refs],
|
|
"decision_refs": [item.to_dict() for item in self.decision_refs],
|
|
"record_refs": [item.to_dict() for item in self.record_refs],
|
|
"opened_at": self.opened_at.isoformat(),
|
|
"recorded_at": self.recorded_at.isoformat(),
|
|
"deadline_at": self.deadline_at.isoformat() if self.deadline_at else None,
|
|
"closed_at": self.closed_at.isoformat() if self.closed_at else None,
|
|
"change_reason": self.change_reason,
|
|
"metadata": dict(self.metadata),
|
|
}
|
|
|
|
@classmethod
|
|
def from_mapping(cls, value: Mapping[str, object]) -> "CaseRecord":
|
|
reference = _mapping(value, "reference")
|
|
context = _mapping(value, "context")
|
|
service_ref = value.get("service_ref")
|
|
metadata = value.get("metadata")
|
|
if metadata is None:
|
|
metadata = {}
|
|
if not isinstance(metadata, Mapping):
|
|
raise InstitutionalContextError("Case metadata must be an object.")
|
|
return cls(
|
|
reference=InstitutionalReference.from_mapping(reference),
|
|
case_number=_text(value, "case_number"),
|
|
case_type_key=_text(value, "case_type_key"),
|
|
status_key=_text(value, "status_key"),
|
|
title=_text(value, "title"),
|
|
access_mode=str(value.get("access_mode") or "tenant"),
|
|
access_grants=tuple(
|
|
CaseGrant.from_mapping(item)
|
|
for item in _items(value.get("access_grants"), "Case access grants")
|
|
),
|
|
context=GovernedContextEnvelope.from_mapping(context),
|
|
service_ref=(
|
|
InstitutionalReference.from_mapping(service_ref)
|
|
if isinstance(service_ref, Mapping)
|
|
else None
|
|
),
|
|
party_refs=_institutional_references(value.get("party_refs")),
|
|
assignment_refs=_institutional_references(
|
|
value.get("assignment_refs")
|
|
),
|
|
evidence_refs=_evidence_references(value.get("evidence_refs")),
|
|
decision_refs=_institutional_references(value.get("decision_refs")),
|
|
record_refs=_institutional_references(value.get("record_refs")),
|
|
opened_at=_datetime(value, "opened_at"),
|
|
recorded_at=_datetime(value, "recorded_at"),
|
|
deadline_at=_optional_datetime(value.get("deadline_at")),
|
|
closed_at=_optional_datetime(value.get("closed_at")),
|
|
change_reason=_text(value, "change_reason"),
|
|
metadata=dict(metadata),
|
|
)
|
|
|
|
|
|
def _validate_references(
|
|
tenant_id: str,
|
|
references: tuple[InstitutionalReference, ...],
|
|
*,
|
|
expected_kinds: set[str],
|
|
label: str,
|
|
) -> None:
|
|
if any(item.tenant_id != tenant_id for item in references):
|
|
raise InstitutionalContextError(f"{label} cannot cross tenants.")
|
|
invalid = {item.kind for item in references} - expected_kinds
|
|
if invalid:
|
|
raise InstitutionalContextError(
|
|
f"{label} contain unsupported reference kinds: "
|
|
+ ", ".join(sorted(invalid))
|
|
)
|
|
|
|
|
|
def _same_object(
|
|
left: InstitutionalReference,
|
|
right: InstitutionalReference,
|
|
) -> bool:
|
|
return (
|
|
left.kind,
|
|
left.owner_module,
|
|
left.object_id,
|
|
left.tenant_id,
|
|
) == (
|
|
right.kind,
|
|
right.owner_module,
|
|
right.object_id,
|
|
right.tenant_id,
|
|
)
|
|
|
|
|
|
def _require_aware(value: datetime | None, label: str) -> None:
|
|
if value is not None and (value.tzinfo is None or value.utcoffset() is None):
|
|
raise InstitutionalContextError(f"{label} must include a timezone.")
|
|
|
|
|
|
def _mapping(value: Mapping[str, object], key: str) -> Mapping[str, object]:
|
|
item = value.get(key)
|
|
if not isinstance(item, Mapping):
|
|
raise InstitutionalContextError(f"Case {key} must be an object.")
|
|
return item
|
|
|
|
|
|
def _text(value: Mapping[str, object], key: str) -> str:
|
|
item = str(value.get(key) or "").strip()
|
|
if not item:
|
|
raise InstitutionalContextError(f"Case {key} is required.")
|
|
return item
|
|
|
|
|
|
def _datetime(value: Mapping[str, object], key: str) -> datetime:
|
|
item = _optional_datetime(value.get(key))
|
|
if item is None:
|
|
raise InstitutionalContextError(f"Case {key} is required.")
|
|
return item
|
|
|
|
|
|
def _optional_datetime(value: object) -> datetime | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
result = value
|
|
else:
|
|
try:
|
|
result = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
|
except ValueError as exc:
|
|
raise InstitutionalContextError("Case timestamp is invalid.") from exc
|
|
_require_aware(result, "Case timestamp")
|
|
return result
|
|
|
|
|
|
def _items(value: object, label: str) -> tuple[Mapping[str, object], ...]:
|
|
if value is None:
|
|
return ()
|
|
if not isinstance(value, (list, tuple)) or any(
|
|
not isinstance(item, Mapping) for item in value
|
|
):
|
|
raise InstitutionalContextError(f"{label} must be a list of objects.")
|
|
return tuple(value) # type: ignore[return-value]
|
|
|
|
|
|
def _institutional_references(
|
|
value: object,
|
|
) -> tuple[InstitutionalReference, ...]:
|
|
return tuple(
|
|
InstitutionalReference.from_mapping(item)
|
|
for item in _items(value, "Case references")
|
|
)
|
|
|
|
|
|
def _evidence_references(value: object) -> tuple[EvidenceReference, ...]:
|
|
return tuple(
|
|
EvidenceReference.from_mapping(item)
|
|
for item in _items(value, "Case evidence")
|
|
)
|
|
|
|
|
|
__all__ = ["CaseGrant", "CaseRecord"]
|