extract_mysql_binlog_stream_to_csv.py

# Important Notice: This script requires the user
# to have the appropriate permissions.
# SUPER or REPLICATION CLIENT
from pymysqlreplication import BinLogStreamReader
from pymysqlreplication import row_event
import configparser
import pymysqlreplication
import csv
import boto3

# get the MySQL connection info

parser = configparser.ConfigParser()
parser.read('pipeline.conf')
hostname = parser.get('mysql_config', 'hostname')
port = parser.get('mysql_config', 'port')
username = parser.get('mysql_config', 'username')
password = parser.get('mysql_config', 'password')

mysql_settings = {
    'host': hostname,
    'port': int(port),
    'user': username,
    'passwd': password
}

b_stream = BinLogStreamReader(
    connection_settings=mysql_settings,
    server_id=100,
    only_events=[
        row_event.DeleteRowsEvent,
        row_event.WriteRowsEvent,
        row_event.UpdateRowsEvent
    ]
)

order_events = []

for binlogevent in b_stream:
    for row in binlogevent.rows:  # type: ignore
        if binlogevent.table == 'Orders':  # type: ignore
            event = {}
            if isinstance(binlogevent, row_event.DeleteRowsEvent):
                event['action'] = 'delete'
                event.update(row['values'].items())
            elif isinstance(binlogevent, row_event.UpdateRowsEvent):
                event['action'] = 'update'
                event.update(row['after_values'].items())
            elif isinstance(binlogevent, row_event.WriteRowsEvent):
                event['action'] = 'insert'
                event.update(row['values'].items())
            order_events.append(event)
b_stream.close()

keys = order_events[0].keys()
local_filename = 'orders_extract.csv'
with open(
    local_filename,
    'w',
    newline=''
) as output_file:
    dict_writer = csv.DictWriter(output_file, keys, delimiter='|')
    dict_writer.writerows(order_events)


acces_key = parser.get('aws_boto_credentials', 'access_key')
secret_key = parser.get('aws_boto_credentials', 'secret_key')
bucket_name = parser.get('aws_boto_credentials', 'bucket_name')

s3 = boto3.client('s3', aws_access_key_id=acces_key,
                  aws_secret_access_key=secret_key)
s3_file = local_filename
s3.upload_file(local_filename, bucket_name, s3_file)