SQLAlchemy - Try За исключением различных ошибок целостности

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) не выполнено.

Я полагаю, что я мог бы разобрать «уникальное» и имя столбца в th е 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