fraudtwin.contracts.registry

Versioned Avro contract validation and mapping.

Status: Stable

Classes

fraudtwin.contracts.registry.AvroContractRegistry	Parsed source-controlled registry and its Avro schemas.
fraudtwin.contracts.registry.AvroDatumMapper	Map domain records to and from one loaded observable contract registry.
fraudtwin.contracts.registry.RegistryReport	Summary returned after validating a complete registry.

Exceptions

fraudtwin.contracts.registry.ContractValidationError

Raised when registry metadata or an Avro contract is invalid.

Functions

fraudtwin.contracts.registry.contract_registry	Load and validate a registry in one call.
fraudtwin.contracts.registry.default_registry_path	Return the bundled registry path in a checkout or installed wheel.
fraudtwin.contracts.registry.load_contract_registry	Load registry metadata and parse every declared Avro schema.
fraudtwin.contracts.registry.validate_avro_datum	Validate one mapped datum with Apache Avro's implementation.

Detailed API

Avro contract loading, compatibility checking, and observable mapping.

The M8 fraudtwin.schema helpers intentionally remain separate: those helpers model data-quality schema faults, while this module validates the clean observable contracts that a future M25 producer will publish.

class fraudtwin.contracts.registry.AvroContractRegistry(root, compatibility, subjects)

Bases: object

Parsed source-controlled registry and its Avro schemas.

Parameters:

• root (Path)

• compatibility (str)

• subjects (tuple[ContractSubject, ...])

mapper()

Return the explicit datum mapper for this registry.

Return type:

AvroDatumMapper

validate()

Validate metadata, canonical fingerprints, and full compatibility.

Return type:

RegistryReport

datum(subject, record)

Convert a domain record to a clean observable Avro datum.

Latent truth fields are never copied, even when present on the source model. Values are converted to Avro logical Python values (datetime and Decimal) without mutating the source model.

Return type:

dict[str, Any]

Parameters:

• subject (str)

• record (BaseModel | Mapping[str, Any])

encode(subject, record)

Encode one observable datum using the latest subject schema.

Return type:

bytes

Parameters:

• subject (str)

• record (BaseModel | Mapping[str, Any])

decode(subject, payload)

Decode one payload using the latest subject schema.

Return type:

dict[str, Any]

Parameters:

• subject (str)

• payload (bytes)

class fraudtwin.contracts.registry.AvroDatumMapper(registry)

Bases: object

Map domain records to and from one loaded observable contract registry.

Parameters:

registry (AvroContractRegistry)

to_datum(subject, record)

Return a validated observable Avro datum for subject.

Return type:

dict[str, Any]

Parameters:

• subject (str)

• record (BaseModel | Mapping[str, Any])

encode(subject, record)

Encode a domain record using the latest subject schema.

Return type:

bytes

Parameters:

• subject (str)

• record (BaseModel | Mapping[str, Any])

decode(subject, payload)

Decode one payload using the latest subject schema.

Return type:

dict[str, Any]

Parameters:

• subject (str)

• payload (bytes)

exception fraudtwin.contracts.registry.ContractValidationError

Bases: ValueError

Raised when registry metadata or an Avro contract is invalid.

class fraudtwin.contracts.registry.RegistryReport(path, subjects, versions, fingerprints)

Bases: object

Summary returned after validating a complete registry.

Parameters:

• path (Path)

• subjects (int)

• versions (int)

• fingerprints (dict[str, str])

fraudtwin.contracts.registry.contract_registry(path=None)

Load and validate a registry in one call.

Return type:

AvroContractRegistry

Parameters:

path (str | Path | None)

fraudtwin.contracts.registry.default_registry_path()

Return the bundled registry path in a checkout or installed wheel.

Return type:

Path

fraudtwin.contracts.registry.load_contract_registry(path=None)

Load registry metadata and parse every declared Avro schema.

Return type:

AvroContractRegistry

Parameters:

path (str | Path | None)

fraudtwin.contracts.registry.validate_avro_datum(schema, datum)

Validate one mapped datum with Apache Avro’s implementation.

Return type:

None

Parameters:

• schema (Schema)

• datum (Mapping[str, Any])