feat: implement institutional service catalogue
This commit is contained in:
@@ -0,0 +1,360 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any, Mapping, Sequence
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from govoplan_core.core.institutional import (
|
||||
InstitutionalContextError,
|
||||
InstitutionalReference,
|
||||
ServiceAvailabilityAssessment,
|
||||
ServiceDefinition,
|
||||
TemporalRevision,
|
||||
)
|
||||
from govoplan_services.backend.db.models import ServiceDefinitionRevision
|
||||
|
||||
|
||||
_PUBLICATION_TRANSITIONS: dict[str, frozenset[str]] = {
|
||||
"draft": frozenset({"draft", "published", "retired"}),
|
||||
"published": frozenset({"published", "suspended", "retired"}),
|
||||
"suspended": frozenset({"suspended", "published", "retired"}),
|
||||
"retired": frozenset(),
|
||||
}
|
||||
|
||||
|
||||
class ServiceStoreError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def record_service_definition(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
definition: ServiceDefinition,
|
||||
expected_revision: str | None = None,
|
||||
) -> ServiceDefinition:
|
||||
tenant_id = _principal_tenant(principal)
|
||||
_validate_definition(definition, tenant_id=tenant_id)
|
||||
payload = definition.to_dict(include_inspection=True)
|
||||
replay = (
|
||||
session.query(ServiceDefinitionRevision)
|
||||
.filter(
|
||||
ServiceDefinitionRevision.tenant_id == tenant_id,
|
||||
ServiceDefinitionRevision.service_id == definition.reference.object_id,
|
||||
ServiceDefinitionRevision.revision == definition.temporal.revision,
|
||||
)
|
||||
.one_or_none()
|
||||
)
|
||||
if replay is not None:
|
||||
if replay.payload != payload:
|
||||
raise ServiceStoreError(
|
||||
"A different Service payload already uses this revision."
|
||||
)
|
||||
return _definition_from_row(replay)
|
||||
|
||||
current = _current_row(
|
||||
session,
|
||||
tenant_id=tenant_id,
|
||||
service_id=definition.reference.object_id,
|
||||
lock=True,
|
||||
)
|
||||
_validate_temporal(definition.temporal)
|
||||
if current is None:
|
||||
if expected_revision is not None:
|
||||
raise ServiceStoreError(
|
||||
"Service revision conflict: no current revision exists."
|
||||
)
|
||||
collision = (
|
||||
session.query(ServiceDefinitionRevision.id)
|
||||
.filter(
|
||||
ServiceDefinitionRevision.tenant_id == tenant_id,
|
||||
ServiceDefinitionRevision.service_key == definition.key,
|
||||
ServiceDefinitionRevision.superseded_at.is_(None),
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if collision is not None:
|
||||
raise ServiceStoreError("Service key is already in use in this tenant.")
|
||||
else:
|
||||
current_definition = _definition_from_row(current)
|
||||
if expected_revision != current.revision:
|
||||
raise ServiceStoreError(
|
||||
"Service revision conflict: the expected revision is stale."
|
||||
)
|
||||
if definition.key != current_definition.key:
|
||||
raise ServiceStoreError("A Service key cannot change across revisions.")
|
||||
if definition.publication_state not in _PUBLICATION_TRANSITIONS[current.publication_state]:
|
||||
raise ServiceStoreError(
|
||||
f"Service publication transition {current.publication_state!r} to "
|
||||
f"{definition.publication_state!r} is not allowed."
|
||||
)
|
||||
current.superseded_at = definition.temporal.recorded_at
|
||||
|
||||
row = ServiceDefinitionRevision(
|
||||
tenant_id=tenant_id,
|
||||
service_id=definition.reference.object_id,
|
||||
service_key=definition.key,
|
||||
revision=definition.temporal.revision,
|
||||
previous_revision_id=current.id if current is not None else None,
|
||||
publication_state=definition.publication_state,
|
||||
valid_from=definition.temporal.valid_from,
|
||||
valid_to=definition.temporal.valid_to,
|
||||
recorded_at=_recorded_at(definition.temporal),
|
||||
search_text=f"{definition.key} {definition.title}".casefold(),
|
||||
payload=payload,
|
||||
created_by=_principal_actor(principal),
|
||||
)
|
||||
session.add(row)
|
||||
session.flush()
|
||||
return _definition_from_row(row)
|
||||
|
||||
|
||||
def get_service_definition(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
service_id: str,
|
||||
revision: str | None = None,
|
||||
) -> ServiceDefinition | None:
|
||||
tenant_id = _principal_tenant(principal)
|
||||
query = session.query(ServiceDefinitionRevision).filter(
|
||||
ServiceDefinitionRevision.tenant_id == tenant_id,
|
||||
ServiceDefinitionRevision.service_id == service_id,
|
||||
)
|
||||
if revision is None:
|
||||
query = query.filter(ServiceDefinitionRevision.superseded_at.is_(None))
|
||||
else:
|
||||
query = query.filter(ServiceDefinitionRevision.revision == revision)
|
||||
row = query.order_by(ServiceDefinitionRevision.recorded_at.desc()).first()
|
||||
return _definition_from_row(row) if row is not None else None
|
||||
|
||||
|
||||
def list_service_definitions(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
query: str = "",
|
||||
publication_states: Sequence[str] | None = None,
|
||||
limit: int = 100,
|
||||
) -> tuple[ServiceDefinition, ...]:
|
||||
tenant_id = _principal_tenant(principal)
|
||||
if not 1 <= limit <= 200:
|
||||
raise ServiceStoreError("Service list limit must be between 1 and 200.")
|
||||
statement = session.query(ServiceDefinitionRevision).filter(
|
||||
ServiceDefinitionRevision.tenant_id == tenant_id,
|
||||
ServiceDefinitionRevision.superseded_at.is_(None),
|
||||
)
|
||||
if publication_states is not None:
|
||||
statement = statement.filter(
|
||||
ServiceDefinitionRevision.publication_state.in_(tuple(publication_states))
|
||||
)
|
||||
clean_query = query.strip().casefold()
|
||||
if clean_query:
|
||||
statement = statement.filter(
|
||||
ServiceDefinitionRevision.search_text.contains(clean_query)
|
||||
)
|
||||
rows = statement.order_by(
|
||||
ServiceDefinitionRevision.service_key.asc(),
|
||||
ServiceDefinitionRevision.recorded_at.desc(),
|
||||
).limit(limit).all()
|
||||
return tuple(_definition_from_row(row) for row in rows)
|
||||
|
||||
|
||||
class SqlServiceDefinitionProvider:
|
||||
def get_service_definition(
|
||||
self,
|
||||
session: object,
|
||||
principal: object,
|
||||
*,
|
||||
reference: InstitutionalReference,
|
||||
effective_at: datetime | None = None,
|
||||
) -> ServiceDefinition | None:
|
||||
tenant_id = _principal_tenant(principal)
|
||||
if reference.kind != "service" or reference.tenant_id != tenant_id:
|
||||
raise InstitutionalContextError(
|
||||
"Service definition lookup requires a same-tenant Service reference."
|
||||
)
|
||||
item = get_service_definition(
|
||||
_session(session),
|
||||
principal,
|
||||
service_id=reference.object_id,
|
||||
revision=reference.version,
|
||||
)
|
||||
if item is None or (
|
||||
effective_at is not None and not item.temporal.effective_at(effective_at)
|
||||
):
|
||||
return None
|
||||
return item
|
||||
|
||||
def list_service_definitions(
|
||||
self,
|
||||
session: object,
|
||||
principal: object,
|
||||
*,
|
||||
tenant_id: str,
|
||||
query: str = "",
|
||||
limit: int = 100,
|
||||
) -> Sequence[ServiceDefinition]:
|
||||
if tenant_id != _principal_tenant(principal):
|
||||
raise InstitutionalContextError(
|
||||
"Service catalogue lookup cannot cross tenants."
|
||||
)
|
||||
return list_service_definitions(
|
||||
_session(session),
|
||||
principal,
|
||||
query=query,
|
||||
publication_states=("published", "suspended"),
|
||||
limit=limit,
|
||||
)
|
||||
|
||||
|
||||
class RegistryServiceAvailabilityEvaluator:
|
||||
def __init__(self, registry: object | None) -> None:
|
||||
self._registry = registry
|
||||
|
||||
def evaluate_service_availability(
|
||||
self,
|
||||
session: object,
|
||||
principal: object,
|
||||
*,
|
||||
definition: ServiceDefinition,
|
||||
effective_at: datetime,
|
||||
) -> ServiceAvailabilityAssessment:
|
||||
del session
|
||||
tenant_id = _principal_tenant(principal)
|
||||
if definition.reference.tenant_id != tenant_id:
|
||||
raise InstitutionalContextError(
|
||||
"Service availability cannot be evaluated across tenants."
|
||||
)
|
||||
states: dict[str, bool] = {}
|
||||
reasons: list[str] = []
|
||||
if not definition.temporal.effective_at(effective_at):
|
||||
reasons.append("service.outside_effective_interval")
|
||||
if definition.publication_state != "published":
|
||||
reasons.append(f"service.publication.{definition.publication_state}")
|
||||
for requirement in definition.availability_requirements:
|
||||
if requirement.kind == "module":
|
||||
states[requirement.key] = _registry_has(self._registry, requirement.reference)
|
||||
elif requirement.kind == "capability":
|
||||
states[requirement.key] = _registry_has_capability(
|
||||
self._registry,
|
||||
requirement.reference,
|
||||
)
|
||||
return ServiceAvailabilityAssessment(
|
||||
requirement_states=states,
|
||||
reason_codes=tuple(reasons),
|
||||
)
|
||||
|
||||
|
||||
def definition_from_mapping(value: Mapping[str, object]) -> ServiceDefinition:
|
||||
try:
|
||||
return ServiceDefinition.from_mapping(value)
|
||||
except InstitutionalContextError as exc:
|
||||
raise ServiceStoreError(str(exc)) from exc
|
||||
|
||||
|
||||
def _current_row(
|
||||
session: Session,
|
||||
*,
|
||||
tenant_id: str,
|
||||
service_id: str,
|
||||
lock: bool,
|
||||
) -> ServiceDefinitionRevision | None:
|
||||
query = session.query(ServiceDefinitionRevision).filter(
|
||||
ServiceDefinitionRevision.tenant_id == tenant_id,
|
||||
ServiceDefinitionRevision.service_id == service_id,
|
||||
ServiceDefinitionRevision.superseded_at.is_(None),
|
||||
)
|
||||
if lock:
|
||||
query = query.with_for_update()
|
||||
return query.one_or_none()
|
||||
|
||||
|
||||
def _definition_from_row(row: ServiceDefinitionRevision) -> ServiceDefinition:
|
||||
payload: dict[str, Any] = dict(row.payload)
|
||||
temporal = dict(payload.get("temporal") or {})
|
||||
temporal["superseded_at"] = _datetime_text(row.superseded_at)
|
||||
payload["temporal"] = temporal
|
||||
return ServiceDefinition.from_mapping(payload)
|
||||
|
||||
|
||||
def _validate_definition(definition: ServiceDefinition, *, tenant_id: str) -> None:
|
||||
if definition.reference.owner_module != "services":
|
||||
raise ServiceStoreError("Service definitions must be owned by Services.")
|
||||
if definition.reference.tenant_id != tenant_id:
|
||||
raise ServiceStoreError("Service definitions cannot cross tenants.")
|
||||
if definition.reference.version != definition.temporal.revision:
|
||||
raise ServiceStoreError(
|
||||
"Service reference version must match its temporal revision."
|
||||
)
|
||||
if definition.temporal.superseded_at is not None:
|
||||
raise ServiceStoreError("Clients cannot set Service superseded_at.")
|
||||
|
||||
|
||||
def _validate_temporal(temporal: TemporalRevision) -> None:
|
||||
_recorded_at(temporal)
|
||||
if not str(temporal.change_reason or "").strip():
|
||||
raise ServiceStoreError(
|
||||
"A Service revision requires recorded_at and change_reason."
|
||||
)
|
||||
|
||||
|
||||
def _recorded_at(temporal: TemporalRevision) -> datetime:
|
||||
if temporal.recorded_at is None:
|
||||
raise ServiceStoreError("A Service revision requires recorded_at.")
|
||||
return temporal.recorded_at
|
||||
|
||||
|
||||
def _principal_tenant(principal: object) -> str:
|
||||
tenant_id = str(getattr(principal, "tenant_id", "") or "").strip()
|
||||
if not tenant_id:
|
||||
raise InstitutionalContextError(
|
||||
"Service operations require a tenant-bound principal."
|
||||
)
|
||||
return tenant_id
|
||||
|
||||
|
||||
def _principal_actor(principal: object) -> str | None:
|
||||
for name in ("account_id", "identity_id", "membership_id"):
|
||||
value = str(getattr(principal, name, "") or "").strip()
|
||||
if value:
|
||||
return value
|
||||
return None
|
||||
|
||||
|
||||
def _session(value: object) -> Session:
|
||||
if not hasattr(value, "query"):
|
||||
raise InstitutionalContextError("Service provider requires a database session.")
|
||||
return value # type: ignore[return-value]
|
||||
|
||||
|
||||
def _registry_has(registry: object | None, module_id: str) -> bool:
|
||||
return bool(registry is not None and hasattr(registry, "has") and registry.has(module_id))
|
||||
|
||||
|
||||
def _registry_has_capability(registry: object | None, capability: str) -> bool:
|
||||
return bool(
|
||||
registry is not None
|
||||
and hasattr(registry, "has_capability")
|
||||
and registry.has_capability(capability)
|
||||
)
|
||||
|
||||
|
||||
def _datetime_text(value: datetime | None) -> str | None:
|
||||
if value is None:
|
||||
return None
|
||||
if value.tzinfo is None:
|
||||
value = value.replace(tzinfo=UTC)
|
||||
return value.isoformat()
|
||||
|
||||
|
||||
__all__ = [
|
||||
"RegistryServiceAvailabilityEvaluator",
|
||||
"ServiceStoreError",
|
||||
"SqlServiceDefinitionProvider",
|
||||
"definition_from_mapping",
|
||||
"get_service_definition",
|
||||
"list_service_definitions",
|
||||
"record_service_definition",
|
||||
]
|
||||
Reference in New Issue
Block a user