python寫了一個簡單的mysql數據同步腳本,只作為學習練習,大佬勿噴

# -*- coding: utf-8 -*-
"""
@Time:2025/5/29 14:38
@Auth:HEhandsome
"""
import pymysql
from pymysql import Connectclass Mysql:def __init__(self):#源數據庫self.sou_host = 'sou_host'self.sou_user = 'sou_user'self.sou_port = 3306self.sou_password = 'sou_password'self.sou_database = 'sou_database'#目標數據庫self.tag_host = 'tag_host'self.tag_user = 'tag_user'self.tag_port = 3306self.tag_password = 'tag_password'self.tag_database = 'tag_database'#存儲表結構self.create_table_sql=Noneself.select_table_sql=Noneself.columns=None#連接源數據庫和目標數據庫self.sou_conn=pymysql.Connect(host=self.sou_host,user=self.sou_user,port=self.sou_port,password=self.sou_password,database=self.sou_database)self.tag_conn=pymysql.Connect(host=self.tag_host,user=self.tag_user,port=self.tag_port,password=self.tag_password,database=self.tag_database)def get_source_db(self):try:with self.sou_conn.cursor() as cursor:#獲取表結構cursor.execute('SHOW CREATE TABLE users')# print(cursor.fetchone())result = cursor.fetchone()self.create_table_sql = result[1]except Exception as e:print(f'獲取失敗:{e}')def get_tag_db(self):with self.tag_conn.cursor() as cursor:cursor.execute('DROP TABLE IF EXISTS users')cursor.execute(self.create_table_sql)self.tag_conn.commit()def read_sou_data(self):with self.sou_conn.cursor() as cursor:#獲取表結構cursor.execute('SELECT * FROM users')result = cursor.fetchall()self.select_table_sql = result#獲取當前查詢結果的所有列名，形成一個列表self.columns = [desc[0] for desc in cursor.description]  # 獲取列名def write_tag_db(self):with self.tag_conn.cursor() as cursor:# 構建插入語句placeholders = ','.join(['%s'] * len(self.columns))cols = ','.join(self.columns)sql = f"INSERT INTO users ({cols}) VALUES ({placeholders})"#插入數據cursor.executemany(sql, self.select_table_sql)self.tag_conn.commit()print(f"成功插入 {cursor.rowcount} 條記錄")
if __name__ == '__main__':db = Mysql()db.get_source_db()db.get_tag_db()db.read_sou_data()db.write_tag_db()