如何使用SQLAlchemy找到导致IntegrityError的属性

Question

我有一个非常简单的SqlAlchemy模型

class User(Base):
    """ The SQLAlchemy declarative model class for a User object. """
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    phone = Column(String, unique=True)
    email = Column(String, unique=True)

当插入一个新的用户时，如果电子邮件或电话是重复的，则可能会出现 IntegrityError。

有没有办法检测是哪个列违反了完整性错误？或者唯一的方法是执行单独的查询来查看某个值是否存在？

- Peter Smit

---

你可以将每个操作都放在 try/except 子句中，捕获 IntegrityError 并作出相应的反应... 但是，两个人分享一个手机不也是可能的吗？（我和我的妻子就这样...） - mgilson

@mgilson 是的，我明白了，但我想知道哪一列是违规的；电子邮件还是电话。对于这个业务案例，需要确保电话号码是唯一的（通过电话登录）。 - Peter Smit

我认为你无法在不解析后端（或rdbms驱动程序）提供的异常消息的情况下完成此操作。 - van

Answer 1

你可以使用以下方法获取基础代码、消息，并相应地格式化消息。

except exc.IntegrityError as e:
       errorInfo = e.orig.args
       print(errorInfo[0])  #This will give you error code
       print(errorInfo[1])  #This will give you error message

顺便提一下，您需要从SQLAlchemy中导入exc：from sqlalchemy import exc 如果您需要其他信息，请告诉我。 我可以试试。

有关SQLAlchemy exc的更多信息，请查找代码：https://github.com/zzzeek/sqlalchemy/blob/master/lib/sqlalchemy/exc.py

- SANDEEP MACHIRAJU

---

从 e.orig.args 获取的信息取决于您正在运行的数据库引擎。例如，sqlite 的结果是一个大小为 1 的元组，并且没有错误代码。 - Jacob Pavlock

Answer 2

try:
    ....
except IntegrityError as e:
    print(e.orig.diag.message_detail)

这对我有帮助。

- cacheoff

---

正是我所需要的。谢谢！ - Jim Bray

Answer 3

很遗憾，没有一种干净的方法来完成这个任务，但我使用IntegrityError上的orig属性和parse模块来解决：parse。

try:
    db.session.add(user)
    db.session.commit()
except IntegrityError, e:
    dupe_field = parse('duplicate key value violates unique constraint "{constraint}"\nDETAIL:  Key ({field})=({input}) already exists.\n', str(e.orig))["field"]

这可能不是IntegrityError抛出的唯一错误字符串，未来SQLAlchemy的更新可能会更改它，因此这并不理想。

- pixelperfect

---

这是哪个“parse”模块？ - Asav Patel

@AsavPatel 给你链接 https://pypi.org/project/parse/ - Michael

Answer 4

与 @pixelperfect 相同的解决方案，但使用标准库（re 模块）

def get_conflicting_field(err: sqlalchemy.exc.IntegrityError) -> tuple[str, str] | None:
    """
    Parses the IntegrityError message and returns tuple with conflicting field name and value.
    """
    pattern = re.compile(r'DETAIL\:\s+Key \((?P<field>.+?)\)=\((?P<value>.+?)\) already exists')
    match = pattern.search(str(err))
    if match is not None:
        return match['field'], match['value']

- Dany

Answer 5

我通常会使用 try catch 来处理这个问题。

try:
    session.commit()
catch:   
str(sys.exc_info()[0]) + " \nDESCRIPTION:  "+ str(sys.exc_info()[1]) + "\n" + str(sys.exc_info()[2])

当我遇到完整性错误时，我会收到以下消息，并跳过该特定事务并继续进行其余的事务。

DESCRIPTION:  (IntegrityError) duplicate key value violates unique constraint "test_code"
DETAIL:  Key (test_code)=(5342) already exists.
'INSERT INTO test_table (pk, test_code, test_name) VALUES (%(pk)s, %(test_code)s, %(test_name)s)' { 'pk': '1', 'test_code': '5342', 'test_name': 'test' }

- Jacob George