Define tenant lifecycle event contract

2026-07-11 00:46:09 +02:00
parent 59f3f22f7a
commit cd7cc674c7
4 changed files with 290 additions and 2 deletions
@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Literal
+
+TenantLifecyclePhase = Literal[
+    "created",
+    "suspended",
+    "resumed",
+    "deletion_requested",
+    "erasure_completed",
+]
+
+TENANT_EVENT_CREATED = "tenant.created"
+TENANT_EVENT_SUSPENDED = "tenant.suspended"
+TENANT_EVENT_RESUMED = "tenant.resumed"
+TENANT_EVENT_DELETION_REQUESTED = "tenant.deletion_requested"
+TENANT_EVENT_ERASURE_COMPLETED = "tenant.erasure_completed"
+
+TENANT_LIFECYCLE_EVENTS: tuple[str, ...] = (
+    TENANT_EVENT_CREATED,
+    TENANT_EVENT_SUSPENDED,
+    TENANT_EVENT_RESUMED,
+    TENANT_EVENT_DELETION_REQUESTED,
+    TENANT_EVENT_ERASURE_COMPLETED,
+)
+
+_EVENT_BY_PHASE: Mapping[TenantLifecyclePhase, str] = {
+    "created": TENANT_EVENT_CREATED,
+    "suspended": TENANT_EVENT_SUSPENDED,
+    "resumed": TENANT_EVENT_RESUMED,
+    "deletion_requested": TENANT_EVENT_DELETION_REQUESTED,
+    "erasure_completed": TENANT_EVENT_ERASURE_COMPLETED,
+}
+
+
+@dataclass(frozen=True, slots=True)
+class TenantLifecycleEvent:
+    event_type: str
+    tenant_id: str
+    tenant_slug: str | None = None
+    tenant_name: str | None = None
+    actor_account_id: str | None = None
+    requested_by_account_id: str | None = None
+    reason: str | None = None
+    counts: Mapping[str, int] = field(default_factory=dict)
+    occurred_at: datetime | None = None
+    details: Mapping[str, object] = field(default_factory=dict)
+
+    def audit_details(self) -> dict[str, object]:
+        payload: dict[str, object] = dict(self.details)
+        if self.tenant_slug is not None:
+            payload["slug"] = self.tenant_slug
+        if self.tenant_name is not None:
+            payload["name"] = self.tenant_name
+        if self.actor_account_id is not None:
+            payload["actor_account_id"] = self.actor_account_id
+        if self.requested_by_account_id is not None:
+            payload["requested_by_account_id"] = self.requested_by_account_id
+        if self.reason is not None:
+            payload["reason"] = self.reason
+        if self.counts:
+            payload["counts"] = dict(self.counts)
+        if self.occurred_at is not None:
+            payload["occurred_at"] = self.occurred_at.isoformat()
+        return payload
+
+
+def tenant_lifecycle_event_type(phase: TenantLifecyclePhase) -> str:
+    return _EVENT_BY_PHASE[phase]
+
+
+def tenant_lifecycle_event(
+    phase: TenantLifecyclePhase,
+    *,
+    tenant_id: str,
+    tenant_slug: str | None = None,
+    tenant_name: str | None = None,
+    actor_account_id: str | None = None,
+    requested_by_account_id: str | None = None,
+    reason: str | None = None,
+    counts: Mapping[str, int] | None = None,
+    occurred_at: datetime | None = None,
+    details: Mapping[str, object] | None = None,
+) -> TenantLifecycleEvent:
+    return TenantLifecycleEvent(
+        event_type=tenant_lifecycle_event_type(phase),
+        tenant_id=tenant_id,
+        tenant_slug=tenant_slug,
+        tenant_name=tenant_name,
+        actor_account_id=actor_account_id,
+        requested_by_account_id=requested_by_account_id,
+        reason=reason,
+        counts=dict(counts or {}),
+        occurred_at=occurred_at,
+        details=dict(details or {}),
+    )