SQLAlchemy - さまざまな整合性エラーを除いて試してください

Question

SQLAlchemyは、ユニーク制約違反と外部キー制約違反の両方に対してIntegrityErrorを発生させます。 try/exceptブロックでこれを包み込み、例外のさまざまな原因を区別するにはどうすればよいですか？私は違反が発生した列と、それがUniqueキーかForiegn Key違反かどうかを知りたい。

たとえば、このクラスを使用します。

class User(db.Model): 
    id = db.Column(db.Integer, primary_key=True) 
    name = db.Column(db.String, unique=True) 
    org_id = db.column(db.Integer, db.ForeignKey('company_id') 

以下は、2つの異なるメッセージの1つをもたらし得る：

1. nameが一意でない場合、メッセージは、（x.IntegrityError）でorg_id場合列名は、
2. 一意ではありませんがユニークでない場合、メッセージは（x.IntegrityError）外部キー制約が失敗した

私は "ユニーク" e第1の場合。後者の場合、「外部キー」を解析できますが、列名を取得できませんでした。テーブルに複数の外部キーがある場合、各親テーブルへの別の呼び出しを行わずにどのような違反が発生したかをどのように伝えることができますか？

user = User(name='Matthew Moisen', org_id=999) 
db.session.add(user) 
try: 
    db.session.commit() 
except IntegrityError as ex: 
    # This gives either: 
    # (x.IntegrityError) column name is not unique 
    # (x.IntegrityError) foreign key constraint failed 
    logger.exception(ex.message) 

Answer 1

あなたは本当に正しいです、あなたはエラーを解析する必要があり、それはデータベース特有です。 MySQLで使用できる例を次に示します。

from sqlalchemy import UniqueConstraint 
from sqlalchemy.exc import IntegrityError 

class MyUser(Base): 
    __tablename__ = 'MyUser' 
    id = Column(Integer, primary_key=True) 
    name = Column(String(64)) 
    org_id = Column(Integer, ForeignKey('Company.company_id')) 
    # notice how we define UK here : 
    __table_args__ = (
    UniqueConstraint('name', name='MyUser_uk_1'), 
) 

class UserTest(unittest2.TestCase): 

    def commit_try_cath(self): 
    try: 
     self.session.commit() 
    except IntegrityError, e: 
     if e.orig[1].startswith('Duplicate entry'): 
     # make it a little prettie, this is just an example : 
     uk_columns = [i._pending_colargs for i in self.session.identity_map.values()[0].__table_args__ if isinstance(i, (UniqueConstraint,))] 
     print("UK vialation for one of the UK columns: {0}".format(uk_columns)) 
     self.session.rollback() 
     elif 'foreign key constraint fails' in e.orig[1]: 
     # make it a little prettie, this is just an example : 
     fks = [i for i in self.session.identity_map.values()[0].__table__.foreign_keys] 
     print("FK vialation for one of the FK columns: {0}".format(fks)) 
     self.session.rollback() 
    except Exception, e: 
     print("Somethig else") 

    def test_try_catch(self): 
    # first user should go through : 
    user_01 = MyUser(name='unittest', org_id=1) 
    self.session.add(user_01) 
    self.commit_try_cath() 
    # UK vialation, dup name : 
    user_02 = MyUser(name='unittest', org_id=1) 
    self.session.add(user_02) 
    self.commit_try_cath() 
    # FK vialation, org_id 0 doesn't exist : 
    user_03 = MyUser(name='unittest_new', org_id=0) 
    self.session.add(user_03) 
    self.commit_try_cath() 

$ nosetests -v -s user_test.py:UserTest.test_try_catch 
test_dev_debug (user_test.UserTest) ... 
UK vialation for one of the UK columns: [['name']] 
FK vialation for one of the FK columns: [ForeignKey('Company.company_id')] 
ok 

---------------------------------------------------------------------- 
Ran 1 test in 0.370s