如何用SQLAlchemy 1.4实现Oracle的条件多表INSERT ALL语句？

针对你的问题，SQLAlchemy 1.4可以通过两种方式实现Oracle的INSERT ALL/FIRST语法，完全规避客户端拉取数据后循环插入的内存和性能开销：

方法1：直接执行原生Oracle文本SQL #

如果不需要ORM的类型校验，直接复用Oracle原生语法是最快速的实现方式：

from sqlalchemy import text

# 编写匹配需求的原生INSERT ALL语句
insert_sql = """
INSERT ALL
    WHEN condition1 THEN
        INTO table_1 (col1, col2) VALUES (val1, val2)
    WHEN condition2 THEN
        INTO table_2 (col1, col2) VALUES (val1, val2)
    ELSE
        INTO table_3 (col1, col2) VALUES (val1, val2)
SELECT val1, val2 FROM source_table WHERE your_filter_condition
"""

# 执行语句（使用begin保证事务原子性）
with engine.begin() as conn:
    conn.execute(text(insert_sql))

这种方式完全和Oracle原生语句对齐，性能与直接执行SQL一致，没有额外的客户端处理开销。

方法2：用SQLAlchemy Core构建类型安全的表达式 #

如果希望利用SQLAlchemy的模型映射和类型检查，1.4版本的Oracle方言提供了专门的扩展语法支持：

首先定义你的表结构（以Core为例）：

from sqlalchemy import Table, Column, Integer, String, MetaData, func
from sqlalchemy.dialects.oracle import insert

metadata = MetaData()

# 源表
source_table = Table(
    "source_table", metadata,
    Column("id", Integer, primary_key=True),
    Column("col1", String(50)),
    Column("col2", Integer)
)

# 目标表
table_1 = Table("table_1", metadata, Column("col1", String(50)), Column("col2", Integer))
table_2 = Table("table_2", metadata, Column("col1", String(50)), Column("col2", Integer))
table_3 = Table("table_3", metadata, Column("col1", String(50)), Column("col2", Integer))

然后构建INSERT ALL表达式：

# 定义子查询
subquery = source_table.select().where(source_table.c.id > 100)  # 替换为你的过滤条件

# 构建多表插入语句
insert_stmt = insert().all(
    # 第一个条件分支
    insert(table_1).values(col1=subquery.c.col1, col2=subquery.c.col2).when(subquery.c.col2 > 10),
    # 第二个条件分支
    insert(table_2).values(col1=subquery.c.col1, col2=subquery.c.col2).when(subquery.c.col2 <= 10),
    # ELSE分支（用when(True)表示默认匹配）
    insert(table_3).values(col1=subquery.c.col1, col2=subquery.c.col2).when(True)
).from_select(subquery.columns, subquery)

# 执行语句
with engine.begin() as conn:
    conn.execute(insert_stmt)

如果需要使用INSERT FIRST逻辑，只需将.all()替换为.first()即可。

核心优势 #

• 两种方案都将所有逻辑放在数据库端执行，避免了客户端拉取大量数据的内存消耗和循环插入的性能损耗
• 方法2通过SQLAlchemy的类型安全机制，减少了硬编码SQL的拼写错误风险，同时保持代码的可维护性

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