core-cdc#

It provides the core mechanism and required resources to implement “Change Data Capture” services…

PyPI Downloads Python Versions License Pipeline Status Docs Status Security

Documentation Contents#

Index:

Features#

Multi-Database CDC Support

  • MySQL Binary Log (BinLog) based change capture

  • MongoDB Change Streams for real-time event streaming

  • PostgreSQL logical replication (pgoutput) through psycopg2

  • Extensible processor architecture for additional database engines

Comprehensive Event Handling

  • DML operations: INSERT, UPDATE, DELETE

  • DDL operations: CREATE, ALTER, DROP (schemas and tables), delivered to the targets as engine independent DdlEvent objects

  • Configurable event filtering by operation type

Flexible Target Replication

  • Implement your own target by subclassing ITarget

  • Send records to any destination: database, queue, data warehouse, etc.

  • Support for multiple simultaneous targets

Standardized Data Format

  • Common Record (data) and DdlEvent (schema changes) structures for cross-service integration

  • Includes metadata: timestamps, transaction IDs, source position

  • JSON serialization support for streaming and messaging systems

Production-Ready Features

  • Error handling: a failing event or target is logged without stopping the stream

  • Comprehensive logging for monitoring and debugging

  • Optional event timestamp column for UPSERT/MERGE operations

Installation#

Install from PyPI using pip:

pip install core-cdc
uv pip install core-cdc     # Or using UV...
pip install -e ".[dev]"     # For development...

Setting Up Environment#

  1. Install required libraries:

pip install --upgrade pip
pip install virtualenv

  1. Create Python virtual environment:

virtualenv --python=python3.12 .venv

  1. Activate the virtual environment:

source .venv/bin/activate

Install packages#

pip install .
pip install -e ".[dev]"

Optional libraries#

pip install '.[all]'    # MySQL + MongoDB + PostgreSQL
pip install '.[mysql]'  # MySQL BinLog support
pip install '.[mongo]'  # MongoDB Change Streams support
pip install '.[postgres]'  # PostgreSQL logical replication support

Check tests and coverage#

python manager.py run-tests                   # unit tests
python manager.py run-tests --test-type integration
python manager.py run-coverage

The documentation is built with Sphinx, the docs extra has what it needs:

pip install -e ".[all,docs]"
cd docs && make clean && make html

Functional Tests#

Functional tests execute against real database servers and are not run automatically by pytest or tox. They must be invoked explicitly after the required Docker containers are running.

The quickest way to run both MySQL and MongoDB functional tests in one shot is the helper script tests/functional/quick_test.sh. It checks connectivity, runs both test suites via python -m pytest, and prints a metrics summary from the JSON result files written by each test. The PostgreSQL test is a separate file (tests/functional/check_postgres_wal.py): see the next section for the server and the command.

bash tests/functional/quick_test.sh

All connection parameters are overridable via environment variables (see the table below). The script defaults match the Docker commands in the next section.

Besides the data of the tables, the functional tests check against the real servers: the DDL events and add_event_timestamp, that the transaction control queries (BEGIN, COMMIT) are not sent as DDL, the JSON serialization of the records with every type of column, the warning about binlog_row_metadata, the MongoDB DDL and invalidate events, the resume token and the events of the transactions. The tests of the types and of the warning change binlog_row_metadata (SET GLOBAL, it needs the privilege) and restore it.

Alternatively, run individual test files directly:

python manager.py run-tests --test-type functional --pattern "*.py"

Environment variables accepted by the functional tests:

Variable

Default

Description

HOST_TEST_MONGO

localhost

MongoDB host

PORT_TEST_MONGO

27017

MongoDB port

DATABASE_TEST_MONGO

test

MongoDB database name

USER_TEST_MONGO

(none)

MongoDB username (optional)

PASSWORD_TEST_MONGO

(none)

MongoDB password (optional)

HOST_TEST_MYSQL

localhost

MySQL host

PORT_TEST_MYSQL

3306

MySQL port

DATABASE_TEST_MYSQL

tests

MySQL database name

USER_TEST_MYSQL

root

