SqlAlchemy/Python正确捕获psycopg2.errors.UndefinedTable

Question

我有以下代码：

try: 
Session = sessionmaker(bind=engine)
session = Session()

newentry = MyTable( 
some_data = "some data",
)
session.add(newentry)
session.commit()

except psycopg2.OperationalError as oe:
print("Not possible to connect to DB")
except psycopg2.ProgrammingError as dbex:
print("Table does not exist")
Base.metadata.create_all(engine)
except Exception as ex:
print(ex)
finally:    
session.close()

如何正确捕获psycopg2.errors.UndefinedTable？

psycopg2.ProgrammingError似乎不起作用，因为异常是在第三个catch打印的，并且没有创建表。我也尝试了psycopg2.Error，得到了同样的结果。

感谢的帮助

Answer 1

终于找到了一种方法。诀窍是首先捕获初始错误，然后重新提升以捕获原始错误。

import pandas as pd
import sqlalchemy as sa
from psycopg2 import errors, errorcodes
url = "postgresql+psycopg2://user:password@host/database"
def get_version(url: str):

try:
return pd.read_sql("select version_num from alembic_version;", sa.create_engine(url)).loc[0, "version_num"]

except sa.exc.ProgrammingError as e:

try:
raise e.orig

except errors.lookup(errorcodes.UNDEFINED_TABLE):
return None

print(get_version(url))

Answer 2

如果您通过sqlalchemy构建会话，以下将捕获UndefinedTable错误：

from sqlalchemy.exc import ProgrammingError
# import other dependencies and create your session as before
try: 
Session = sessionmaker(bind=engine)
session = Session()

newentry = MyTable( 
some_data = "some data",
)
session.add(newentry)
session.commit()
except ProgrammingError as pe:
print(f'Exception Caught: {pe}')