如何在SQLAlchemy中获取存储过程输出的复杂Oracle自定义类型

你遇到的问题核心在于SQLAlchemy本身不会自动识别Oracle的自定义类型，得先通过底层的cx_Oracle驱动加载这些类型，再正确绑定参数。下面是一步步的解决方案：

步骤1：获取原生cx_Oracle连接并加载自定义类型 #

SQLAlchemy的连接是封装过的，你需要先拿到底层的原生cx_Oracle连接，然后加载你的自定义对象类型和表类型：

from sqlalchemy import text, outparam
import cx_Oracle
import logging

# 假设engine是你已经初始化好的SQLAlchemy引擎
connection = engine.connect()
try:
    # 获取底层原生cx_Oracle连接
    raw_conn = connection.connection
    
    # 加载自定义类型：先加载对象类型TYPE1，再加载表类型T_TYPE1
    # 注意：如果创建类型时用了双引号区分大小写，这里要严格匹配名称
    type1_obj = raw_conn.gettype("TYPE1")
    t_type1_obj = raw_conn.gettype("T_TYPE1")
    
    # 定义转换函数：把Oracle自定义对象转成Python字典，方便后续处理
    def type1_to_dict(obj):
        return {
            "param1": obj.PARAM1,
            # 处理CLOB类型：调用read()获取文本内容
            "param2": obj.PARAM2.read() if obj.PARAM2 is not None else None
        }

步骤2：正确绑定输出参数并调用存储过程 #

接下来修改你的存储过程调用代码，在outparam中指定加载好的t_type1_obj作为类型，同时处理其他输出参数：

# 构建存储过程调用的PL/SQL语句
    proc_query = text("""
        BEGIN
            DO_SOMETHING(:results, :status, :status_msg);
        END;
    """)
    
    # 绑定输出参数：results指定为加载好的表类型，其余参数用cx_Oracle预定义类型
    bind_params = [
        outparam("results", type_=t_type1_obj, isout=True),
        outparam("status", cx_Oracle.DB_TYPE_CHAR, isout=True),
        outparam("status_msg", cx_Oracle.DB_TYPE_VARCHAR, isout=True)
    ]
    
    # 执行存储过程
    result = connection.execute(proc_query, bind_params)
    
    # 提取输出参数并格式化
    raw_results = result.out_parameters["results"]
    status = result.out_parameters["status"].strip()
    status_msg = result.out_parameters["status_msg"].strip()
    
    # 将Oracle表类型的结果转换为Python列表字典
    formatted_results = [type1_to_dict(item) for item in raw_results]
    
    # 打印结果
    print(f"执行状态: {status}")
    print(f"状态信息: {status_msg}")
    print("返回结果列表:")
    for idx, res in enumerate(formatted_results):
        print(f"  条目{idx+1}: {res}")
        
except Exception as e:
    logging.exception(e)
    print("调用存储过程出错:", e)
    raise e
finally:
    connection.close()

关键说明 #

• 加载自定义类型：必须通过原生cx_Oracle连接的gettype方法加载，因为SQLAlchemy无法直接识别这些自定义类型，这正是你之前收到"该类型不被cx_oracle支持"错误的核心原因。
• CLOB类型处理：自定义对象中的param2是CLOB类型，需要调用read()方法来获取其文本内容，否则拿到的只是cx_Oracle的CLOB对象实例，不是可读的字符串。
• 参数绑定：outparam中要明确指定type_为加载好的表类型t_type1_obj，其他基础类型参数可以直接使用cx_Oracle提供的预定义类型常量（比如cx_Oracle.DB_TYPE_CHAR、cx_Oracle.DB_TYPE_VARCHAR）。

额外注意：确保你的Oracle数据库用户拥有访问这些自定义类型和存储过程的权限；如果创建对象时使用了双引号区分大小写，调用gettype时要严格匹配名称的大小写。

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