357 lines
11 KiB
Python
357 lines
11 KiB
Python
"""JSONB-oriented durable store record serialization."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Mapping as MappingABC
|
|
from dataclasses import dataclass, fields, is_dataclass
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
from types import NoneType, UnionType
|
|
from typing import Any, Callable, Mapping, Union, get_args, get_origin, get_type_hints
|
|
|
|
from user_engine.domain import (
|
|
Account,
|
|
AccessProfile,
|
|
ActiveAccessContext,
|
|
Application,
|
|
ApplicationBinding,
|
|
AuditRecord,
|
|
Catalog,
|
|
ExternalIdentity,
|
|
FamilyInvitation,
|
|
IdentityFactor,
|
|
Membership,
|
|
OnboardingJourney,
|
|
OutboxEvent,
|
|
PreparedAccount,
|
|
ProfileValue,
|
|
RegistrationSession,
|
|
TenantAccount,
|
|
User,
|
|
WelcomeProtocol,
|
|
)
|
|
from user_engine.migrations import USER_ENGINE_STORE_RECORD_TYPES
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class StoreRecord:
|
|
"""One generic durable-store row for a domain object payload."""
|
|
|
|
record_type: str
|
|
record_key: str
|
|
payload: Mapping[str, Any]
|
|
tenant: str | None = None
|
|
user_id: str | None = None
|
|
application_id: str | None = None
|
|
scope_type: str | None = None
|
|
scope_id: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class StoreRecordCodec:
|
|
"""Codec rule for one user-engine store record type."""
|
|
|
|
record_type: str
|
|
model_type: type[Any]
|
|
record_key: Callable[[Any], str]
|
|
metadata: Callable[[Any], Mapping[str, str | None]]
|
|
|
|
|
|
def store_record_for(value: Any) -> StoreRecord:
|
|
"""Return a generic durable-store record for a supported domain object."""
|
|
codec = _CODECS_BY_MODEL.get(type(value))
|
|
if codec is None:
|
|
raise TypeError(f"unsupported store record type: {type(value).__name__}")
|
|
|
|
metadata = dict(codec.metadata(value))
|
|
return StoreRecord(
|
|
record_type=codec.record_type,
|
|
record_key=codec.record_key(value),
|
|
payload=_encode_dataclass(value),
|
|
tenant=metadata.get("tenant"),
|
|
user_id=metadata.get("user_id"),
|
|
application_id=metadata.get("application_id"),
|
|
scope_type=metadata.get("scope_type"),
|
|
scope_id=metadata.get("scope_id"),
|
|
)
|
|
|
|
|
|
def composite_record_key(*parts: str | None) -> str:
|
|
"""Return the deterministic composite key used by durable store records."""
|
|
return _composite_key(*parts)
|
|
|
|
|
|
def domain_record_from_store_record(record: StoreRecord) -> Any:
|
|
"""Decode a durable-store record payload into its domain dataclass."""
|
|
codec = _CODECS_BY_RECORD_TYPE.get(record.record_type)
|
|
if codec is None:
|
|
raise ValueError(f"unsupported store record type: {record.record_type}")
|
|
return _decode_dataclass(codec.model_type, record.payload)
|
|
|
|
|
|
def validate_store_record_codecs() -> tuple[str, ...]:
|
|
"""Return codec coverage errors against the durable-store manifest."""
|
|
errors: list[str] = []
|
|
manifest_types = set(USER_ENGINE_STORE_RECORD_TYPES)
|
|
codec_types = set(_CODECS_BY_RECORD_TYPE)
|
|
missing = sorted(manifest_types - codec_types)
|
|
extra = sorted(codec_types - manifest_types)
|
|
if missing:
|
|
errors.append(f"missing codecs for: {', '.join(missing)}")
|
|
if extra:
|
|
errors.append(f"extra codecs for: {', '.join(extra)}")
|
|
return tuple(errors)
|
|
|
|
|
|
def _encode_dataclass(value: Any) -> Mapping[str, Any]:
|
|
if not is_dataclass(value):
|
|
raise TypeError(f"expected dataclass, got {type(value).__name__}")
|
|
return {
|
|
field.name: _encode_value(getattr(value, field.name))
|
|
for field in fields(value)
|
|
}
|
|
|
|
|
|
def _encode_value(value: Any) -> Any:
|
|
if is_dataclass(value):
|
|
return _encode_dataclass(value)
|
|
if isinstance(value, datetime):
|
|
return value.isoformat()
|
|
if isinstance(value, Enum):
|
|
return value.value
|
|
if isinstance(value, tuple):
|
|
return [_encode_value(item) for item in value]
|
|
if isinstance(value, list):
|
|
return [_encode_value(item) for item in value]
|
|
if isinstance(value, MappingABC):
|
|
return {str(key): _encode_value(item) for key, item in value.items()}
|
|
return value
|
|
|
|
|
|
def _decode_dataclass(model_type: type[Any], payload: Mapping[str, Any]) -> Any:
|
|
hints = get_type_hints(model_type)
|
|
kwargs = {
|
|
field.name: _decode_value(payload[field.name], hints[field.name])
|
|
for field in fields(model_type)
|
|
if field.name in payload
|
|
}
|
|
return model_type(**kwargs)
|
|
|
|
|
|
def _decode_value(value: Any, type_hint: Any) -> Any:
|
|
if value is None:
|
|
return None
|
|
if type_hint is Any:
|
|
return value
|
|
origin = get_origin(type_hint)
|
|
args = get_args(type_hint)
|
|
|
|
if origin in (UnionType, Union):
|
|
non_none_args = tuple(arg for arg in args if arg is not NoneType)
|
|
if len(non_none_args) == 1 and len(non_none_args) != len(args):
|
|
return _decode_value(value, non_none_args[0])
|
|
for arg in non_none_args:
|
|
try:
|
|
return _decode_value(value, arg)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return value
|
|
|
|
if type_hint is datetime:
|
|
return datetime.fromisoformat(str(value))
|
|
if isinstance(type_hint, type) and issubclass(type_hint, Enum):
|
|
return type_hint(value)
|
|
if isinstance(type_hint, type) and is_dataclass(type_hint):
|
|
return _decode_dataclass(type_hint, value)
|
|
if origin is tuple:
|
|
if not args:
|
|
return tuple(value)
|
|
item_hint = args[0] if len(args) == 2 and args[1] is Ellipsis else None
|
|
if item_hint is not None:
|
|
return tuple(_decode_value(item, item_hint) for item in value)
|
|
return tuple(
|
|
_decode_value(item, args[index])
|
|
for index, item in enumerate(value)
|
|
)
|
|
if origin is list:
|
|
item_hint = args[0] if args else Any
|
|
return [_decode_value(item, item_hint) for item in value]
|
|
if origin in (dict, Mapping, MappingABC):
|
|
return dict(value)
|
|
if type_hint in (str, int, float, bool):
|
|
return type_hint(value)
|
|
return value
|
|
|
|
|
|
def _single_key(value: str) -> str:
|
|
return value
|
|
|
|
|
|
def _composite_key(*parts: str | None) -> str:
|
|
return json.dumps(list(parts), separators=(",", ":"), ensure_ascii=True)
|
|
|
|
|
|
def _enum_value(value: Any) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, Enum):
|
|
return str(value.value)
|
|
return str(value)
|
|
|
|
|
|
_CODECS = (
|
|
StoreRecordCodec(
|
|
"users",
|
|
User,
|
|
lambda value: _single_key(value.user_id),
|
|
lambda value: {"user_id": value.user_id},
|
|
),
|
|
StoreRecordCodec(
|
|
"accounts",
|
|
Account,
|
|
lambda value: _single_key(value.user_id),
|
|
lambda value: {"user_id": value.user_id},
|
|
),
|
|
StoreRecordCodec(
|
|
"external_identities",
|
|
ExternalIdentity,
|
|
lambda value: _composite_key(value.issuer, value.subject),
|
|
lambda value: {"user_id": value.user_id},
|
|
),
|
|
StoreRecordCodec(
|
|
"tenant_accounts",
|
|
TenantAccount,
|
|
lambda value: _composite_key(value.tenant, value.user_id),
|
|
lambda value: {"tenant": value.tenant, "user_id": value.user_id},
|
|
),
|
|
StoreRecordCodec(
|
|
"memberships",
|
|
Membership,
|
|
lambda value: _single_key(value.membership_id),
|
|
lambda value: {
|
|
"tenant": value.tenant,
|
|
"user_id": value.user_id,
|
|
"scope_type": value.scope_type,
|
|
"scope_id": value.scope_id,
|
|
},
|
|
),
|
|
StoreRecordCodec(
|
|
"applications",
|
|
Application,
|
|
lambda value: _single_key(value.application_id),
|
|
lambda value: {"application_id": value.application_id},
|
|
),
|
|
StoreRecordCodec(
|
|
"application_bindings",
|
|
ApplicationBinding,
|
|
lambda value: _single_key(value.application_id),
|
|
lambda value: {"application_id": value.application_id},
|
|
),
|
|
StoreRecordCodec(
|
|
"catalogs",
|
|
Catalog,
|
|
lambda value: _single_key(value.catalog_id),
|
|
lambda value: {"application_id": value.owning_application_id},
|
|
),
|
|
StoreRecordCodec(
|
|
"family_invitations",
|
|
FamilyInvitation,
|
|
lambda value: _single_key(value.invitation_id),
|
|
lambda value: {
|
|
"tenant": value.tenant,
|
|
"user_id": value.user_id,
|
|
"application_id": value.application_id,
|
|
"scope_type": "family",
|
|
"scope_id": value.family_scope_id,
|
|
},
|
|
),
|
|
StoreRecordCodec(
|
|
"registration_sessions",
|
|
RegistrationSession,
|
|
lambda value: _single_key(value.registration_id),
|
|
lambda value: {"tenant": value.tenant, "user_id": value.user_id},
|
|
),
|
|
StoreRecordCodec(
|
|
"identity_factors",
|
|
IdentityFactor,
|
|
lambda value: _single_key(value.factor_id),
|
|
lambda value: {
|
|
"user_id": value.user_id,
|
|
"scope_type": "registration" if value.registration_id else None,
|
|
"scope_id": value.registration_id,
|
|
},
|
|
),
|
|
StoreRecordCodec(
|
|
"prepared_accounts",
|
|
PreparedAccount,
|
|
lambda value: _single_key(value.prepared_account_id),
|
|
lambda value: {"tenant": value.tenant},
|
|
),
|
|
StoreRecordCodec(
|
|
"access_profiles",
|
|
AccessProfile,
|
|
lambda value: _single_key(value.access_profile_id),
|
|
lambda value: {
|
|
"tenant": value.tenant,
|
|
"scope_type": _enum_value(value.scope_type),
|
|
"scope_id": value.scope_id,
|
|
},
|
|
),
|
|
StoreRecordCodec(
|
|
"active_access_contexts",
|
|
ActiveAccessContext,
|
|
lambda value: _composite_key(value.user_id, value.tenant),
|
|
lambda value: {
|
|
"tenant": value.tenant,
|
|
"user_id": value.user_id,
|
|
"scope_type": _enum_value(value.scope_type),
|
|
"scope_id": value.scope_id,
|
|
},
|
|
),
|
|
StoreRecordCodec(
|
|
"welcome_protocols",
|
|
WelcomeProtocol,
|
|
lambda value: _single_key(value.protocol_id),
|
|
lambda value: {
|
|
"tenant": value.tenant,
|
|
"application_id": value.application_id,
|
|
},
|
|
),
|
|
StoreRecordCodec(
|
|
"onboarding_journeys",
|
|
OnboardingJourney,
|
|
lambda value: _single_key(value.journey_id),
|
|
lambda value: {"tenant": value.tenant, "user_id": value.user_id},
|
|
),
|
|
StoreRecordCodec(
|
|
"profile_values",
|
|
ProfileValue,
|
|
lambda value: _composite_key(
|
|
value.user_id,
|
|
value.attribute_key,
|
|
_enum_value(value.scope),
|
|
value.scope_id,
|
|
),
|
|
lambda value: {
|
|
"user_id": value.user_id,
|
|
"scope_type": _enum_value(value.scope),
|
|
"scope_id": value.scope_id,
|
|
},
|
|
),
|
|
StoreRecordCodec(
|
|
"audit_records",
|
|
AuditRecord,
|
|
lambda value: _single_key(value.audit_id),
|
|
lambda value: {"tenant": value.tenant, "user_id": value.subject},
|
|
),
|
|
StoreRecordCodec(
|
|
"outbox_events",
|
|
OutboxEvent,
|
|
lambda value: _single_key(value.event_id),
|
|
lambda value: {"tenant": value.tenant},
|
|
),
|
|
)
|
|
|
|
_CODECS_BY_RECORD_TYPE = {codec.record_type: codec for codec in _CODECS}
|
|
_CODECS_BY_MODEL = {codec.model_type: codec for codec in _CODECS}
|