Models Layer

The Models Layer contains the schema validations and representation structures for Debezium CDC records, Kafka messages, and internal payload states.

Module Summary

`DebeziumEventModel`	Top-level canonical Debezium CDC event envelope.
`DebeziumPayloadModel`	Normalized payload extracted from a Debezium CDC envelope.
`ConnectMessageExtractor`	Wraps a raw Java ChangeEvent (JSON / polling mode) and extracts its fields.
`SourceRecordExtractor`	Wraps a raw Java SourceRecord (Connect mode) and converts it to Python.
`DebeziumSchema`	Lightweight wrapper around the raw Debezium schema dict.
`DebeziumRecord`	Python-native intermediate representation of a single Debezium CDC event.

API Reference

Canonical Pydantic v2 event models for Debezium CDC events.

All ingestion modes (JSON + Connect) normalize to these models before entering the transformation layer, providing a stable internal contract.

Models:: DebeziumSchemaField — a single field in a Debezium schema definition DebeziumSchemaModel — the schema section of a Debezium JSON envelope DebeziumPayloadModel — the payload section (before/after/op/ts_ms) DebeziumEventModel — top-level canonical event envelope

class DebeziumSchemaField(*, field=None, type=None, optional=None, name=None, version=None, parameters=None)[source]

Bases: BaseModel

A single field in a Debezium schema definition.

Parameters:

field (str | None)
type (str | None)
optional (bool | None)
name (str | None)
version (int | None)
parameters (dict[str, Any] | None)

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class DebeziumSchemaModel(*, type=None, fields=None, name=None, optional=None, version=None)[source]

Bases: BaseModel

The schema section of a Debezium JSON envelope.

Parameters:

type (str | None)
fields (list[DebeziumSchemaField] | None)
name (str | None)
optional (bool | None)
version (int | None)

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class DebeziumPayloadModel(*, before=None, after=None, op, ts_ms=None, source=None)[source]

Bases: BaseModel

Normalized payload extracted from a Debezium CDC envelope.

Parameters:

before (dict[str, Any] | None)
after (dict[str, Any] | None)
op (str)
ts_ms (int | None)
source (dict[str, Any] | None)

property current_row: dict[str, Any] | None: Return current row state (‘after’ for c/u/r, ‘before’ for d).

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class DebeziumEventModel(*, destination, partition=None, key=None, schema=None, payload)[source]

Bases: BaseModel

Top-level canonical Debezium CDC event envelope.

Parameters:

destination (str)
partition (int | None)
key (str | None)
schema (DebeziumSchemaModel | None)
payload (DebeziumPayloadModel)

model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod from_json_record(record)[source]

Parse from a pydbzengine ChangeEvent (JSON / polling mode).

Parameters:: record (Any)
Return type:: DebeziumEventModel

classmethod from_dict(destination, key, payload)[source]

Build directly from a raw dict (Connect-mode extraction or testing).

Parameters:

destination (str)
key (str | None)
payload (dict[str, Any])

Return type:

DebeziumEventModel

classmethod from_record(record)[source]

Build a validated DebeziumEventModel from a DebeziumRecord.

This is the bridge between the intermediate dataclass layer and the Pydantic-validated pipeline layer.

Parameters:: record (Any) – A DebeziumRecord instance (from models/datastructures.py).
Returns:: A validated DebeziumEventModel.
Return type:: DebeziumEventModel

property table_name: str: Last dotted segment of destination, e.g. ‘orders’ from ‘server.public.orders’.

property schema_name: str: Second-to-last segment (schema/namespace), or empty string.

property namespace: str: Full schema.table string.

extract_primary_key(pk_fields=None)[source]

Extract the primary key string for this event.

Priority order: 1. Parse the JSON key field (Debezium Connect key schema). 2. Extract from payload using pk_fields if provided. 3. Return None if no primary key is found.

Parameters:: pk_fields (list[str] | None)
Return type:: str | None

Python utilities for Debezium CDC events.

Provides:

