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:

  1. if name not unique, message (x.integrityerror) column name not unique
  2. 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

Popular posts from this blog

ios - RestKit 0.20 — CoreData: error: Failed to call designated initializer on NSManagedObject class (again) -

java - Digest auth with Spring Security using javaconfig -

laravel - PDOException in Connector.php line 55: SQLSTATE[HY000] [1045] Access denied for user 'root'@'localhost' (using password: YES) -