如何让Python捕获Oracle PL/SQL存储过程抛出的异常？

我正在开发一个采用Oracle存储过程实现CRUD操作的应用，每个存储过程都包含异常处理逻辑，其中一个还关联了触发器，但在使用Python代码调用时，这些异常均未被处理。例如，当尝试输入重复的ID（ID应为唯一值）时，在Oracle中执行会正常触发异常处理，但从Python执行时，系统显示记录已插入，而实际并未插入。

Python插入代码 #

import cx_Oracle

#INSERT EMPLOYEE
def crearEmpleado():
    validEmpno = False
    while(not validEmpno):
        p_empno = input('Ingrese el número de empleado: ')
        if p_empno.isnumeric() and int(p_empno) > 0 and int(p_empno) <= 999999:
            validEmpno = True
            p_empno = int(p_empno)
        else:
            print('Error')

    validEname = False
    while(not validEname):
        p_ename = input('Ingrese el nombre del empleado: ')
        if len(p_ename) > 0 and len(p_ename) <= 10:
            validEname = True
        else:
            print('Error')

    validJob = False
    while(not validJob):
        p_job = input('Ingrese el puesto del empleado: ')
        if len(p_job) > 0 and len(p_job) <= 9:
            validJob = True
        else:
            print('Error')
    
    validMgr = False
    while(not validMgr):
        p_mgr = input('Ingrese el número de manager: ')
        if p_mgr.isnumeric() and int(p_mgr) > 0 and int(p_mgr) <= 9999:
            validMgr = True
            p_mgr = int(p_mgr)
        else:
            print('Error')

    validHiredate = False
    while(not validHiredate):
        p_hiredate = input('Ingrese la fecha de contratación en formato DD/MM/YYYY: ')
        if len(p_hiredate) > 0 and len(p_hiredate) <= 10:
            validHiredate = True
        else:
            print('Error')

    validSal = False
    while(not validSal):
        p_sal = input('Ingrese el salario: ')
        if float(p_sal) > 0 and float(p_sal) <= 99999.99:
            validSal = True
            p_sal = float(p_sal)
        else:
            print('Error')

    validComm = False
    while(not validComm):
        p_comm = input('Ingrese el porcentaje de comisión: ')
        if float(p_comm) <= 99999.99:
            validComm = True
            p_comm = float(p_comm)
        else:
            print('Error')

    validDeptno = False
    while(not validDeptno):
        p_deptno = input('Ingrese el número de departamento: ')
        if p_deptno.isnumeric() and int(p_deptno) > 0 and int(p_deptno) <= 99:
            validDeptno = True
            p_deptno = int(p_deptno)
        else:
            print('Error')

    data = [p_empno, p_ename, p_job, p_mgr, p_hiredate, p_sal, p_comm, p_deptno]
    return data

#CONECT
def connCrearEmpleado(data):
        conn = cx_Oracle.connect('HR/hr@localhost:1521/xepdb1')
        try:
            cursor = conn.cursor()
            cursor.callproc('add_emp', data)
            print('\n¡Empleado registrado!\n')
        except cx_Oracle.DatabaseError as err:
            print('Database connection error: {}'.format(err))
            print(err.code)
            print(err.message)
            print(err.context)
            raise

数据表结构 #

CREATE TABLE emp (
    empno NUMBER(4) NOT NULL,
    ename VARCHAR2(10) NOT NULL,
    job VARCHAR2(9) NOT NULL,
    mgr NUMBER(4) NOT NULL,
    hiredate DATE NOT NULL,
    sal NUMBER(7,2) NOT NULL,
    comm NUMBER(7,2),
    deptno NUMBER(2) NOT NULL
)

PL/SQL存储过程 #

create or replace PROCEDURE add_emp
    (p_empno in NUMBER, p_ename in VARCHAR2, p_job in VARCHAR2, p_mgr in NUMBER, p_hiredate in VARCHAR2, p_sal in NUMBER, p_comm in NUMBER, p_deptno in NUMBER)

IS
    v_empno                 NUMBER;
    e_unique_empno          EXCEPTION;              
    e_notnull_empno         EXCEPTION;
    e_notnull_ename         EXCEPTION;
    e_notnull_job           EXCEPTION;
    e_notnull_mgr           EXCEPTION;
    e_notnull_hiredate      EXCEPTION;
    e_valid_sal             EXCEPTION;
    e_notnull_sal           EXCEPTION;  
    v_deptno                NUMBER;
    e_notnull_deptno        EXCEPTION;

