PostgresCdcTrigger
class PostgresCdcTrigger(
conn: PostgresCdcConn,
tables: QualifiedTablesSpec,
start_from: PostgresCdcStartFromSpec,
cdc_format: CdcFormat = CdcFormat(values_format='columns', flatten_values=True),
buffer_max_rows: int = 10000,
buffer_max_bytes: int | None = None,
buffer_max_secs: int = 60,
trigger_max_rows: int | None = None,
trigger_max_bytes: int | None = None,
trigger_max_secs: int = 60,
poll_interval_seconds: int = 1,
blocking_timeout_seconds: int = 1,
replication_slot: str | None = None,
publication_name: str | None = None,
slot_behavior: ReplicationSlotBehaviorSpec = 'reuse',
start: datetime | None = None,
end: datetime | None = None,
)
Bases: CdcTrigger
Categories: trigger
A CdcTrigger that captures changes from a PostgreSQL database using WAL
logical replication.
This trigger connects to a PostgreSQL server, reads the write-ahead log via the wal2json output plugin for change data capture events (inserts, updates, deletes) on specified tables, and stages them.
Initialize a PostgresCdcTrigger instance.
Parameters
connThe PostgreSQL CDC connection configuration.
tablesQualified table names to track for changes.
start_fromThe position to start reading changes from (LSN or relative position).
cdc_formatThe output format specification for CDC data.
buffer_max_rowsMaximum rows in memory before flushing to the working directory.
buffer_max_bytesMaximum bytes in memory before flushing to the working directory.
buffer_max_secsMaximum seconds before flushing memory data to the working directory.
trigger_max_rowsMaximum total rows before staging from working directory to the stage location.
trigger_max_bytesMaximum total bytes before staging from working directory to the stage location.
trigger_max_secsMaximum seconds before staging from working directory to the stage location.
poll_interval_secondsInterval in seconds between polling for new changes.
blocking_timeout_secondsTimeout in seconds for blocking reads from the replication stream.
replication_slotOptional name of the logical replication slot to use.
publication_nameOptional name of the PostgreSQL publication to subscribe to.
slot_behaviorWhether to create a new replication slot or reuse an existing one. Defaults to "reuse".
startOptional datetime for the trigger to start working.
endOptional datetime for the trigger to stop working.
Properties
blocking_timeout_secondsintTimeout in seconds for blocking reads from the replication stream.
buffer_max_bytesint | NoneMaximum bytes in memory before flushing to the working directory.
buffer_max_rowsintMaximum rows in memory before flushing to the working directory.
buffer_max_secsintMaximum seconds before flushing memory data to the working directory.
cdc_formatCdcFormatOutput format specification for CDC data.
connConnCDC connection configuration.
enddatetime | Noneoutdict[int, str] | NoneMapping of table indices to qualified table names.
poll_interval_secondsintInterval in seconds between polling for new changes.
publication_namestr | NoneName of the PostgreSQL publication to subscribe to.
replication_slotstr | NoneName of the logical replication slot.
slot_behaviorReplicationSlotBehaviorSpecWhether to create a new replication slot or reuse an existing one.
startdatetime | Nonestart_fromPostgresCdcStartFromSpec | NonePosition to start reading changes from.
tableslist[str] | NoneQualified table names being tracked for changes.
trigger_max_bytesint | NoneMaximum total bytes before staging from working directory to the stage location.
trigger_max_rowsint | NoneMaximum total rows before staging from working directory to the stage location.
trigger_max_secsintMaximum seconds before staging from working directory to the stage location.
Methods
collect_tablesdef collect_tables(
session: Any,
cdc_format: CdcFormat = CdcFormat(values_format='columns', flatten_values=True),
) -> dict[str, TrackedTable]
harvest_tablesdef harvest_tables(
session: Any,
cdc_format: CdcFormat = CdcFormat(values_format='columns', flatten_values=True),
) -> dict[str, TrackedTable]