如何在Python控制台中从Oracle动态SQL存储过程返回列

问题背景 #

我正在开发一个使用Oracle存储过程实现CRUD操作的应用，在READ模块中，编写了一个可根据指定属性返回一行或多行完整数据的存储过程，该过程在Oracle SQL Developer中可正常运行，但在Python中无法获取并显示数据。使用fetchall()方法时会返回“not a query”错误，其他方法则完全无结果。

Python代码（READ模块） #

def verEmpleado():
    opcionCorrecta = False
    while(not opcionCorrecta):
        print('================= 查询 =================')
        print('[1] 员工ID')
        print('[2] 员工姓名')
        print('[3] 职位')
        print('[4] 经理')
        print('[5] 雇佣日期')
        print('[6] 薪资')
        print('[7] 佣金')
        print('[8] 部门ID')
        print('============================================')

        command = int(input('请选择列：'))
        match command:
            case 1:
                p_filter_att = 'empno'
                opcionCorrecta = True
            case 2:
                p_filter_att = 'ename'
                opcionCorrecta = True
            case 3:
                p_filter_att = 'job'
                opcionCorrecta = True
            case 4:
                p_filter_att = 'mgr'
                opcionCorrecta = True
            case 5:
                p_filter_att = 'hiredate'
                opcionCorrecta = True
            case 6:
                p_filter_att = 'sal'
                opcionCorrecta = True
            case 7:
                p_filter_att = 'comm'
                opcionCorrecta = True
            case 8:
                p_filter_att = 'deptno'
                opcionCorrecta = True
            case _:
                print('命令错误，请重试...\n')


    p_value_att = input('请输入要筛选的列值：')

    data = [p_filter_att, p_value_att]
    return data

#连接数据库
def connVerEmpleado(data):
        try:
            conn = cx_Oracle.connect('HR/hr@localhost:1521/xepdb1')
        except Exception as err:
            print('创建连接时异常:', err)
        else:
            try:
                cursor = conn.cursor()
                cursor.callproc('query_emp', data)

                result = cursor.fetchall()
                print(result)
                cursor.close()

            except Exception as err:
                print('连接错误:', err)

Oracle存储过程代码 #

create or replace PROCEDURE query_emp(p_filter_att IN VARCHAR2, p_value_att IN VARCHAR2)
IS
    sql_qry              VARCHAR2(1000);
    TYPE bc_v_empno      IS TABLE OF NUMBER(4);
    v_empno              bc_v_empno;
    TYPE bc_v_ename      IS TABLE OF VARCHAR2(10);
    v_ename              bc_v_ename;
    TYPE bc_v_job        IS TABLE OF VARCHAR2(9);
    v_job                bc_v_job;
    TYPE bc_v_mgr        IS TABLE OF NUMBER(4);
    v_mgr                bc_v_mgr;
    TYPE bc_v_hiredate   IS TABLE OF DATE;
    v_hiredate           bc_v_hiredate;
    TYPE bc_v_sal        IS TABLE OF NUMBER(7,2);
    v_sal                bc_v_sal;
    TYPE bc_v_comm       IS TABLE OF NUMBER(7,2);
    v_comm               bc_v_comm;
    TYPE bc_v_deptno     IS TABLE OF NUMBER(2);
    v_deptno             bc_v_deptno;
BEGIN
    sql_qry := 'SELECT * FROM emp WHERE ' || p_filter_att || ' = UPPER(:p_value_att)';
    EXECUTE IMMEDIATE sql_qry BULK COLLECT INTO v_empno, v_ename, v_job, v_mgr, v_hiredate, v_sal, v_comm, v_deptno
    USING p_value_att;

    FOR i IN 1..v_empno.COUNT
        LOOP
            dbms_output.put_line('ID: ' || v_empno(i) || ' | NAME: ' || v_ename(i) || ' | JOB: ' || v_job(i) || ' | MANAGER: ' || v_mgr(i) || ' | HIRE DATE: ' || v_hiredate(i) || ' | SALARY: ' || v_sal(i) || ' | COMMISSION: ' || v_comm(i) || ' | DEPARTMENT: ' || v_deptno(i));
        END LOOP;

END query_emp;
/

问题原因及解决方法 #

原因分析 #

当前存储过程仅通过dbms_output.put_line()在Oracle服务器端打印结果，并未将数据返回给调用端（Python）；Python调用callproc后使用fetchall()，由于存储过程没有返回游标或结果集，因此触发“not a query”错误。

解决步骤 #

1. 修改Oracle存储过程，添加输出游标参数

调整存储过程，定义输出游标将查询结果返回给调用方，替代原有的服务器端打印逻辑：

create or replace PROCEDURE query_emp(
    p_filter_att IN VARCHAR2, 
    p_value_att IN VARCHAR2,
    p_result OUT SYS_REFCURSOR  -- 添加输出游标参数
)
IS
    sql_qry VARCHAR2(1000);
BEGIN
    -- 校验筛选列合法性，避免SQL注入
    IF p_filter_att NOT IN ('empno','ename','job','mgr','hiredate','sal','comm','deptno') THEN
        RAISE_APPLICATION_ERROR(-20001, '非法的筛选列');
    END IF;
    
    sql_qry := 'SELECT * FROM emp WHERE ' || p_filter_att || ' = UPPER(:p_value_att)';
    OPEN p_result FOR sql_qry USING p_value_att;  -- 打开游标关联查询语句
END query_emp;
/

2. 修改Python代码，接收并处理输出游标

更新connVerEmpleado函数，指定输出游标参数，从游标中读取返回结果：

def connVerEmpleado(data):
    try:
        conn = cx_Oracle.connect('HR/hr@localhost:1521/xepdb1')
    except Exception as err:
        print('创建连接时异常:', err)
    else:
        try:
            cursor = conn.cursor()
            # 定义输出游标变量
            output_cursor = cursor.var(cx_Oracle.CURSOR)
            # 调用存储过程，传入输入参数和输出游标
            cursor.callproc('query_emp', [data[0], data[1], output_cursor])
            
            # 从游标中获取结果并打印
            result = output_cursor.getvalue().fetchall()
            for row in result:
                print(f'ID: {row[0]} | NAME: {row[1]} | JOB: {row[2]} | MANAGER: {row[3]} | HIRE DATE: {row[4]} | SALARY: {row[5]} | COMMISSION: {row[6]} | DEPARTMENT: {row[7]}')
            
            # 释放资源
            cursor.close()
            conn.close()
        except Exception as err:
            print('操作错误:', err)

额外注意事项 #

• SQL注入防护：存储过程中添加了筛选列白名单校验，避免恶意参数拼接导致的SQL注入风险。
• 资源管理：Python代码中添加了数据库连接关闭操作，避免资源泄漏。

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