ConversionConfig : controls numeric/tz output
ConnectMessageExtractor : wraps JSON-mode ChangeEvent (no JPype)
SourceRecordExtractor : wraps Connect-mode SourceRecord (JPype)
struct_to_dict : recursive Java Struct → Python dict
print_record_info : debug utility for SourceRecord introspection
extract_all : batch helper for JSON mode
extract_connect_all : batch helper for Connect mode

Logical types supported

Kafka Connect primitives:: org.apache.kafka.connect.data.Timestamp → datetime (UTC) org.apache.kafka.connect.data.Date → datetime.date org.apache.kafka.connect.data.Time → datetime.time org.apache.kafka.connect.data.Decimal → decimal.Decimal (with scale)
Debezium time types:: io.debezium.time.Date → datetime.date io.debezium.time.Time → datetime.time (ms) io.debezium.time.MicroTime → datetime.time (µs) io.debezium.time.NanoTime → datetime.time (ns) io.debezium.time.Timestamp → datetime (ms, UTC) io.debezium.time.MicroTimestamp → datetime (µs, UTC) io.debezium.time.NanoTimestamp → datetime (ns, UTC) io.debezium.time.ZonedTimestamp → datetime (tz-aware) io.debezium.time.ZonedTime → datetime.time (tz-aware) io.debezium.time.MicroDuration → datetime.timedelta io.debezium.time.Interval → str (ISO-8601 interval)
Debezium data types:: io.debezium.data.Uuid → uuid.UUID io.debezium.data.Bits → bytes io.debezium.data.Json → dict | str io.debezium.data.Enum → str io.debezium.data.EnumSet → set[str] io.debezium.data.VariableScaleDecimal → decimal.Decimal io.debezium.data.geometry.Geometry → dict (GeoJSON-like) io.debezium.data.geometry.Point → dict {“x”:..,”y”:..,”srid”:..} io.debezium.data.geometry.Geography → dict (GeoJSON-like)

class ConversionConfig(numeric_mode='native', tz_aware=True)[source]

Bases: object

Controls how Debezium Connect schema types are converted to Python.

Variables:

numeric_mode (Literal['native', 'numpy']) – "native" → int / float / Decimal; "numpy" → numpy.int64 / numpy.float64 (requires numpy).
tz_aware (bool) – Return timezone-aware datetimes when True (default).

Parameters:

numeric_mode (Literal['native', 'numpy'])
tz_aware (bool)

struct_to_dict(struct, schema, cfg=None, topic_class_map=None, topic='')[source]

Convert a JPype Kafka Connect Struct to a Python dict.

Optionally maps the result to a typed dataclass via topic_class_map.

Parameters:

struct (Any) – JPype Struct object from a Debezium SourceRecord.
schema (Any) – The Kafka Connect Schema for this struct.
cfg (ConversionConfig | None) – Conversion configuration (defaults to native mode, tz_aware=True).
topic_class_map (Mapping[str, type[Any]] | None) – Maps topic/table names to Python classes. Keys can be full topic ("server.public.orders"), schema-qualified ("public.orders"), or bare table ("orders").
topic (str) – Topic name for class-map lookup.

Returns:

A plain Python dict, or an instance of the mapped class if a match found.

Return type:

Any

class ConnectMessageExtractor(java_record)[source]

Bases: object

Wraps a raw Java ChangeEvent (JSON / polling mode) and extracts its fields.

No JPype SourceRecord access — parses the JSON string produced by the Debezium engine’s JSON serializer.

Usage:

extractor = ConnectMessageExtractor(java_change_event)
print(extractor.destination, extractor.op, extractor.after)

Parameters:: java_record (Any)

class SourceRecordExtractor(source_record, conversion_config=None, topic_class_map=None)[source]

Bases: object

Wraps a raw Java SourceRecord (Connect mode) and converts it to Python.

Zero JSON serialization overhead — uses struct_to_dict internally.

Parameters:

source_record (Any) – Raw Java SourceRecord from handleConnectBatch.
conversion_config (ConversionConfig | None) – Optional ConversionConfig.
topic_class_map (Mapping[str, type[Any]] | None) – Optional dict mapping topic/table names to Python classes for typed after_typed / before_typed access.

Usage:

