Resolvers

Resolvers — reach PII in external systems through one interface.

Resolver

Protocol — implement these members in your own class; do not subclass.

class Resolver(Protocol):
    name: str

One external system that holds personal data (Stripe, S3, a CRM).

Implementations MUST be idempotent: erasing a subject that is already gone returns success (already_absent=True), never an error. The saga runner retries failed calls; a non-idempotent resolver would turn retries into errors or double-effects.

Implementations MUST raise ResolverError only for non-retryable failures; transient errors (timeouts, rate limits) should raise the underlying exception and let the runner retry.

Implementations MUST NOT bind event-loop-affine resources (async HTTP clients) at construction — create them inside the call. Resolver methods may be driven from different event loops: the exporter’s per-call loop and the saga runner’s loop (ADR 0006).

Fields:

• name (str): Stable, unique resolver name ("stripe"); recorded in audits.

Resolver.erase_subject

async def erase_subject(ref: SubjectRef) -> ResolverErasure

Erase the subject from the external system (Art. 17).

Args:

• ref (SubjectRef): Opaque subject reference in this resolver’s namespace.

Returns:

• ResolverErasure — The outcome; already_absent=True when there was nothing to
• ResolverErasure — erase — that is success, not an error.

Resolver.export_subject

async def export_subject(ref: SubjectRef) -> ResolverExport

Collect what the external system holds on the subject (Art. 15).

Args:

• ref (SubjectRef): Opaque subject reference in this resolver’s namespace.

Returns:

• ResolverExport — The system’s records for the subject; empty when it holds none.

ResolverErasure

class ResolverErasure(BaseModel):
    already_absent: bool = False
    detail: str | None = None
    resolver: str = Field(min_length=1)

Outcome of one external erasure call.

Idempotency contract: erasing a subject the external system no longer knows is success (already_absent=True), never an error — saga retries depend on this.

Fields:

• already_absent (bool): The subject was already gone — still success.
• detail (str | None): Short human-readable note for the audit trail (no PII).
• resolver (str): Name of the resolver that performed the erasure.

ResolverExport

class ResolverExport(BaseModel):
    records: tuple[ExportRecord, ...] = ()
    resolver: str = Field(min_length=1)

What one external system holds on a subject.

Fields:

• records (tuple[ExportRecord, ...]): The exported values with their metadata.
• resolver (str): Name of the resolver that produced this.

ResolverRegistry

class ResolverRegistry:
    def __init__() -> None

Holds every resolver an application has wired in.

Registration is explicit on purpose — no auto-discovery, no entry-point magic. The registry doubles as the auditable declaration of every place this application holds PII outside its own database.

ResolverRegistry.all

def all() -> tuple[Resolver, ...]

Every registered resolver, in registration order.

ResolverRegistry.get

def get(name: str) -> Resolver

Return one resolver by name.

Args:

• name (str): The resolver’s stable name.

Raises:

• ResolverError — If no resolver with that name is registered.

ResolverRegistry.register

def register(resolver: Resolver) -> None

Add one resolver.

Args:

• resolver (Resolver): The resolver to add.

Raises:

• ResolverError — If a resolver with the same name is already registered — silent replacement would falsify the audit picture.