SQLAlchemy - 嘗試除了不同的完整性錯誤

Question

對於唯一約束和外鍵違規違規，SQLAlchemy引發了IntegrityError。我怎樣才能把這個包裝在try/except區塊中並區分異常的各種原因？我想知道違規發生在哪一列，以及它是否是唯一或Foriegn密鑰違規。

以此課程爲例。

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

可能導致以下兩種不同的消息中的一個：

1. 如果name不是唯一的，所述消息是（x.IntegrityError）柱名稱不是唯一
2. 如果org_id是不是唯一的，消息是（x.IntegrityError）外鍵約束失敗

我想我可以解析出「唯一」和在第第一種情況。在第二種情況下，我可以解析出「外鍵」，但我無法獲得列名。如果表上有多個外鍵，那麼如何判斷違規發生的位置，而不必對每個父表進行另一次調用？

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