BEGIN
    IF p_empno IS NULL THEN
        RAISE e_notnull_empno;
    ELSIF p_ename IS NULL THEN
        RAISE e_notnull_ename;
    ELSIF p_job IS NULL THEN
        RAISE e_notnull_job;
    ELSIF p_mgr IS NULL THEN
        RAISE e_notnull_mgr;
    ELSIF p_hiredate IS NULL THEN
        RAISE e_notnull_hiredate;
    ELSIF p_sal IS NULL THEN
        RAISE e_notnull_sal;
    ELSIF p_sal <= 0 THEN
        RAISE e_valid_sal;
    ELSIF p_deptno IS NULL THEN
        RAISE e_notnull_deptno;
    END IF;
    
    SELECT COUNT(*) INTO v_empno FROM emp WHERE p_empno = empno;
        
    IF v_empno > 0 THEN
        RAISE e_unique_empno;
    END IF;

    INSERT INTO emp
    VALUES(p_empno, UPPER(p_ename), UPPER(p_job), p_mgr, TO_DATE(p_hiredate, 'DD-MM-YYYY'), p_sal, p_comm, p_deptno);
    COMMIT; 

EXCEPTION
    WHEN e_unique_empno THEN
        dbms_output.put_line('El ID ya existe.');
    WHEN e_notnull_empno THEN
        dbms_output.put_line('El campo EMPNO no puede ser nulo.');
    WHEN e_notnull_ename THEN
        dbms_output.put_line('El campo ENAME no puede ser nulo.');
    WHEN e_notnull_job THEN
        dbms_output.put_line('El campo JOB no puede ser nulo.');
    WHEN e_notnull_mgr  THEN
        dbms_output.put_line('El campo MGR no puede ser nulo.');
    WHEN e_notnull_hiredate  THEN
        dbms_output.put_line('El campo HIRE_DATE no puede ser nulo.');
    WHEN e_valid_sal THEN
        dbms_output.put_line('El salario no puede ser menor o igual a cero.');  
    WHEN e_notnull_sal THEN
        dbms_output.put_line('El campo SALARY no puede ser nulo.');
    WHEN e_notnull_deptno THEN
        dbms_output.put_line('El campo DEPTNO no puede ser nulo.');
    WHEN OTHERS THEN
        dbms_output.put_line(SQLERRM);
END add_emp;
/

问题原因 #

存储过程中的异常处理仅使用dbms_output.put_line输出信息，该输出仅在Oracle服务器端可见，Python的cx_Oracle客户端无法捕获。同时，异常被捕获后没有重新抛出，相当于异常被“吞掉”，Python端感知不到错误，因此会执行成功提示逻辑。

解决办法 #

1. 重新抛出异常（推荐）

在每个异常处理分支中，输出信息后通过RAISE重新抛出异常，让cx_Oracle能捕获到错误：

EXCEPTION
    WHEN e_unique_empno THEN
        dbms_output.put_line('El ID ya existe.');
        RAISE; -- 重新抛出当前异常
    WHEN e_notnull_empno THEN
        dbms_output.put_line('El campo EMPNO no puede ser nulo.');
        RAISE;
    -- 其他异常分支同理
    WHEN OTHERS THEN
        dbms_output.put_line(SQLERRM);
        RAISE;

2. 使用自定义应用错误

用RAISE_APPLICATION_ERROR抛出带有自定义错误码的异常，客户端可直接捕获错误信息：

EXCEPTION
    WHEN e_unique_empno THEN
        RAISE_APPLICATION_ERROR(-20001, 'El ID ya existe.');
    WHEN e_notnull_empno THEN
        RAISE_APPLICATION_ERROR(-20002, 'El campo EMPNO no puede ser nulo.');
    -- 其他异常分支分配不同错误码
    WHEN OTHERS THEN
        RAISE_APPLICATION_ERROR(-20999, SQLERRM);

注：自定义错误码需在-20000到-20999范围内。

3. 优化Python代码的资源管理

添加finally块确保数据库连接和游标被正确关闭，避免资源泄漏：

def connCrearEmpleado(data):
    conn = cx_Oracle.connect('HR/hr@localhost:1521/xepdb1')
    cursor = None
    try:
        cursor = conn.cursor()
        cursor.callproc('add_emp', data)
        print('\n¡Empleado registrado!\n')
    except cx_Oracle.DatabaseError as err:
        print('Database error: {}'.format(err))
        raise
    finally:
        if cursor:
            cursor.close()
        conn.close()

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