feat: implement governed cases workspace
This commit is contained in:
@@ -0,0 +1,364 @@
|
||||
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"]
|
||||
Reference in New Issue
Block a user