watchful_snake.py

#!/usr/bin/env python3

# Watchful Snake is a MySQL BinLog to Redis Streams Data Pipeline.

# Watchful Snake is licensed under the
# GNU General Public License Version 3.
# Dimitar D. Mitov 2021
# https://github.com/ddmitov/watchful-snake

# Python Standard Library modules:
import json

# PyPI modules:
import pymysql

import redis

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

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

REDIS_HOST = '127.0.0.1'
REDIS_PORT = 6379
REDIS_DB = 0
REDIS_PASSWORD = ''

PREFIX = 'watchful_snake:mysql'


def redis_stream_appender(redis_server, schema, table, action, data):
    return redis_server.xadd(
        PREFIX + ':' + schema + ':' + table + ':' + action,
        data
    )


def mysql_binlog_watcher():
    # Redis connection:
    redis_server = redis.Redis(
        host=REDIS_HOST,
        port=REDIS_PORT,
        db=REDIS_DB,
        password=REDIS_PASSWORD
    )

    try:
        # MySQL settings in Redis:
        mysql_settings = redis_server.get(PREFIX + ':settings')

        if mysql_settings is None:
            print('MySQL settings are not found!')
            exit(1)

        mysql_settings_json = json.loads(mysql_settings)

        mysql_connection_settings = mysql_settings_json['connection']
        mysql_schemas = mysql_settings_json['schemas']
        mysql_tables = mysql_settings_json['tables']

        # Last MySQL binary log file recorded in Redis:
        saved_binlog_file_bytes = redis_server.get(
            PREFIX + ':last_binlog_file'
        )

        saved_binlog_file = ''

        if saved_binlog_file_bytes is not None:
            saved_binlog_file = saved_binlog_file_bytes.decode('utf-8')

        # Last MySQL binary log position recorded in Redis:
        saved_binlog_position_bytes = redis_server.get(
            PREFIX + ':last_binlog_position'
        )

        saved_binlog_position = 0

        if saved_binlog_position_bytes is not None:
            saved_binlog_position = int(saved_binlog_position_bytes)
    except (ConnectionRefusedError, redis.ConnectionError):
        print('No Redis!')
        exit(1)

    mysql_binlog_stream = None

    # MySQL connection:
    try:
        mysql_binlog_stream = BinLogStreamReader(
            connection_settings=mysql_connection_settings,
            server_id=1,
            blocking=True,
            only_schemas=mysql_schemas,
            only_tables=mysql_tables,
            only_events=[
                DeleteRowsEvent,
                WriteRowsEvent,
                UpdateRowsEvent
            ],
            slave_uuid=MYSQL_BINLOG_SLAVE_UUID,
        )
    except (ConnectionRefusedError, ConnectionError):
        print('No MySQL!')
        exit(1)

    try:
        for binlogevent in mysql_binlog_stream:
            binlog_file = mysql_binlog_stream.log_file
            binlog_position = int(mysql_binlog_stream.log_pos)

            replicate = False

            binlog_file_number = binlog_file.split('.')[-1]
            binlog_file_number = binlog_file_number.lstrip('0')
            binlog_file_number = int(binlog_file_number)

            saved_binlog_file_number = 0

            if len(saved_binlog_file) > 0:
                saved_binlog_file_number = saved_binlog_file.split('.')[-1]
                saved_binlog_file_number = (
                    saved_binlog_file_number.lstrip('0')
                )
                saved_binlog_file_number = int(saved_binlog_file_number)

            if binlog_file == saved_binlog_file:
                if binlog_position >= saved_binlog_position:
                    replicate = True

            if binlog_file != saved_binlog_file:
                if binlog_file_number >= saved_binlog_file_number:
                    replicate = True

            if replicate is True:
                print(binlog_file)
                print(binlog_position)

                schema = binlogevent.schema
                table = binlogevent.table

                print(schema + '.' + table)

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

                        print('delete')
                        print(values)

                        redis_stream_appender(
                            redis_server,
                            schema,
                            table,
                            'delete',
                            values
                        )

                    # Row update:
                    elif isinstance(binlogevent, UpdateRowsEvent):
                        combined_values = {
                            'before_values': {
                                row['before_values']
                            },
                            'after_values': {
                                row['after_values']
                            }
                        }

                        combined_values_json = json.dumps(
                            combined_values
                        )

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

                        redis_stream_appender(
                            redis_server,
                            schema,
                            table,
                            'update',
                            combined_values_json
                        )

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

                        print('insert')
                        print(values)

                        redis_stream_appender(
                            redis_server,
                            schema,
                            table,
                            'insert',
                            values
                        )

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

            # Last MySQL binary log position is
            # saved only after the last database update is
            # stored in Redis:
            if binlog_file == saved_binlog_file:
                if binlog_position >= saved_binlog_position:
                    redis_server.set(
                        PREFIX + ':last_binlog_position',
                        binlog_position
                    )

            # Last MySQL binary log file is
            # saved only after the last database update is
            # stored in Redis:
            if binlog_file != saved_binlog_file:
                if binlog_file_number >= saved_binlog_file_number:
                    redis_server.set(
                        PREFIX + ':last_binlog_file',
                        binlog_file
                    )
    except (pymysql.err.OperationalError):
        print('No MySQL!')
        exit(1)


def main():
    try:
        mysql_binlog_watcher()
    except (KeyboardInterrupt):
        print('')
        exit(0)


if __name__ == '__main__':
    main()