SqlAlchemy/Python如何正确捕获psycopg2.errors.UndefinedTable异常

你无法直接捕获psycopg2.ProgrammingError的核心原因是SQLAlchemy会对底层数据库驱动抛出的原生异常做包装，你实际捕获到的外层异常是sqlalchemy.exc.ProgrammingError，原生的psycopg2异常存放在该异常对象的orig属性中，所以直接匹配psycopg2的异常类会匹配失败。

方案1：捕获SQLAlchemy包装异常，判断原生异常类型 #

先导入对应依赖，修改异常捕获逻辑，同时修复原代码中finally块可能出现的session未定义就调用close()的问题，示例代码如下：

import psycopg2.errors
from sqlalchemy.exc import OperationalError, ProgrammingError

try: 
    Session = sessionmaker(bind=engine)
    session = Session()
           
    newentry = MyTable( 
        some_data = "some data",
    )
    session.add(newentry)
    session.commit()
     
except OperationalError as oe:
    print("Not possible to connect to DB")
except ProgrammingError as dbex:
    # 从包装异常中取出原生psycopg2异常判断类型
    if isinstance(dbex.orig, psycopg2.errors.UndefinedTable):
        print("Table does not exist")
        Base.metadata.create_all(engine)
    else:
        print(dbex)
except Exception as ex:
    print(ex)
finally:
    # 加判断避免session未初始化成功时调用close报错
    if 'session' in locals() and session.is_active:
        session.close()

方案2：通过PostgreSQL错误码判断（无需额外导入psycopg2异常类） #

UndefinedTable对应的PostgreSQL固定SQLSTATE错误码为42P01，可以直接通过错误码匹配：

except ProgrammingError as dbex:
    if dbex.orig.pgcode == '42P01':
        print("Table does not exist")
        Base.metadata.create_all(engine)

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