SQLAlchemy - 尝试除了不同的完整性错误答案

【问题标题】：SQLAlchemy - Try Except for different Integrity ErrorsSQLAlchemy - 尝试除了不同的完整性错误
【发布时间】：2016-09-26 23:24:59
【问题描述】：

SQLAlchemy 为唯一约束和外键约束违规引发IntegrityError。如何将其包装在 try/except 块中并区分导致异常的各种原因？我想知道违规发生在哪一列，是唯一违规还是外键违规。

以这门课为例。

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

以下可能导致两种不同消息之一：

如果name 不是唯一的，则消息是 (x.IntegrityError) 列名不是唯一的
如果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)

【问题讨论】：

这取决于后端。如果您检查ex.orig，您可以查看原始 DBAPI 错误，该错误可能包含受影响的列，具体取决于后端。

标签： python sqlalchemy

【解决方案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

【讨论】：