extractor = SourceRecordExtractor(
    source_record,
    conversion_config=ConversionConfig(numeric_mode="numpy"),
    topic_class_map={"inventory.customers": Customer},
)
print(extractor.destination)
print(extractor.after)        # plain dict
print(extractor.after_typed)  # Customer instance (if mapped)

property after_typed: Any: after as a typed class instance if topic_class_map matches.

property before_typed: Any: before as a typed class instance if topic_class_map matches.

property raw_key: str | None: JSON string of key_dict for compatibility with DebeziumEventModel.

extract_all(records, conversion_config=None)[source]

Wrap a batch of JSON-mode ChangeEvent records.

Parameters:

records (list[Any])
conversion_config (ConversionConfig | None)

Return type:

list[ConnectMessageExtractor]

extract_connect_all(records, conversion_config=None, topic_class_map=None)[source]

Wrap a batch of Connect-mode SourceRecord objects.

Parameters:

records (list[Any])
conversion_config (ConversionConfig | None)
topic_class_map (Mapping[str, type[Any]] | None)

Return type:

list[SourceRecordExtractor]

print_record_info(source_record)[source]

Print a human-readable summary of a raw Java SourceRecord for debugging.

Parameters:: source_record (Any)
Return type:: None

datastructures.py

Pure Python data structures representing a Debezium CDC event.

These are built from ConnectMessageExtractor (JSON mode) or SourceRecordExtractor (Connect mode) and serve as the intermediate layer between raw Java objects and Pydantic models.

Flow:

Java ChangeEvent (JPype)

→ ConnectMessageExtractor (connect_message.py)

→ DebeziumRecord (this file): → DebeziumEventModel (models/event.py)

class DebeziumSchema(raw=<factory>)[source]

Bases: object

Lightweight wrapper around the raw Debezium schema dict.

Populated from the JSON schema block when available (JSON mode). Not populated in Connect mode (schema is consumed by struct_to_dict).

Parameters:: raw (dict[str, Any])

class DebeziumRecord(destination, op, before, after, ts_ms, key=None, partition=None, schema=None)[source]

Bases: object

Python-native intermediate representation of a single Debezium CDC event.

Fields:: destination: Topic / table identifier. op: Operation type: ‘c’=create, ‘u’=update, ‘d’=delete, ‘r’=read/snapshot. before: Row state before the change (None for inserts/reads). after: Row state after the change (None for deletes). ts_ms: Debezium capture timestamp in milliseconds since epoch. key: The record key (JSON string). partition: Kafka partition (if applicable). schema: The Debezium schema envelope (JSON mode only).

Parameters:

destination (str)
op (str)
before (dict[str, Any] | None)
after (dict[str, Any] | None)
ts_ms (int | None)
key (str | None)
partition (int | None)
schema (DebeziumSchema | None)

classmethod from_extractor(extractor)[source]

Build a DebeziumRecord from a JSON-mode ConnectMessageExtractor.

Parameters:: extractor (ConnectMessageExtractor)
Return type:: DebeziumRecord

classmethod from_java_record(java_record)[source]

Build directly from a raw Java ChangeEvent (JSON mode).

Parameters:: java_record (Any)
Return type:: DebeziumRecord

classmethod from_source_record(source_record, conversion_config=None, topic_class_map=None)[source]

Build from a raw Java SourceRecord (Connect mode).

No JSON serialization/deserialization — uses struct_to_dict internally.

Parameters:

source_record (Any)
conversion_config (ConversionConfig | None)
topic_class_map (Mapping[str, type[Any]] | None)

Return type:

DebeziumRecord

get_current_state()[source]

Returns the most recent state: after for c/u/r, before for d.

Return type:: dict[str, Any] | None

records_from_batch(java_records)[source]

Convert a batch of JSON-mode Java ChangeEvent objects to DebeziumRecords.

Parameters:: java_records (list[Any])
Return type:: list[DebeziumRecord]

connect_records_from_batch(source_records, conversion_config=None, topic_class_map=None)[source]

Convert a batch of Connect-mode Java SourceRecord objects to DebeziumRecords.

Parameters:

source_records (list[Any])
conversion_config (ConversionConfig | None)
topic_class_map (Mapping[str, type[Any]] | None)

Return type:

list[DebeziumRecord]