Transformation Layer

The Transformation Layer maps raw database rows (from insert, update, or delete operations) into standard LangChain Document instances, manages unique document keys, and sanitizes input data.

Module Summary

DocumentBuilder	Converts a DebeziumEventModel into a BuildResult (Document + ID + op).
IdStrategy	Abstract ID strategy — produce a stable document ID from a CDC event.
TablePkIdStrategy	Default strategy: {schema}.{table}:{primary_key}.
CompositeIdStrategy	SHA-256 hash of specified PK fields — safe for composite or binary keys.
CustomIdStrategy	User-supplied callable ID strategy for full control.
ProjectionPolicy	Top-level projection policy with optional per-table overrides.
TableProjectionPolicy	Field-level projection configuration for a single table.
sanitize_metadata	Coerce metadata values to types Chroma/PGVector/Milvus can store.

API Reference

Document builder — converts CDC events into LangChain Documents.

This is the bridge between the ingestion/canonical layer and the synchronization/vector-store layer. Each CDC event produces a (Document, doc_id, op) BuildResult that the SyncManager consumes.

class BuildResult(doc_id, op, document=None)

Bases: object

Output of DocumentBuilder.build() for a single CDC event.

Variables:

• doc_id (str) – Stable, deterministic document ID for vector store operations.

• op (str) – CDC operation: ‘c’ (create), ‘u’ (update), ‘d’ (delete), ‘r’ (read/snapshot).

• document (langchain_core.documents.base.Document | None) – The LangChain Document (None for delete operations).

Parameters:

• doc_id (str)

• op (str)

• document (Document | None)

class DocumentBuilder(id_strategy=None, projection_policy=None, extra_metadata=None, sanitizer=None)

Bases: object

Converts a DebeziumEventModel into a BuildResult (Document + ID + op).

Wires together:

• IdStrategy → stable document ID

• ProjectionPolicy → which fields → page_content / metadata

Additional system metadata injected into every document:

• _table table name

• _schema schema/namespace

• _op CDC operation character

• _ts_ms source event timestamp (ms since epoch)

• _source optional Debezium source block fields

Usage:

builder = DocumentBuilder()
result = builder.build(event)
# result.doc_id, result.op, result.document

Parameters:

• id_strategy (IdStrategy | None)

• projection_policy (ProjectionPolicy | None)

• extra_metadata (dict[str, Any] | None)

• sanitizer (Callable[[dict[str, Any]], dict[str, Any]] | None)

build(event, allow_soft_delete=False)

Build a BuildResult from a canonical CDC event.

Parameters:

• event (DebeziumEventModel) – The DebeziumEventModel to convert.

• allow_soft_delete (bool) – If True, builds a Document with _is_deleted=True instead of returning None for delete operations.

Return type:

BuildResult

Deterministic document ID strategies for vector store synchronization.

A stable, unique document ID per logical row is the foundation of correct CDC-to-vector-store synchronization. The same row must always produce the same ID across insert, update, and delete events so upsert/replace semantics work correctly.

class IdStrategy

Bases: ABC

Abstract ID strategy — produce a stable document ID from a CDC event.

abstract generate(event)

Return a stable, unique document ID for this event’s logical row.

Parameters:

event (DebeziumEventModel)

Return type:

str

class TablePkIdStrategy(pk_fields=None)

Bases: IdStrategy

Default strategy: {schema}.{table}:{primary_key}.

Examples

• public.orders:42

• public.customers:john@example.com

• inventory.products:SKU-001|warehouse-A (composite key)

Uses event.extract_primary_key(pk_fields) which reads from the Debezium JSON key field first, then falls back to payload columns.

Parameters:

pk_fields (list[str] | None)

generate(event)

Return a stable, unique document ID for this event’s logical row.

Parameters:

event (DebeziumEventModel)

Return type:

str

class CompositeIdStrategy(pk_fields)

Bases: IdStrategy

SHA-256 hash of specified PK fields — safe for composite or binary keys.

Produces IDs like public.orders:a3f2c1d4e5b6... (20-char hex prefix).

Parameters:

pk_fields (list[str])

generate(event)

Return a stable, unique document ID for this event’s logical row.

Parameters:

event (DebeziumEventModel)

Return type:

str

class CustomIdStrategy(fn)

Bases: IdStrategy

User-supplied callable ID strategy for full control.

Parameters:

fn (Callable[[DebeziumEventModel], str])

generate(event)

Return a stable, unique document ID for this event’s logical row.

Parameters:

event (DebeziumEventModel)

Return type:

str

resolve_id_strategy(strategy, **kwargs)

Resolve a strategy name string or an existing IdStrategy instance.

Parameters:

• strategy (str | IdStrategy) – "table_pk", "composite", or an IdStrategy instance.

• **kwargs (object) – Forwarded to the strategy constructor (e.g. pk_fields).

Raises:

ValueError – If the name is not recognised.

Return type:

IdStrategy

Field projection and content-template policies for document building.

A ProjectionPolicy controls:

• Which row fields become the page_content of a LangChain Document.

• Which fields appear in LangChain metadata (for filtered retrieval).

• Optional pre-compiled string.Template content templates (e.g. "$name — $description").

• Per-table overrides so different tables have different rendering rules.

class TableProjectionPolicy(include_fields=None, exclude_fields=<factory>, metadata_fields=None, content_template=None)

Bases: object

Field-level projection configuration for a single table.

Variables:

• include_fields (list[str] | None) – If set, only these fields enter the content block.

• exclude_fields (list[str]) – Fields excluded (binary blobs, secrets, etc.). Applies only when include_fields is not specified.

• metadata_fields (list[str] | None) – Fields copied into Document.metadata; None = all.

• content_template (str | None) – Optional string.Template pattern. Uses $variable or ${variable} syntax. Example: "$product_name — price: $price"

Parameters:

• include_fields (list[str] | None)

• exclude_fields (list[str])

• metadata_fields (list[str] | None)

• content_template (str | None)

project_row(row)

Project a row dict into (page_content, metadata).

Returns:

A tuple of the page_content string and a metadata dict.

Parameters:

row (dict[str, Any])

Return type:

tuple[str, dict[str, Any]]

class ProjectionPolicy(default=None, overrides=None)

Bases: object

Top-level projection policy with optional per-table overrides.

Usage:

policy = ProjectionPolicy(
    default=TableProjectionPolicy(
        exclude_fields=["password_hash", "avatar_blob"],
    ),
    overrides={
        "products": TableProjectionPolicy(
            content_template="$name — $description — $$${price}",
            metadata_fields=["id", "category", "price"],
        )
    },
)

page_content, metadata = policy.project(event)

Parameters:

• default (TableProjectionPolicy | None)

• overrides (dict[str, TableProjectionPolicy] | None)

project(event)

Project the current row of a CDC event into (page_content, metadata).

Returns empty content string for delete events (no after row).

Parameters:

event (DebeziumEventModel)

Return type:

tuple[str, dict[str, Any]]

classmethod default_policy()

Return a sensible default policy (all fields, no template).

Return type:

ProjectionPolicy

Metadata sanitization utilities for vector database compatibility.

sanitize_metadata(meta)

Coerce metadata values to types Chroma/PGVector/Milvus can store.

Chroma accepts: str, int, float, bool. This converts bytes→hex str, Decimal→float, date/datetime→isoformat, None→”null”, and everything else→str.

Parameters:

meta (dict[str, Any])

Return type:

dict[str, Any]