mysql_binlog_reader.py

#!/usr/bin/env python3

# Snake Works 2020.
# https://github.com/ddmitov/
#
#               /^\/^\
#             _| O|  O|
#    \/     /   \_/\_/ \
#     \____|____         \
#            \_______      \        ______
#                    `\     \      /      \
#                      |     |     \______/
#                     /      /     /      \
#                    /     /       \______/
#                  /      /        /      \
#                 /     /          \______/
#               /     /             _----_
#              /     /           _-~      ~-_          / \
#             (      (        _-~    _--_    ~-_     _/   )
#              \      ~-____-~    _-~    ~-_    ~-_-~    /
#                ~-_           _-~          ~-_       _-~
#                   ~--______-~                ~-___-~   - jurcy -
#
# https://asciiart.website/index.php?art=animals/reptiles/snakes

import os

from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import (
    DeleteRowsEvent,
    UpdateRowsEvent,
    WriteRowsEvent,
)

MYSQL_SETTINGS = {
    'host': '127.0.0.1',
    'port': 3306,
    'user': 'user',
    'passwd': 'password'
}

MYSQL_DB = 'database'
MYSQL_TABLE = 'table'

# Generated by the 'uuidgen' utility:
SLAVE_UUID = 'e7d14803-b12f-45eb-9c33-abd2e5e9ac77'

LAST_LOG_FILE = 'mysql-bin.000009'
LAST_LOG_POSITION = 955


def main():
    try:
        stream = BinLogStreamReader(
            connection_settings=MYSQL_SETTINGS,
            server_id=1,
            blocking=True,
            only_schemas=[MYSQL_DB],
            only_tables=[MYSQL_TABLE],
            only_events=[
                DeleteRowsEvent,
                WriteRowsEvent,
                UpdateRowsEvent
            ],
            slave_uuid=SLAVE_UUID,
        )

        for binlogevent in stream:
            log_file = stream.log_file
            log_position = stream.log_pos

            replicate = False

            if log_file == LAST_LOG_FILE:
                if log_position > LAST_LOG_POSITION:
                    replicate = True
            else:
                log_file_number = int(log_file.split('.')[-1])
                last_log_file_number = int(LAST_LOG_FILE.split('.')[-1])

                if log_file_number > last_log_file_number:
                    replicate = True

            if replicate is True:
                print(log_position)

                database = binlogevent.schema
                table = binlogevent.table

                print(database + '.' + table)

                for row in binlogevent.rows:
                    if isinstance(binlogevent, DeleteRowsEvent):
                        values = row['values']

                        print('delete')
                        print(values)

                    elif isinstance(binlogevent, UpdateRowsEvent):
                        old_values = row['before_values']
                        new_values = row['after_values']

                        print('update')
                        print(old_values)
                        print(new_values)

                    elif isinstance(binlogevent, WriteRowsEvent):
                        values = row['values']

                        print('insert')
                        print(values)

                print('==============================')

        stream.close()
    except (KeyboardInterrupt, SystemExit):
        print('\n')
        exit(0)


if __name__ == '__main__':
    main()