MySQL user

PASSWORD_TEST_MYSQL

mysql_password

MySQL password

HOST_TEST_POSTGRES

localhost

PostgreSQL host

PORT_TEST_POSTGRES

5432

PostgreSQL port

DATABASE_TEST_POSTGRES

postgres

PostgreSQL database name

USER_TEST_POSTGRES

postgres

PostgreSQL user (it needs REPLICATION)

PASSWORD_TEST_POSTGRES

postgres_password

PostgreSQL password

Spinning Up Local Servers with Docker#

The functional tests connect to these defaults. To use another port, map it in docker run and set the variable of the last column.

Engine

Container

Port

Test

Variable of the port

MongoDB

mongo1 to mongo3

27017

quick_test.sh

PORT_TEST_MONGO

MySQL

(no name)

3306

quick_test.sh

PORT_TEST_MYSQL

PostgreSQL

pg

5432

check_postgres_wal.py

PORT_TEST_POSTGRES

MongoDB Replica Set#

Required for Change Streams (three nodes, then rs.initiate):

docker network create mongoCluster

docker run -d --rm -p 27017:27017 --name mongo1 --network mongoCluster \
    mongo:5 mongod --replSet myReplicaSet --bind_ip localhost,mongo1

docker run -d --rm -p 27018:27017 --name mongo2 --network mongoCluster \
    mongo:5 mongod --replSet myReplicaSet --bind_ip localhost,mongo2

docker run -d --rm -p 27019:27017 --name mongo3 --network mongoCluster \
    mongo:5 mongod --replSet myReplicaSet --bind_ip localhost,mongo3

