100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
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 {}),
|
|
)
|