如何在SQLAlchemy执行Oracle插入时返回硬编码成功提示？

我定义了如下Oracle表结构：

create table myschema.src_table
(
dest_id integer, 
srce_value varchar2(20) not null primary key,
insert_status varchar2(100)
);
create table myschema.dest_table
(
id number GENERATED BY DEFAULT ON NULL AS IDENTITY,
value varchar2(20) not null,
constraint dest_table_pk primary key(id)
);

随后我向myschema.src_table中插入了1000条数据：

insert into myschema.src_table(srce_value) values (<<various non-unique string>>);
insert into myschema.src_table(srce_value) values (<<various non-unique string>>);
...

我的现有Python代码如下：

from sqlalchemy import text, Table, Column, Integer, MetaData
import pandas as pd
if __name__=='__main__':
     src_frame = df.read_sql('select src_value as value from myschema.src_table, con=oraEngine)
     metadata = sa.MetaData('myschema')
     metadata.create_all(oraEngine)
     dest_table= Table('dest_table',metadata, Column('id', Integer, primary_key=True), 
     Column('value', Integer), schema='myschema')
     with oraEgine.connect() as con:
       result=con.execute(sa.insert(dest_table).returning(dest_table), src_frame.to_dict(orient='records'))
     con.commit()

请问是否可以在执行插入操作时返回类似'INSERT DATA SUCCESS'的硬编码值，就像如下SQL语句实现的效果：

insert into myschema.dest_table(value) values ('this string') 
returning 'INSERT DATA SUCCESS' INTO :da_status;

可以实现，只需在SQLAlchemy的insert语句中，通过returning()方法直接指定硬编码字符串即可，逻辑和Oracle原生SQL一致。先修正你现有代码中的几处错误：

• 缺少import sqlalchemy as sa导入
• df.read_sql应为pd.read_sql，且SQL语句的括号、参数格式有误
• dest_table的value字段类型错误，原表是varchar2(20)，需用String(20)而非Integer
• 连接对象拼写错误oraEgine应为oraEngine

修正并实现返回硬编码状态的完整代码如下：

from sqlalchemy import text, Table, Column, Integer, String, MetaData
import sqlalchemy as sa
import pandas as pd

if __name__=='__main__':
    # 读取源表数据
    src_frame = pd.read_sql('select srce_value as value from myschema.src_table', con=oraEngine)
    
    # 映射目标表结构
    metadata = sa.MetaData(schema='myschema')
    dest_table = Table(
        'dest_table', metadata,
        Column('id', Integer, primary_key=True),
        Column('value', String(20), nullable=False),
        schema='myschema'
    )
    
    # 执行插入并返回硬编码状态
    with oraEngine.connect() as con:
        # 指定返回硬编码字符串，别名status方便获取
        insert_stmt = sa.insert(dest_table).values(src_frame.to_dict(orient='records')).returning(text("'INSERT DATA SUCCESS' AS status"))
        result = con.execute(insert_stmt)
        # 获取所有返回的状态值（批量插入时每条数据对应一条状态）
        status_list = [row[0] for row in result.fetchall()]
        con.commit()
    
    # 示例：打印前5条状态
    print(status_list[:5])

关键说明 #

• returning(text("'INSERT DATA SUCCESS' AS status"))：用text()包裹原生SQL片段，直接在返回子句中传入硬编码字符串，AS status给返回值起别名，方便后续读取。
• 批量插入时，每成功插入一条数据都会返回一条INSERT DATA SUCCESS，你可以根据需求选择获取全部结果，或通过插入行数判断操作是否成功。

内容的提问来源于stack exchange，提问作者Z T Minhas