Alembic迁移向SQLAlchemy datetime字段插入NULL报错咨询

报错核心是字符串拼接SQL的类型处理缺陷：你用.format()拼接SQL时，无论传入"NULL"、"null"还是None，最终生成的SQL里对应位置都是被单引号包裹的字符串值（'NULL'/'None'），PostgreSQL会尝试把这些字符串解析为timestamp类型，自然触发格式错误。就算传入sqlalchemy.sql.null()，字符串格式化也会把它转成对象的字符串表示，无法被数据库识别为NULL值。
另外这种手动拼接SQL的写法本身就存在SQL注入、特殊字符转义失败的风险，哪怕是内部迁移脚本也不推荐使用。

直接用SQLAlchemy/Alembic原生支持的参数化查询即可，不需要重复写分支逻辑，也不需要导入业务模型，驱动会自动完成Python类型到数据库类型的映射，Python侧的None会被正确转换为数据库的NULL值：

from alembic import op

con = op.get_bind()
items = con.execute("SELECT * FROM public.A")

for item in items:
    updated_at = item.created_at
    # 直接计算deleted_at，未删除时传None即可
    deleted_at = item.created_at if item.deleted else None

    # 插入表B，使用命名参数占位，不要手动拼接字符串
    op.execute(
        """
        INSERT INTO B (id, some_attributes, created_at, updated_at, deleted_at)
        VALUES (:b_id, :some_attributes, :created_at, :updated_at, :deleted_at)
        ON CONFLICT DO NOTHING
        """,
        {
            "b_id": item.b,
            "some_attributes": item.some_attributes,
            "created_at": item.created_at,
            "updated_at": updated_at,
            "deleted_at": deleted_at
        }
    )

    # 插入表C
    op.execute(
        """
        INSERT INTO C (id, some_other_attributes, created_at, updated_at, deleted_at, B_id)
        VALUES (:c_id, :some_other_attributes, :created_at, :updated_at, :deleted_at, :b_id)
        """,
        {
            "c_id": item.c,
            "some_other_attributes": item.some_other_attributes,
            "created_at": item.created_at,
            "updated_at": updated_at,
            "deleted_at": deleted_at,
            "b_id": item.b
        }
    )

方案优势 #

• 不需要为NULL值单独写分支逻辑，代码简洁无冗余
• 不需要依赖业务模型类，不会因为后续业务模型迭代导致历史迁移脚本失效
• 数据库驱动自动处理类型转换、特殊字符转义，不会出现时间格式解析错误、SQL语法错误，也避免了SQL注入风险
• 迁移脚本可重入，执行逻辑和数据类型绑定更稳定

• 分支重复编写插入SQL：代码冗余度高，后续调整插入字段时需要同步修改多个分支，极易出现两边逻辑不一致的问题，维护成本高
• 导入业务模型类执行插入：Alembic迁移是版本化的独立快照，后续业务模型迭代（比如删字段、改字段类型）后，旧迁移脚本导入新模型会直接运行失败，破坏整个迁移链路的可执行性，是迁移脚本开发的典型反模式。

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