如何通过SQLAlchemy将本地SQLite数据行迁移至远程MySQL数据库？

嘿，我来帮你搞定把SQLite数据追加到GCP MySQL的事儿～ 基于你已经定义好的Readings模型，咱们可以通过以下步骤实现数据同步：

步骤1：创建两个数据库的连接会话 #

首先要分别建立本地SQLite和远程MySQL的引擎与会话，记得替换MySQL的连接信息为你自己的GCP实例配置：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from SensorCollection import Readings, Base

# 本地SQLite数据库连接（替换成你的DB路径）
sqlite_engine = create_engine('sqlite:///local_sensor.db')
sqlite_session = sessionmaker(bind=sqlite_engine)()

# GCP MySQL数据库连接（替换用户名、密码、主机名、库名）
mysql_engine = create_engine('mysql+pymysql://your_user:your_pass@gcp-mysql-host:3306/remote_sensor_db')
mysql_session = sessionmaker(bind=mysql_engine)()

步骤2：读取SQLite中的数据 #

直接通过ORM查询获取所有需要同步的行：

# 拉取SQLite里的全部读数数据
sqlite_records = sqlite_session.query(Readings).all()

步骤3：批量插入到MySQL（处理主键冲突） #

因为time是主键，同步时大概率会遇到重复主键的情况，这里提供两种常用处理方式：

方式一：跳过已存在的记录（适合只追加新数据的场景） #

逐条检查MySQL中是否存在对应主键，不存在则插入：

for record in sqlite_records:
    # 检查当前时间戳的记录是否已存在
    exists = mysql_session.query(Readings).filter_by(time=record.time).first()
    if not exists:
        mysql_session.add(record)

# 提交事务完成同步
mysql_session.commit()

方式二：重复时更新字段（需要同步最新数据的场景） #

如果数据量较大，逐条检查效率低，可以用MySQL的ON DUPLICATE KEY UPDATE批量处理：

from sqlalchemy.dialects.mysql import insert

# 将ORM对象转换为字典列表，方便批量插入
record_dicts = [
    {
        "time": r.time,
        "box_name": r.box_name,
        "FS": r.FS,
        "IS": r.IS,
        "VS": r.VS
        # 补充你模型里的其他字段
    }
    for r in sqlite_records
]

# 构建插入语句，主键重复时更新指定字段
insert_stmt = insert(Readings).values(record_dicts)
update_stmt = insert_stmt.on_duplicate_key_update(
    box_name=insert_stmt.inserted.box_name,
    FS=insert_stmt.inserted.FS,
    IS=insert_stmt.inserted.IS,
    VS=insert_stmt.inserted.VS
    # 对应需要更新的字段
)

# 执行批量操作并提交
mysql_session.execute(update_stmt)
mysql_session.commit()

步骤4：清理资源 #

操作完成后记得关闭两个会话，释放连接：

sqlite_session.close()
mysql_session.close()

额外提醒 #

• GCP MySQL需要配置访问权限（比如本地IP白名单、Cloud SQL代理）才能正常连接
• 如果数据量极大，建议分批次读取插入，避免内存溢出
• 可以添加日志打印，方便跟踪同步进度和排查问题

内容的提问来源于stack exchange，提问作者James Matherly