fraudtwin.lakehouse#

Optional Iceberg materialization and verification.

Status: Optional

Classes#

fraudtwin.lakehouse.BronzeRecord

Canonical raw-record envelope used by both ingestion paths.

fraudtwin.lakehouse.IcebergLakehouse

Small PyIceberg writer with immutable append and snapshot metadata.

fraudtwin.lakehouse.LakehouseEnvironment

Non-secret connection settings resolved from environment variables.

fraudtwin.lakehouse.LakehouseMaterializationResult

Immutable metadata produced by a batch or streaming materialization.

Exceptions#

fraudtwin.lakehouse.LakehouseConfigurationError

Raised for missing or unsafe lakehouse configuration.

fraudtwin.lakehouse.LakehouseDependencyError

Raised when an Iceberg/Kafka integration extra is not installed.

Functions#

fraudtwin.lakehouse.build_bronze_records

Build a deterministic raw envelope from a complete generated run.

fraudtwin.lakehouse.consume_kafka_once

Consume a bounded Kafka batch into immutable Bronze and Silver tables.

fraudtwin.lakehouse.deduplicate_records

Apply a stable Silver decision while retaining deterministic winners.

fraudtwin.lakehouse.logical_fingerprint

Hash logical table content, excluding arrival and broker metadata.

fraudtwin.lakehouse.materialize_run

Backfill one generated run and optionally commit it to Iceberg.

fraudtwin.lakehouse.silver_rows

Normalize deduplicated Bronze records without rewriting raw payloads.

fraudtwin.lakehouse.verify_materialization

Validate a materialization manifest without contacting Iceberg.

Detailed API#

Optional Iceberg lakehouse adapters for generated FraudTwin runs.

The generator remains file-first. This module provides deterministic batch backfill and a small Kafka consumer adapter at the integration boundary. All Iceberg and Kafka imports are lazy so normal installs remain dependency-free.

class fraudtwin.lakehouse.BronzeRecord(run_id, source, subject, table_name, record_id, record_key, contract_fingerprint, payload, raw_payload_b64, headers, event_time, received_at, kafka_topic=None, kafka_partition=None, kafka_offset=None)[source][source]

Bases: object

Canonical raw-record envelope used by both ingestion paths.

Parameters:

  • run_id (str)

  • source (str)

  • subject (str)

  • table_name (str)

  • record_id (str)

  • record_key (str | None)

  • contract_fingerprint (str | None)

  • payload (dict[str, Any])

  • raw_payload_b64 (str | None)

  • headers (dict[str, str])

  • event_time (str | None)

  • received_at (str)

  • kafka_topic (str | None)

  • kafka_partition (int | None)

  • kafka_offset (int | None)

class fraudtwin.lakehouse.IcebergLakehouse(environment, config=None)[source][source]

Bases: object

Small PyIceberg writer with immutable append and snapshot metadata.

Parameters:
append_stream(table, rows, *, batch_size=10_000)[source][source]

Append an unbounded row stream in bounded batches.

Return type:

int | None

Parameters:

  • table (str)

  • rows (Iterable[Mapping[str, Any]])

  • batch_size (int)

maintenance(table, operation, *, snapshot_id=None, execute=False)[source][source]

Plan or execute one explicitly requested table-maintenance action.

Dry-run is the default. Execution is intentionally snapshot-scoped; callers must name the snapshot to expire so retained run snapshots cannot be removed accidentally.

Return type:

dict[str, object]

Parameters:

  • table (str)

  • operation (str)

  • snapshot_id (int | None)

  • execute (bool)

exception fraudtwin.lakehouse.LakehouseConfigurationError[source][source]

Bases: ValueError

Raised for missing or unsafe lakehouse configuration.

exception fraudtwin.lakehouse.LakehouseDependencyError[source][source]

Bases: RuntimeError

Raised when an Iceberg/Kafka integration extra is not installed.

class fraudtwin.lakehouse.LakehouseEnvironment(catalog_uri, warehouse, s3_endpoint=None, s3_region='us-east-1', s3_access_key=None, s3_secret_key=None, catalog_name='fraudtwin')[source][source]

Bases: object

Non-secret connection settings resolved from environment variables.

Parameters:

  • catalog_uri (str)

  • warehouse (str)

  • s3_endpoint (str | None)

  • s3_region (str)

  • s3_access_key (str | None)

  • s3_secret_key (str | None)

  • catalog_name (str)

catalog_properties()[source][source]

Return PyIceberg REST/S3 properties resolved from the environment.

Return type:

dict[str, str]

class fraudtwin.lakehouse.LakehouseMaterializationResult(materialization_id, run_id, source_mode, row_counts, logical_fingerprint, snapshots, manifest_path)[source][source]

Bases: object

Immutable metadata produced by a batch or streaming materialization.

Parameters:

  • materialization_id (str)

  • run_id (str)

  • source_mode (str)

  • row_counts (dict[str, int])

  • logical_fingerprint (str)

  • snapshots (dict[str, int | None])

  • manifest_path (Path | None)

fraudtwin.lakehouse.build_bronze_records(entities, behavior, manifest, *, include_oracle=False, registry=None, source='parquet')[source][source]

Build a deterministic raw envelope from a complete generated run.

Return type:

tuple[BronzeRecord, ...]

Parameters:

  • entities (Any)

  • behavior (Any)

  • manifest (RunManifest)

  • include_oracle (bool)

  • registry (AvroContractRegistry | None)

  • source (str)

fraudtwin.lakehouse.consume_kafka_once(*, environment, lakehouse, max_messages=100, timeout_seconds=5.0)[source][source]

Consume a bounded Kafka batch into immutable Bronze and Silver tables.

Return type:

int

Parameters:
fraudtwin.lakehouse.deduplicate_records(records)[source][source]

Apply a stable Silver decision while retaining deterministic winners.

Return type:

tuple[BronzeRecord, ...]

Parameters:

records (Iterable[BronzeRecord])

fraudtwin.lakehouse.logical_fingerprint(rows)[source][source]

Hash logical table content, excluding arrival and broker metadata.

Return type:

str

Parameters:

rows (Iterable[Mapping[str, Any]])

fraudtwin.lakehouse.materialize_run(run_dir, *, config=None, environment=None, write_iceberg=True, include_oracle=None, source_mode='batch')[source][source]

Backfill one generated run and optionally commit it to Iceberg.

Return type:

LakehouseMaterializationResult

Parameters:

  • run_dir (Path)

  • config (LakehouseConfig | None)

  • environment (LakehouseEnvironment | None)

  • write_iceberg (bool)

  • include_oracle (bool | None)

  • source_mode (str)

fraudtwin.lakehouse.silver_rows(records)[source][source]

Normalize deduplicated Bronze records without rewriting raw payloads.

Return type:

tuple[dict[str, Any], ...]

Parameters:

records (Iterable[BronzeRecord])

fraudtwin.lakehouse.verify_materialization(path)[source][source]

Validate a materialization manifest without contacting Iceberg.

Return type:

dict[str, Any]

Parameters:

path (Path)