core-cdc (CDC a.k.a Change Data Capture)

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

Installation

Install from PyPI using pip:

pip install core-cdc
uv pip install core-cdc  # Or using UV...

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

Quick Start

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

pip install --upgrade pip
pip install virtualenv

virtualenv --python=python3.12 .venv

source .venv/bin/activate

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

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

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

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

bash tests/functional/quick_test.sh

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

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\"}
  ]
})"

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

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

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"

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

docker stop mongo1 mongo2 mongo3 pg

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.

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:

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.

Links

• Documentation: https://core-cdc.readthedocs.io/en/latest/

• Repository: https://gitlab.com/bytecode-solutions/core/core-cdc

• Issues: https://gitlab.com/bytecode-solutions/core/core-cdc/-/issues

• Changelog: https://gitlab.com/bytecode-solutions/core/core-cdc/-/blob/master/CHANGELOG.md

• PyPI: https://pypi.org/project/core-cdc/

Support

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

Authors

• Alejandro Cora González - Initial work - alek.cora.glez@gmail.com