docker exec -it mongo1 mongosh --eval "rs.initiate({
  _id: \"myReplicaSet\",
  members: [
    {_id: 0, host: \"mongo1\"},
    {_id: 1, host: \"mongo2\"},
    {_id: 2, host: \"mongo3\"}
  ]
})"

Check cluster status:

docker ps
docker exec -it mongo1 mongosh --eval "rs.status()"

MySQL#

BinLog replication is enabled by default in the official image:

docker run \
  --env=MYSQL_ROOT_PASSWORD=mysql_password \
  --volume=/var/lib/mysql \
  -p 3306:3306 \
  --restart=no \
  -d mysql:latest

Note

Recent mysql:latest images are MySQL 8.4 or later, where SHOW MASTER STATUS was removed and BinLogStreamReader(resume_stream=False) fails. Use mysql:8.0 or pass resume_stream=True with an explicit log_file and log_pos (see the MySQL section).

PostgreSQL#

Logical replication, for PostgresWalProcessor. It is one container, and the server has to be started with wal_level=logical:

docker run -d --rm --name pg -p 5432:5432 \
    -e POSTGRES_PASSWORD=postgres_password \
    postgres:16 -c wal_level=logical

# It must print "logical":
docker exec pg psql -U postgres -tAc "show wal_level"

  • -c wal_level=logical is what enables logical replication. With the default (replica) the server refuses to create the slot: logical decoding requires wal_level >= logical.

  • POSTGRES_PASSWORD=postgres_password is the password the functional test uses by default (PASSWORD_TEST_POSTGRES). The postgres user is a superuser, so it has the REPLICATION attribute and nothing else has to be created.

  • The image tag is the version of PostgreSQL: 14, 15, 16, 17 and 18 were tested (for example postgres:18).

  • If 5432 is taken (a local PostgreSQL, for example) map another host port, -p 5433:5432, and run the test with PORT_TEST_POSTGRES=5433.

Run the functional test. It creates (and drops) its own table, publication and slot, it needs the driver, and it is not part of quick_test.sh:

pip install -e ".[postgres]"
python -m pytest -s tests/functional/check_postgres_wal.py

Stopping the containers#

The MongoDB and PostgreSQL containers were started with --rm, stopping them removes them:

docker stop mongo1 mongo2 mongo3 pg

The MySQL container has no name: use docker ps to find it and docker stop <id>.

Tested Versions#

These are the versions the library was verified with. Live means the functional tests (tests/functional) ran against real servers; the unit and integration tests use mocks and do not need any server.

Component

Versions

How it was tested

MySQL

9.7.2, 8.4.11, 8.0.46, 5.7.44

Live (official Docker images, ROW format). GTIDs on and off (9.7, 8.4). binlog_row_metadata FULL and MINIMAL (9.7, 8.4, 8.0).

MongoDB

6.0.28, 5.0.31

Live (replica set, official mongo:6 and mongo:5 images): change streams, DDL, invalidate, resume tokens and transactions. In 6.0 also the expanded events (create, createIndexes, dropIndexes, modify) and wallTime.

PostgreSQL

18.6, 17.11, 16.13, 15.17 and 14.24 (pgoutput, wal_level=logical)

Live (official postgres images, psycopg2-binary 2.9.13, Python 3.14 only): INSERT, UPDATE, DELETE, jsonb, TOAST columns, the update of the key, TRUNCATE, REPLICA IDENTITY FULL with primary_keys, the ALTER TABLE detection and the confirmation of the position (of a commit and while idle). Not tested: 10 to 13.

Python

CPython 3.9, 3.10, 3.11, 3.12, 3.13, 3.14 and PyPy 3.11

tox: unit and integration tests, and the live functional tests (MySQL 8.4 and MongoDB 5), on each one.

Libraries

mysql-replication 1.0.15 and 1.0.17; pymongo 4.17.0 and 4.18.1; PyMySQL 1.1.3 and 1.2.3; core-mixins 3.2.0 and 3.2.2

Both sets of versions ran the live tests.

Minimum declared libraries

mysql-replication 1.0.9, pymongo 4.0.0, core-mixins 3.2.0

Unit and integration tests on Python 3.9 (not live).

Notes:

  • MySQL 5.7 must be started with the binary log enabled (--server-id=1 --log-bin=mysql-bin --binlog-format=ROW). Four of the six MySQL functional tests run there: the other two are about binlog_row_metadata, a variable that does not exist in 5.7 (its column types are read from the database, see the MySQL section).

  • MongoDB create, createIndexes, dropIndexes and modify events need MongoDB 6.0+ and showExpandedEvents; rename, drop, dropDatabase and invalidate are sent by 5.0 too. Only the sharding operations (shardCollection, refineCollectionShardKey, reshardCollection) were tested with simulated events, since they need a sharded cluster.

Not tested (they may work, there is no verification): MySQL 8.1 to 8.3 and the 9.x versions other than 9.7.2, MariaDB, Percona and managed services (Amazon RDS / Aurora, Cloud SQL, Atlas), MongoDB before 5.0 and 7.0 or later, other minor versions of the ones listed, and sharded MongoDB clusters.

Implemented CDC Engines#

The following database engines have CDC implementations:

Fully Implemented

MySQL - Binary Log (BinLog) based CDC

  • Uses mysql-replication library

  • Captures INSERT, UPDATE, DELETE operations

  • Supports DDL events (CREATE, ALTER, DROP)

  • Fallback mechanism for column name resolution

  • See: core_cdc/processors/mysql/

MongoDB - Change Streams based CDC

  • Uses native MongoDB Change Streams

  • Captures INSERT, UPDATE, DELETE operations

  • Requires replica set configuration

  • Real-time event streaming

  • See: core_cdc/processors/mongo/

Planned / Documentation Only

MS SQL Server and Oracle implementations are not yet included. Reference guides and implementation templates are available in the documentation.

Contributing#

Contributions are welcome! Please:

  1. Fork the repository

  2. Create a feature branch

  3. Write tests for new functionality

  4. Ensure all tests pass and coverage stays at 100 %: python manager.py run-coverage

  5. Run the linters and type checkers: ruff check core_cdc, ty check core_cdc, mypy --explicit-package-bases core_cdc, pyright core_cdc and pylint core_cdc

  6. Run the security checks: bandit -r core_cdc and pip-audit

  7. Submit a pull request

License#

This project is licensed under the MIT License. See the LICENSE file for details.

Support#

For questions or support, please open an issue on GitLab or contact the maintainers.

Authors#