python - 使 SQLAlchemy 错误更加用户友好和详细
问题描述
我有一个这样的模型:
class Company(db.Model):
__tablename__ = "my_table"
id = db.Column(db.Integer(), primary_key=True)
name = db.Column(db.String(128), unique=True, nullable=False)
slug = db.Column(db.String(128), unique=True, nullable=False)
如您所见,我正在使用Flask-SQLAlchemy
的类和方法,但这不是重点,假设我有一个执行以下行的视图:
c = Company("Test", "test")
try:
db.session.add(c)
db.session.commit()
return "Added!"
except Exception as e:
db.session.rollback()
return f"{e}"
上面的代码,创建一个 的对象Company
,尝试将其添加到数据库中,在异常时回滚事务,
问题就在这里,因为数据是硬编码的,它应该总是返回异常,SQLAlchemy 会引发一个IntegrityError
.
这IntegrityError
对用户来说非常丑陋和无用,例如:
(sqlite3.IntegrityError) UNIQUE constraint failed: my_table.name [SQL: 'INSERT INTO my_table (name, slug) VALUES (?, ?)'] [parameters: ('Test', 'tests')] (Background on this error at: http://sqlalche.me/e/gkpj)
我正在寻找一种使其美化和用户友好的方法,在此之前我使用db.validates
装饰器并检查验证时的重复数据,但这对我来说感觉不对
我最不需要的是找出导致问题的字段而无需硬编码
解决方案
SQLAlchemy 包含一种允许使用 handle_error 事件挂钩自定义 DBAPI 错误的机制。我在 Openstack oslo.db中使用了这个 API,可以在这个文件中看到:https ://github.com/openstack/oslo.db/blob/master/oslo_db/sqlalchemy/exc_filters.py 。
由于 stackoverflow 讨厌链接到代码,这里有一个基于上述链接方法的 POC:
import collections
from sqlalchemy import event
from sqlalchemy import exc as sqla_exc
import re
class DuplicateKeyError(Exception):
"""Duplicate entry at unique column error."""
def __init__(self, columns=None, inner_exception=None, value=None):
self.columns = columns or []
self.value = value
self.inner_exception = inner_exception
def __str__(self):
return "Duplicate key for columns %s" % (
self.columns,
)
_registry = collections.defaultdict(lambda: collections.defaultdict(list))
def filters(ame, exception_type, regex):
"""Mark a function as receiving a filtered exception."""
def _receive(fn):
_registry[ame][exception_type].extend(
(fn, re.compile(reg))
for reg in ((regex,) if not isinstance(regex, tuple) else regex)
)
return fn
return _receive
# each @filters() lists a database name, a SQLAlchemy exception to catch,
# and a list of regular expressions that will be matched. If all the
# conditions match, the handler is called which then raises a nicer
# error message.
@filters(
"sqlite",
sqla_exc.IntegrityError,
(
r"^.*columns?(?P<columns>[^)]+)(is|are)\s+not\s+unique$",
r"^.*UNIQUE\s+constraint\s+failed:\s+(?P<columns>.+)$",
r"^.*PRIMARY\s+KEY\s+must\s+be\s+unique.*$",
),
)
def _sqlite_dupe_key_error(integrity_error, match, engine_name, is_disconnect):
columns = []
try:
columns = match.group("columns")
columns = [c.split(".")[-1] for c in columns.strip().split(", ")]
except IndexError:
pass
raise DuplicateKeyError(columns, integrity_error)
@filters(
"mysql",
sqla_exc.IntegrityError,
r"^.*\b1062\b.*Duplicate entry '(?P<value>.*)'"
r" for key '(?P<columns>[^']+)'.*$",
)
@filters(
"postgresql",
sqla_exc.IntegrityError,
(
r'^.*duplicate\s+key.*"(?P<columns>[^"]+)"\s*\n.*'
r"Key\s+\((?P<key>.*)\)=\((?P<value>.*)\)\s+already\s+exists.*$",
r"^.*duplicate\s+key.*\"(?P<columns>[^\"]+)\"\s*\n.*$",
),
)
def _default_dupe_key_error(
integrity_error, match, engine_name, is_disconnect
):
columns = match.group("columns")
uniqbase = "uniq_"
if not columns.startswith(uniqbase):
if engine_name == "postgresql":
columns = [columns[columns.index("_") + 1 : columns.rindex("_")]]
else:
columns = [columns]
else:
columns = columns[len(uniqbase) :].split("0")[1:]
value = match.groupdict().get("value")
raise DuplicateKeyError(columns, integrity_error, value)
def handler(context):
"""Iterate through available filters and invoke those which match.
The first one which raises wins.
"""
def _dialect_registries(engine):
if engine.dialect.name in _registry:
yield _registry[engine.dialect.name]
if "*" in _registry:
yield _registry["*"]
for per_dialect in _dialect_registries(context.engine):
for exc in (context.sqlalchemy_exception, context.original_exception):
for super_ in exc.__class__.__mro__:
if super_ in per_dialect:
regexp_reg = per_dialect[super_]
for fn, regexp in regexp_reg:
match = regexp.match(exc.args[0])
if match:
fn(
exc,
match,
context.engine.dialect.name,
context.is_disconnect,
)
if __name__ == '__main__':
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import Session
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Company(Base):
__tablename__ = "my_table"
id = Column(Integer(), primary_key=True)
name = Column(String(128), unique=True, nullable=False)
slug = Column(String(128), unique=True, nullable=False)
def __init__(self, name, slug):
self.name = name
self.slug = slug
e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
event.listen(e, "handle_error", handler)
s = Session(e)
s.add(Company("Test", "test"))
s.commit()
s.add(Company("Test", "test"))
s.commit()
运行它,我们看到:
2019-03-13 09:44:51,701 INFO sqlalchemy.engine.base.Engine INSERT INTO my_table (name, slug) VALUES (?, ?)
2019-03-13 09:44:51,701 INFO sqlalchemy.engine.base.Engine ('Test', 'test')
2019-03-13 09:44:53,387 INFO sqlalchemy.engine.base.Engine ROLLBACK
Traceback (most recent call last):
# ...
sqlite3.IntegrityError: UNIQUE constraint failed: my_table.slug
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
# ...
__main__.DuplicateKeyError: Duplicate key for columns ['slug']
推荐阅读
- c# - 允许用户通过 Azure AD 用户名和密码登录
- csv - 来自 excel 的 PIMCORE 填充地图
- php - 使用 bindparam 插入多行
- regex - 单词或行尾之前的 Postgresql 正则表达式模式匹配
- javascript - 有什么方法可以用 JavaScript 更改多个标签?
- python - 导入用户定义的包时出错
- c# - 如果按下两个 WASD 键,角色会朝第一次按键的方向移动 - 但一旦抬起,角色就会继续朝那个方向移动
- c - 为什么我的输出字符串的第 17 个字符被替换为“o”?
- bitbucket - Bitbucket 管道上的远程命令
- c++ - 为什么我的代码超过了时间限制,而真正相似的代码却没有(Leetcode 1249)?