多地数据同步服务搭建

import pymysql
from pymysqlreplication import BinLogStreamReader

# 配置源数据库连接
source_db_config = {
    "host": "source_host",
    "port": 3306,
    "user": "source_user",
    "passwd": "source_password",
    "db": "source_db"
}

# 配置目标数据库连接
target_db_config = {
    "host": "target_host",
    "port": 3306,
    "user": "target_user",
    "passwd": "target_password",
    "db": "target_db"
}

def sync_data():
    stream = BinLogStreamReader(
        connection_settings=source_db_config,
        server_id=100,
        blocking=True,
        only_events=[DeleteRowsEvent, UpdateRowsEvent, WriteRowsEvent]
    )

    for binlogevent in stream:
        for row in binlogevent.rows:
            event = {"schema": binlogevent.schema, "table": binlogevent.table}
            if isinstance(binlogevent, WriteRowsEvent):
                event["action"] = "insert"
                event["values"] = row["values"]
            elif isinstance(binlogevent, UpdateRowsEvent):
                event["action"] = "update"
                event["before_values"] = row["before_values"]
                event["after_values"] = row["after_values"]
            elif isinstance(binlogevent, DeleteRowsEvent):
                event["action"] = "delete"
                event["values"] = row["values"]

            apply_change(event)

    stream.close()

def apply_change(event):
    conn = pymysql.connect(**target_db_config)
    cursor = conn.cursor()

    if event["action"] == "insert":
        columns = ", ".join(event["values"].keys())
        values = tuple(event["values"].values())
        sql = f"INSERT INTO {event['schema']}.{event['table']} ({columns}) VALUES {values}"
        cursor.execute(sql)
    elif event["action"] == "update":
        set_clause = ", ".join([f"{k} = %s" for k in event["after_values"].keys()])
        where_clause = " AND ".join([f"{k} = %s" for k in event["before_values"].keys()])
        sql = f"UPDATE {event['schema']}.{event['table']} SET {set_clause} WHERE {where_clause}"
        cursor.execute(sql, list(event["after_values"].values()) + list(event["before_values"].values()))
    elif event["action"] == "delete":
        where_clause = " AND ".join([f"{k} = %s" for k in event["values"].keys()])
        sql = f"DELETE FROM {event['schema']}.{event['table']} WHERE {where_clause}"
        cursor.execute(sql, tuple(event["values"].values()))

    conn.commit()
    cursor.close()
    conn.close()

if __name__ == "__main__":
    sync_data()