Ingestion Layer

The Ingestion Layer is responsible for capturing Debezium change data events from various input streams (Kafka Connect, JSON files, etc.) and converting them into normalized Python models.

Module Summary

BaseIngestionHandler	Abstract base that all ingestion handlers extend.
JsonIngestionHandler	JSON-mode ingestion.
ConnectIngestionHandler	Connect-mode (JPype + io.debezium.embedded.Connect) ingestion handler.

API Reference

Abstract base for all PyDebeziumAI ingestion handlers.

Each concrete handler wraps a pydbzengine engine (JSON or Connect mode) and dispatches normalized DebeziumEventModel instances to registered callbacks.

class BaseIngestionHandler

Bases: ABC

Abstract base that all ingestion handlers extend.

Subclasses implement engine construction; this base handles callback registration and dispatching so the sync pipeline stays decoupled.

Usage:

handler = JsonIngestionHandler()
handler.add_event_callback(sync_manager.handle_event)
handler.add_error_callback(my_error_logger)
engine = handler.build_engine(debezium_props)
engine.run()

add_event_callback(cb)

Register a callback that receives each normalized DebeziumEventModel.

Parameters:

cb (Callable[[DebeziumEventModel], None])

Return type:

None

add_error_callback(cb)

Register a callback for parse/dispatch errors (receives exc + raw record).

Parameters:

cb (Callable[[Exception, Any], None])

Return type:

None

abstract build_engine(properties)

Build and return a configured pydbzengine engine instance.

Parameters:

properties (dict[str, str])

Return type:

Any

JSON-mode ingestion handler.

Wraps pydbzengine’s BasePythonChangeHandler to consume CDC events serialized as JSON strings, parse them into canonical DebeziumEventModel instances, and dispatch to registered callbacks.

This mode works without a JVM bridge for record schema access — the Debezium engine serializes events to JSON before handing them to Python.

class JsonIngestionHandler

Bases: BaseIngestionHandler

JSON-mode ingestion.

Usage:

handler = JsonIngestionHandler()
handler.add_event_callback(sync_manager.handle_event)
engine = handler.build_engine({
    "name": "engine",
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    ...
})
engine.run()

build_engine(properties)

Build and return a configured DebeziumJsonEngine.

Parameters:

properties (dict[str, str])

Return type:

Any

Connect-mode ingestion handler.

Uses JPype + io.debezium.embedded.Connect format to access Debezium SourceRecord objects directly, avoiding the JSON serialization round-trip.

Wires the SourceRecordExtractor into the PyDebeziumAI pipeline:

Java SourceRecord

→ SourceRecordExtractor (struct_to_dict, full type conversion)

→ DebeziumRecord

→ DebeziumEventModel

→ SyncManager callbacks

class ConnectIngestionHandler(conversion_config=None, topic_class_map=None)

Bases: BaseIngestionHandler

Connect-mode (JPype + io.debezium.embedded.Connect) ingestion handler.

Matches the architecture from debezium_connect.py + BaseConnectChangeHandler, integrated into the PyDebeziumAI pipeline.

Parameters:

• conversion_config (ConversionConfig | None) – Controls type conversion (numeric_mode, tz_aware).

• topic_class_map (Mapping[str, type[Any]] | None) – Maps topic names to Python classes for typed access.

Usage:

handler = ConnectIngestionHandler(
    conversion_config=ConversionConfig(numeric_mode="native"),
)
handler.add_event_callback(sync_manager.handle_event)
engine = handler.build_engine(debezium_props)
engine.run()

build_engine(properties)

Build a Connect-format DebeziumEngine via JPype.

Uses io.debezium.embedded.Connect format (Debezium 3.0+), which passes raw SourceRecord objects instead of JSON strings — matching the DebeziumConnectEngine implementation.

Parameters:

properties (dict[str, str])

Return type:

Any