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

2fjabf4q 于 2023-01-14 发布在 Python

关注(0)|答案(2)|浏览(383)

下面的代码：

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似乎不起作用，因为在第三次捕获时打印了异常，并且没有创建表。我也尝试了psycopg2.Error，结果相同。
谢谢你的帮助

python-3.x

来源：https://stackoverflow.com/questions/70002377/sqlalchemy-python-properly-catch-psycopg2-errors-undefinedtable

2条答案

按热度按时间

mm9b1k5b1#

如果您通过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}')

赞(0）回复(0）举报 2023-01-14

dgiusagp2#

终于找到了一种方法来做到这一点。诀窍是首先捕获初始错误，然后重新引发以捕获原始错误。

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))

赞(0）回复(0）举报 2023-01-14

我来回答

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

2条答案

相关问题

热门标签

最新问答