Processors

Index:

• MySQL
  • MySQL BinLog Processor
  • How to Use
• MongoDB
  • Mongo Streams
  • How to Use
• PostgreSQL
  • Server requirements
  • Local server with Docker
  • Usage
  • How it maps to Record
  • Confirming the position
  • Schema changes (DDL)
  • Limitations
  • Further reading
  • API
• MS SQL Server
  • How it works
  • Enabling CDC
  • Implementing with IProcessor
  • Further reading
• Oracle
  • Recommended approaches
  • Implementing with IProcessor
  • Further reading

Base Interface

Base processor interface for CDC (Change Data Capture) implementations. This module defines the IProcessor abstract base class that all CDC processors must implement, along with common CRUD event types.

class core_cdc.processors.base.IProcessor(service: str, targets: list[ITarget], logger: Logger, events_to_stream: Iterable[EventType] = (EventType.INSERT, EventType.UPDATE, EventType.DELETE), add_event_timestamp: bool = False)

Bases: IFactory

Interface for all “Change Data Capture” implementations

__init__(service: str, targets: list[ITarget], logger: Logger, events_to_stream: Iterable[EventType] = (EventType.INSERT, EventType.UPDATE, EventType.DELETE), add_event_timestamp: bool = False) → None

Parameters:

• service – The name of the service that is doing the CDC process.

• events_to_stream – By default, insert, update and, delete operations will be streamed.

• logger – Logger used.

• add_event_timestamp – If True, the column event_timestamp will be added when a table is created (only for the targets that execute SQL, see ITarget.get_add_column_ddl). The column could be useful for UPSERT or MERGE operations.

classmethod registration_key() → str

It returns the name (reference) for the key used to register, like: return self.__name__

execute() → None

Execute the CDC processor to read and process events from the stream. This method processes DDL and DML events, forwarding them to configured targets.

_process_ddl_event(event: Any) → None

It converts the raw DDL event into a DdlEvent and forwards it to the targets with execute_ddl enabled.

_add_event_timestamp_column(target: ITarget, ddl_event: DdlEvent) → None

It adds the column event_timestamp to the table that was just created

abstractmethod get_events() → Iterator[Any]

It returns an iterator with the events to process

abstractmethod get_event_type(event: Any) → EventType

It returns the event type

abstractmethod process_dml_event(event: Any) → list[Record]

It processes the event and return the records to stream

get_ddl_event(event: Any) → DdlEvent

It converts an event classified as EventType.DDL_STATEMENT into a DdlEvent, the engine independent representation the targets work with. It must be implemented by the processors that produce DDL events.

process_event(event: Any) → None

It should be implemented if another event (apart from DDL and DML) must be processed…

_abc_impl = <_abc._abc_data object>

_impls: ClassVar[dict[str, type[Self]]] = {'PostgresWalProcessor': <class 'core_cdc.processors.postgres.postgres_wal.PostgresWalProcessor'>}