84 lines
2.6 KiB
Python
84 lines
2.6 KiB
Python
|
"""Implementation ports for user-engine adapters.
|
||
|
|
|
||
|
|
The domain layer should depend on these protocols. Concrete implementations
|
||
|
|
can be local test adapters, HTTP clients, database-backed stores, or platform
|
||
|
|
adapters without changing domain code.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import Any, Iterable, Mapping, Protocol
|
||
|
|
|
||
|
|
from user_engine.domain import (
|
||
|
|
Actor,
|
||
|
|
ApplicationBinding,
|
||
|
|
AuditRecord,
|
||
|
|
AuthorizationDecision,
|
||
|
|
AuthorizationRequest,
|
||
|
|
Membership,
|
||
|
|
OutboxEvent,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class IdentityClaimsAdapter(Protocol):
|
||
|
|
"""Normalize verified identity claims into a user-engine actor."""
|
||
|
|
|
||
|
|
def normalize(self, claims: Mapping[str, Any]) -> Actor:
|
||
|
|
"""Return a normalized actor from already-verified claims."""
|
||
|
|
|
||
|
|
def identity_key(self, actor: Actor) -> tuple[str, str]:
|
||
|
|
"""Return the stable external identity link key."""
|
||
|
|
|
||
|
|
|
||
|
|
class AuthorizationCheckPort(Protocol):
|
||
|
|
"""Ask whether an actor may perform an action."""
|
||
|
|
|
||
|
|
def check(self, request: AuthorizationRequest) -> AuthorizationDecision:
|
||
|
|
"""Return the authorization decision for one request."""
|
||
|
|
|
||
|
|
def batch_check(
|
||
|
|
self, requests: Iterable[AuthorizationRequest]
|
||
|
|
) -> tuple[AuthorizationDecision, ...]:
|
||
|
|
"""Return decisions in request order."""
|
||
|
|
|
||
|
|
|
||
|
|
class ApplicationBindingStore(Protocol):
|
||
|
|
"""Store links between user-engine applications and external systems."""
|
||
|
|
|
||
|
|
def get(self, application_id: str) -> ApplicationBinding | None:
|
||
|
|
"""Return a binding by user-engine application id."""
|
||
|
|
|
||
|
|
def save(self, binding: ApplicationBinding) -> None:
|
||
|
|
"""Create or replace an application binding."""
|
||
|
|
|
||
|
|
|
||
|
|
class MembershipFactExporter(Protocol):
|
||
|
|
"""Export membership facts as read models for authorization systems."""
|
||
|
|
|
||
|
|
def export(self, memberships: Iterable[Membership]) -> Mapping[str, Any]:
|
||
|
|
"""Return an adapter-neutral membership fact manifest."""
|
||
|
|
|
||
|
|
|
||
|
|
class EventOutbox(Protocol):
|
||
|
|
"""Persist and publish durable domain events."""
|
||
|
|
|
||
|
|
def append(self, event: OutboxEvent) -> None:
|
||
|
|
"""Append an event in the same unit of work as its mutation."""
|
||
|
|
|
||
|
|
def pending(self) -> tuple[OutboxEvent, ...]:
|
||
|
|
"""Return events waiting for delivery."""
|
||
|
|
|
||
|
|
|
||
|
|
class AuditWriter(Protocol):
|
||
|
|
"""Persist local audit records and support platform audit export."""
|
||
|
|
|
||
|
|
def record(self, audit_record: AuditRecord) -> None:
|
||
|
|
"""Persist an audit record."""
|
||
|
|
|
||
|
|
|
||
|
|
class SecretProvider(Protocol):
|
||
|
|
"""Load runtime secret material from the active environment."""
|
||
|
|
|
||
|
|
def get(self, name: str) -> str:
|
||
|
|
"""Return a secret value by logical name."""
|