fraudtwin.contracts#

Versioned data contracts and schema access.

Status: Stable

Classes#

fraudtwin.contracts.AvroContractRegistry

Parsed source-controlled registry and its Avro schemas.

fraudtwin.contracts.AvroDatumMapper

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

fraudtwin.contracts.RegistryReport

Summary returned after validating a complete registry.

Exceptions#

fraudtwin.contracts.ContractValidationError

Raised when registry metadata or an Avro contract is invalid.

Functions#

fraudtwin.contracts.contract_registry

Load and validate a registry in one call.

fraudtwin.contracts.default_registry_path

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

fraudtwin.contracts.load_contract_registry

Load registry metadata and parse every declared Avro schema.

Detailed API#

Bundled Avro contracts for observable operational events.

class fraudtwin.contracts.AvroContractRegistry(root, compatibility, subjects)[source][source]

Bases: object

Parsed source-controlled registry and its Avro schemas.

Parameters:

  • root (Path)

  • compatibility (str)

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

datum(subject, record)[source][source]

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])

decode(subject, payload)[source][source]

Decode one payload using the latest subject schema.

Return type:

dict[str, Any]

Parameters:

  • subject (str)

  • payload (bytes)

encode(subject, record)[source][source]

Encode one observable datum using the latest subject schema.

Return type:

bytes

Parameters:

  • subject (str)

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

mapper()[source][source]

Return the explicit datum mapper for this registry.

Return type:

AvroDatumMapper

validate()[source][source]

Validate metadata, canonical fingerprints, and full compatibility.

Return type:

RegistryReport

class fraudtwin.contracts.AvroDatumMapper(registry)[source][source]

Bases: object

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

Parameters:

registry (AvroContractRegistry)

decode(subject, payload)[source][source]

Decode one payload using the latest subject schema.

Return type:

dict[str, Any]

Parameters:

  • subject (str)

  • payload (bytes)

encode(subject, record)[source][source]

Encode a domain record using the latest subject schema.

Return type:

bytes

Parameters:

  • subject (str)

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

to_datum(subject, record)[source][source]

Return a validated observable Avro datum for subject.

Return type:

dict[str, Any]

Parameters:

  • subject (str)

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

exception fraudtwin.contracts.ContractValidationError[source][source]

Bases: ValueError

Raised when registry metadata or an Avro contract is invalid.

class fraudtwin.contracts.RegistryReport(path, subjects, versions, fingerprints)[source][source]

Bases: object

Summary returned after validating a complete registry.

Parameters:

  • path (Path)

  • subjects (int)

  • versions (int)

  • fingerprints (dict[str, str])

fraudtwin.contracts.contract_registry(path=None)[source][source]

Load and validate a registry in one call.

Return type:

AvroContractRegistry

Parameters:

path (str | Path | None)

fraudtwin.contracts.default_registry_path()[source][source]

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

Return type:

Path

fraudtwin.contracts.load_contract_registry(path=None)[source][source]

Load registry metadata and parse every declared Avro schema.

Return type:

AvroContractRegistry

Parameters:

path (str | Path | None)