mysql-event-stream

An embeddable CDC engine for MySQL 8.4+ and MariaDB 10.11+, as a Python package over a ctypes FFI. It reads the binary log as a replica would and emits row-level change events. No libmysqlclient: the wire protocol is implemented directly in the bundled C++ core.

Install

pip install mysql-event-stream

Python 3.11 or later, no runtime dependencies, typed. Platform wheels are published for Linux x86_64 and aarch64 (recommended for servers) and macOS 15.0 or newer on x86_64 and arm64.

Usage

import os

from mysql_event_stream import CdcStream


async def run() -> None:
    async with CdcStream(
        host="mysql.example.com",
        user="replicator",
        password=os.environ["MYSQL_PASSWORD"],
        server_id=1001,
        include_databases=["shop"],
        start_gtid=await load_checkpoint(),
    ) as stream:
        async for event in stream:
            print(event.type, f"{event.database}.{event.table}")
            await handle(event)  # idempotent
            await save_checkpoint(stream.current_gtid)

An UPDATE arrives with both images of the row. type is the EventType.UPDATE enum member, not the string "UPDATE" — compare it with ==, not against a string:

ChangeEvent(
    type=EventType.UPDATE,
    database="shop",
    table="items",
    before={"id": 8, "name": "Widget", "value": 42},
    after={"id": 8, "name": "Widget", "value": 100},
    timestamp=1773584164,
    position=BinlogPosition(file="mysql-bin.000003", offset=3611),
    names_resolved=True,
    source_sql="",
)

CdcStream is the surface most applications want. Under it, BinlogClient owns the connection and hands back raw event bytes, and CdcEngine decodes binlog bytes from any source with no socket and no thread of its own. Bytes must start at an event boundary; a raw binlog file opens with a 4-byte magic number that has to be skipped first:

from mysql_event_stream import CdcEngine

# The engine holds native state, so scope it rather than waiting for the
# garbage collector to release it.
with CdcEngine() as engine:
    offset = 0
    while offset < len(chunk) or engine.has_events():
        while (event := engine.next_event()) is not None:
            print(event.type, event.database, event.table)

        if offset < len(chunk):
            consumed = engine.feed(chunk[offset:])
            offset += consumed

            # Nothing consumed and nothing queued: the tail is a partial
            # event. Keep chunk[offset:] and prepend it to the next chunk.
            if consumed == 0 and not engine.has_events():
                break

Server requirements

MySQL 8.4+ or MariaDB 10.11+, logging in row format with binlog_row_image=FULL and GTIDs enabled, and an account with REPLICATION SLAVE and REPLICATION CLIENT. Column names need either binlog_row_metadata=FULL or a SELECT grant on the streamed tables.

Server setup has the full configuration, the grants, and the replica identity rule that two processes on defaults will violate.

Documentation

• Python API — every class, option and default.
• Getting started and Examples.
• Change events and Column values — what arrives, and as what type.
• Checkpoints and recovery, Errors, Threading and lifecycle.
• Introduction indexes the rest.

Development

The binding is managed with Rye; requirements.lock and requirements-dev.lock are the source of truth for the development environment.

cd bindings/python
rye sync
rye run pytest

MES_LIB_PATH selects a specific libmes, which is what a source checkout needs when several builds are present.

Also available

npm install @libraz/mysql-event-stream  # Node.js binding

License

Apache-2.0