python - SQLAlchemy - Try Except for different Integrity Errors -
sqlalchemy raises integrityerror
both unique constraint , foreign key contraint violations. how can wrap in try/except
block , discriminate between various causes of exception? want know column violation occurred , whether or not unique or foriegn key violation.
take class example.
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')
the following result in 1 of 2 different message:
- if
name
not unique, message (x.integrityerror) column name not unique - if
org_id
not unique, message (x.integrityerror) foreign key constraint failed
i suppose parse out "unique" , column name in 1st case. in second case parse out "foreign key", couldn't column name. if table has multiple foreign keys on it, how can tell violation happened without making call each parent table?
user = user(name='matthew moisen', org_id=999) db.session.add(user) try: db.session.commit() except integrityerror ex: # gives either: # (x.integrityerror) column name not unique # (x.integrityerror) foreign key constraint failed logger.exception(ex.message)
you're absolutely right, need parse out errors , it's database-specific. here example can use mysql:
from sqlalchemy import uniqueconstraint 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 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 little prettie, example : uk_columns = [i._pending_colargs in self.session.identity_map.values()[0].__table_args__ if isinstance(i, (uniqueconstraint, ))] print("uk vialation 1 of uk columns: {0}".format(uk_columns)) self.session.rollback() elif 'foreign key constraint fails' in e.orig[1]: # make little prettie, example : fks = [i in self.session.identity_map.values()[0].__table__.foreign_keys] print("fk vialation 1 of 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 1 of uk columns: [['name']] fk vialation 1 of fk columns: [foreignkey('company.company_id')] ok ---------------------------------------------------------------------- ran 1 test in 0.370s
Comments
Post a Comment