如何区分SQLAlchemy IntegrityError的原因？
问题描述 投票：0回答：5
当事务存在数据完整性问题时，
sqlalchemy
SQLAlchemy 似乎只会抛出一般性的
IntegrityErrorsession.commit()有没有一种方法，除了尝试以编程方式解析错误消息和/或查询之外，我可以将重复的主键错误与外键错误或失败的 CHECK 约束等区分开来？或者甚至只是一种判断哪个表的哪一列违反了数据完整性的方法？或者只是一种在导致错误的行上立即引发异常而不是等待事务提交的方法？
5个回答
5
投票
投票
IntegrityErrororigstatement鉴于此模型：
class Foo(Base):
__tablename__ = 'foo20201209'
id = sa.Column(sa.Integer, primary_key=True)
bar = sa.Column(sa.String(2), unique=True)
baz = sa.Column(sa.Integer, sa.CheckConstraint('baz >= 0'), default=0)
此代码：
conn_strings = ['postgresql+psycopg2:///test',
'mysql+mysqlconnector:///test',
'sqlite://']
for cs in conn_strings:
engine = sa.create_engine(cs)
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
session = orm.Session(bind=engine)
for kwds in [{'bar': 'a'}, {'bar': 'a'}, {'bar': 'b', 'baz': -11}]:
session.add(Foo(**kwds))
try:
session.commit()
except sa.exc.IntegrityError as ex:
print(ex.orig)
print(ex.statement)
print()
session.rollback()
session.close()
engine.dispose()
将产生以下输出：
duplicate key value violates unique constraint "foo20201209_bar_key"
DETAIL: Key (bar)=(a) already exists.
INSERT INTO foo20201209 (bar, baz) VALUES (%(bar)s, %(baz)s) RETURNING foo20201209.id
new row for relation "foo20201209" violates check constraint "foo20201209_baz_check"
DETAIL: Failing row contains (3, b, -11).
INSERT INTO foo20201209 (bar, baz) VALUES (%(bar)s, %(baz)s) RETURNING foo20201209.id
1062 (23000): Duplicate entry 'a' for key 'bar'
INSERT INTO foo20201209 (bar, baz) VALUES (%(bar)s, %(baz)s)
4025 (23000): CONSTRAINT `foo20201209.baz` failed for `test`.`foo20201209`
INSERT INTO foo20201209 (bar, baz) VALUES (%(bar)s, %(baz)s)
UNIQUE constraint failed: foo20201209.bar
INSERT INTO foo20201209 (bar, baz) VALUES (?, ?)
CHECK constraint failed: foo20201209
INSERT INTO foo20201209 (bar, baz) VALUES (?, ?)
4
投票
投票
正如我原来的答案下面的DMJ评论，不同的数据库引擎在发生完整性错误时会发出不同的错误消息，并且SQLAlchemy不会尝试以一致的方式呈现这些消息，因此对问题的简短回答是：如果不解析错误消息，就无法从异常中获取有用的信息。
也就是说，如果约束以一致的方式命名，并且这些名称出现在错误消息中，则解析错误消息可能不会那么困难。这里 SQLAlchemy 提供了一些帮助：我们可以创建一个 Metadata 对象，并使用定义的约束的命名约定（docs）。有了这个约定，我们创建的任何约束都将遵循它，我们可以匹配错误消息中的约束名称，并使用匹配来查找元数据中的表和约束对象。
以下是如何使用约定解析错误消息的示例。我没有涵盖所有可能的约束类型，也没有处理 Sqlite 的唯一键违规消息，该消息省略了约束名称。这些留给读者作为练习;-)
import re
import sqlalchemy as sa
convention = {
"ix": 'ix_%(column_0_label)s',
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
"pk": "pk_%(table_name)s"
}
metadata = sa.MetaData(naming_convention=convention)
tbl = sa.Table(
't65189213',
metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('foo', sa.String(2), unique=True),
sa.Column('bar', sa.Integer, default=0),
sa.CheckConstraint('bar >= 0', name='positive_bar'),
)
# Basic pattern to match constraint names (not exhaustive).
pattern = r'(?P<type_key>uq|ck)_(?P<table>[a-z][a-z0-9]+)_[a-z_]+'
regex = re.compile(pattern)
def parse_exception(ex):
# Returns an informative message, or the original error message.
# We could return the table, constraint object etc. instead.
types = {'ck': 'Check constraint', 'uq': 'Unique key'}
m = regex.search(str(ex.orig))
if m:
type_ = types[m.groupdict()['type_key']]
table_name = m.groupdict()['table']
constraint_name = m.group(0)
table = metadata.tables[table_name]
constraint = next(c for c in table.constraints if c.name == constraint_name)
columns = ','.join(constraint.columns.keys())
return f'{type_} {constraint_name} has been violated on table {table_name!r} columns: {columns}'
return f'{ex.orig}'
conn_strings = ['postgresql+psycopg2:///test',
'mysql+mysqlconnector://root:root@localhost/test',
'sqlite://']
for cs in conn_strings:
engine = sa.create_engine(cs, future=True)
tbl.drop(engine, checkfirst=True)
tbl.create(engine)
with engine.connect() as conn:
print(engine.dialect.name)
for kwds in [{'foo': 'a'}, {'foo': 'a'}, {'foo': 'b', 'bar': -11}]:
try:
conn.execute(tbl.insert(), kwds)
conn.commit()
except sa.exc.IntegrityError as ex:
print(parse_exception(ex))
conn.rollback()
print('-'*10)
engine.dispose()
输出：
postgresql
Unique key uq_t65189213_foo has been violated on table 't65189213' columns: foo
Check constraint ck_t65189213_positive_bar has been violated on table 't65189213' columns:
----------
mysql
Unique key uq_t65189213_foo has been violated on table 't65189213' columns: foo
Check constraint ck_t65189213_positive_bar has been violated on table 't65189213' columns:
----------
sqlite
UNIQUE constraint failed: t65189213.foo
Check constraint ck_t65189213_positive_bar has been violated on table 't65189213' columns:
----------
1
投票
投票
我最终使用
session.flush()
1
投票
投票
我也遇到了同样的麻烦，例如，对于约束来说，看起来更好的解决方案是给它们命名，然后解析这些异常消息。无论如何，对于我的
SQLAlchemy
==1.4.28，
UNIQUE约束名称仍然没有出现在抛出的异常中。考虑下一个例子：
class M(Model):
a = Column(String)
i = Column(Integer)
UniqueConstraint(M.i, name="my unique")
CheckConstraint(0 <= M.i, name="my check")
def test_test():
db = create_test_database(Model)
try:
with db.session() as s:
# Here we break our UNIQUE constraint
try:
with s.begin_nested():
s.add_all(
[
M(a="Alice", i=1),
M(a="Bob", i=1),
]
)
except IntegrityError as err:
rich.inspect(err)
# Here we break our CHECK constraint
s.add(M(a="Alice", i=-1))
except IntegrityError as err:
rich.inspect(err)
结果如下：
┌────────────────────── <class 'sqlalchemy.exc.IntegrityError'> ───────────────────────┐
│ Wraps a DB-API IntegrityError. │
│ │
│ ┌──────────────────────────────────────────────────────────────────────────────────┐ │
│ │ IntegrityError('(sqlite3.IntegrityError) UNIQUE constraint failed: M.i') │ │
│ └──────────────────────────────────────────────────────────────────────────────────┘ │
│ │
│ args = ('(sqlite3.IntegrityError) UNIQUE constraint failed: M.i',) │
│ code = 'gkpj' │
│ connection_invalidated = False │
│ detail = [] │
│ hide_parameters = False │
│ ismulti = False │
│ orig = IntegrityError('UNIQUE constraint failed: M.i') │
│ params = ('Bob', 1) │
│ statement = 'INSERT INTO "M" (a, i) VALUES (?, ?)' │
└──────────────────────────────────────────────────────────────────────────────────────┘
┌──────────────────────── <class 'sqlalchemy.exc.IntegrityError'> ─────────────────────────┐
│ Wraps a DB-API IntegrityError. │
│ │
│ ┌──────────────────────────────────────────────────────────────────────────────────────┐ │
│ │ IntegrityError('(sqlite3.IntegrityError) CHECK constraint failed: my check') │ │
│ └──────────────────────────────────────────────────────────────────────────────────────┘ │
│ │
│ args = ('(sqlite3.IntegrityError) CHECK constraint failed: my check',) │
│ code = 'gkpj' │
│ connection_invalidated = False │
│ detail = [] │
│ hide_parameters = False │
│ ismulti = False │
│ orig = IntegrityError('CHECK constraint failed: my check') │
│ params = ('Alice', -1) │
│ statement = 'INSERT INTO "M" (a, i) VALUES (?, ?)' │
└──────────────────────────────────────────────────────────────────────────────────────────┘
因此，看起来这个 CHECK
约束名称将始终位于异常字符串中的某个位置，您可以围绕它编写代码。很明显，为什么这是
UNIQUE约束的额外信息，因为您已经有了单词
UNIQUE和字段名称（在我的示例中为
A.i）。我认为他们永远不会改变这些字符串格式，但询问他们会很有趣。
警告：
问题是我的代码与 SQLite
相关，而您的代码可能与另一个数据库相关，并且这些消息会有所不同，因为它们最初来自底层数据库引擎，而不是来自
SQLAlchemy本身。因此，您必须注意抽象这些字符串的代码。
0
投票
投票
也有同样的问题，在调试时我在完整性错误中发现了这个问题
e
：
本质上，这就是我想出的。实际上，我的代码中有更多检查以使其达到生产级。
注意：这是 Postgres 特有的！
from sqlalchemy.exc import IntegrityError, MultipleResultsFound, NoResultFound
from asyncpg import NotNullViolationError, UniqueViolationError
try:
...
except IntegrityError as e:
cause = e.orig.__getattribute__("__cause__")
# If cause is None, just default to constraint violation error
if cause is None:
raise ConstraintViolationError(e)
if isinstance(cause, NotNullViolationError):
# Can't type this well as asyncpg error attributes are set dynamically by a metaclass
raise ConstraintViolationError(cause.__getattribute__("message"))
elif isinstance(cause, UniqueViolationError):
column = cause.__getattribute__("column_name")
raise NotUniqueError(
message=cause.__getattribute__("message"), model_name=self.model.__name__, column_name=column
)
else:
column = cause.__getattribute__("column_name")
raise ConstraintViolationError(cause.__getattribute__("message"))
最新问题
- 在.ddev目录中找不到docker-compose.yaml文件
- 如何打开文件对话框？
- Hibernate 中的外键问题
- 如何将Spring Boot 3 httpRestClient错误response.body记录为可读字符串？
- Python 3 + OpenCV + Mediapipe。多目标跟踪
- 错误：“push”的指令后缀无效[重复]
- 如何检测窗口关闭？
- 如何创建具有多个参数和多个插入语句的存储过程？
- React Material UI 和 React hook 形成清晰按钮动态组件
- 我可以在没有 Mui 主题的情况下使用 Mui 组件吗
- 尝试使用 libgit2 的示例 lg2_checkout 拉取 git 存储库
- 如何从 Google 表格访问单元格值？
- databricks 上的 pyspark 代码永远不会完成执行并挂在中间
- QGIS 属性表记录（各个记录的）修改日期
- '号码 | undefined' 不可分配给类型'string'。打字稿
- 如何使用 SQL 查询从 json 获取第一个元素
- 无法导入模块“lambda_function”：无法从“lxml”aws lambda 导入名称“etree”
- 熊猫填充条件取决于另一列
- 使用 Delaunay 三角剖分在 R 中计算表面拉普拉斯/拉普拉斯贝尔特拉米
- 为什么 libgit2 的 core.longpath 位置限制为 260 个字符？
© www.soinside.com 2019 - 2024. All rights reserved.