3

SQLAlchemy+Pyramidを使用してデータベースを操作しています。ただし、DBに常に存在するとは限らないオプションのテーブルがいくつかあります。だから私はそれらを照会している間、私はそのようなケースをキャッチしようとしますNoSuchTableError

try:
    x = session.query(ABC.name.label('sig_name'),func.count('*').label('count_')).join(DEF).join(MNO).filter(MNO.relevance >= relevance_threshold).group_by(DEF.signature).order_by(desc('count_')).all()[:val]
except NoSuchTableError:
    x = [-1,]

しかし、このステートメントを実行すると、ProgrammingErrorが発生します

ProgrammingError: (ProgrammingError) (1146, "Table 'db.mno' doesn't exist")

SQLAlchemyがより具体的なNoSuchTableErrorではなくより一般的なProgrammingErrorを発生させるのはなぜですか？そして、これが実際に予想される動作である場合、テーブルが存在するかどうかに応じて、アプリが正しい情報を表示するようにするにはどうすればよいですか？

編集1

これは私のWebアプリケーションの一部であるため、DBのモデルはmodels.py（私のピラミッドWebアプリケーションの下に）あります。.iniファイルに、追加のテーブルが使用可能かどうかをユーザーに選択するように求める設定があります。しかし、ユーザーを信頼しないので、テーブルが存在するかどうかを（ビューで）自分で確認できるようにしたいと思います。論争のあるテーブルは（in models.py）のようなものです

class MNO(Base):
    __tablename__="mno"
    id=Column(Integer,primary_key=True,autoincrement=True)
    sid=Column(Integer)
    cid=Column(mysql.MSInteger(unsigned=True))
    affectability=Column(Integer)
    cvss_base=Column(Float)
    relevance=Column(Float)
    __table_args__=(ForeignKeyConstraint(['sid','cid',],['def.sid','def.cid',]),UniqueConstraint('sid','cid'),)

テーブルが存在するかどうかを示す変数を設定できるように（できればアプリのセットアップ中に）、どのようにどこでチェックを行う必要がありますか？

注：この場合、「許しを求める」ではなく、次の場合に試してみる必要があります。

4

1 に答える 1

6

sqlalchemyのドキュメントによると、aNoSuchTableErrorは、「SQLAlchemyがデータベースからテーブルの定義をロードするように要求されたが、テーブルが存在しない」場合にのみスローされます。テーブルの定義をロードし、そこでエラーをキャッチして、それ以外の方法でクエリを実行してみてください。

「許しを求める」ことで物事をやりたい場合：

try:
    table = Table(table_name, MetaData(engine)) 
except NoSuchTableError:
    pass

または、テーブルが存在するかどうかを確認することもできます。

編集：

さらに良いことに、 has_tableメソッドを使用してみませんか：

if engine.dialect.has_table(connection, table_name):
    #do your crazy query

最初にInspectorを使用してテーブル名を取得してみませんか？

多分このようなもの：

from sqlalchemy import create_engine
from sqlalchemy.engine import reflection
#whatever code you already have
engine = create_engine('...')
insp = reflection.Inspector.from_engine(engine)
table_name = 'foo'
table_names = insp.get_table_names()
if table_name in table_names:
    x = session.query(ABC.name.label('sig_name'),func.count('*').label('count_')).join(DEF).join(MNO).filter(MNO.relevance >= relevance_threshold).group_by(DEF.signature).order_by(desc('count_')).all()[:val]
于 2012-05-01T10:21:29.537 に答える