SQLite迁移PostgreSQL时执行INSERT操作触发PRAGMA语法错误

看起来你遇到的问题根源不在迁移代码本身，而是全局的SQLAlchemy连接事件监听在搞鬼！从错误栈里能看到，你的src/app/__init__.py里有个set_sqlite_pragma函数，它会在任何数据库引擎创建连接时执行PRAGMA foreign_keys=ON——但这个语句是SQLite专属的，PostgreSQL根本不认识它，所以当PostgreSQL引擎建立连接时，就抛出了语法错误。

问题定位 #

你应该在项目的初始化文件里写了类似这样的全局事件监听：

from sqlalchemy import event
from sqlalchemy.engine import Engine

@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()

这个监听是对所有Engine实例生效的，不管你连的是SQLite还是PostgreSQL，都会触发执行PRAGMA语句，这就导致了PostgreSQL连接时的语法错误。

解决方案：让PRAGMA只对SQLite生效 #

修改这个事件监听函数，只针对SQLite的连接执行PRAGMA操作，对PostgreSQL连接忽略即可。这里有两种简单的判断方式：

方式1：通过数据库连接对象类型判断

from sqlalchemy import event
from sqlalchemy.engine import Engine
import sqlite3
import psycopg2

@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
    # 仅当连接是SQLite时执行PRAGMA
    if isinstance(dbapi_connection, sqlite3.Connection):
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA foreign_keys=ON")
        cursor.close()
    # PostgreSQL连接直接跳过
    elif isinstance(dbapi_connection, psycopg2.extensions.connection):
        pass

方式2：通过SQLAlchemy方言判断（更通用）

from sqlalchemy import event
from sqlalchemy.engine import Engine

@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
    # 通过引擎的方言名称判断是否为SQLite
    if connection_record.engine.dialect.name == 'sqlite':
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA foreign_keys=ON")
        cursor.close()

额外的小建议 #

修改完事件监听后，你可以再测试迁移代码。另外，如果你只是简单插入数据，也可以尝试用pandas的to_sql方法，代码会更简洁：

# 替换原来的Session执行代码
df_Grupo.to_sql(
    name=Grupo.__tablename__,  # 直接复用模型定义的表名
    con=postgres_engine,
    if_exists='append',  # 根据需求选择：replace/append/fail
    index=False
)

这样处理后，PostgreSQL连接就不会再执行SQLite专属的PRAGMA语句，语法错误也就消失啦！

内容来源于stack exchange