使用Flask SQLAlchemy,我正在使用一个名为bar
的表查询MySQL数据库,并查找与由foo
和country_code
:组成的筛选器匹配的行
foo_filter = 'hello'
country_code_filter = 'ES'
result = Bar.filter_by(foo=foo_filter, country_code=country_code_filter).first()
上述代码将返回包含foo = foo_filter
和country_code = country_code_filter
的第一行。
然而,我们可能没有与foo
匹配的某些国家/地区代码的行。在这些情况下(即,上面的查询返回0个结果的情况(,我想使用"RoW"的默认国家/地区过滤器,因为我们的数据集应该始终为foo
的每个可能值都有一个RoW值。如果意外发生这种情况,也没有返回任何结果,那么应该抛出一个错误。这是我的代码:
foo_filter = 'hello'
country_code_filter = 'ES'
result = Bar.filter_by(foo=foo_filter, country_code=country_code_filter).first()
if not result:
result = Bar.filter_by(foo=foo_filter, country_code='RoW').first()
if not result:
raise RuntimeException(f"No data for combination {foo_filter}, {country_code_filter} or {foo_filter}, RoW")
这种多次运行类似查询并每次检查结果直到我得到一行的方法感觉非常混乱/错误,但我找不到任何更好的方法,当您的初始查询在Flask SQLAlchemy 中返回0行时,可以设置"替代"过滤器
有没有更干净的方法?
此解决方案使用常规的SQLAlchemy,但在Flask SQLAlchemey上运行它只需要几处语法更改。
其想法是,我们同时查询所需的国家/地区代码和后备代码,确保结果始终首先按所需的国/地区代码排序,然后将结果限制为1。例如:
result = (
s.query(Bar)
.filter(
Bar.foo == foo_filter,
Bar.country_code.in_([country_code_filter, "RoW"]),
)
.order_by(func.FIELD(Bar.country_code, country_code_filter, "RoW"))
.limit(1)
.first()
)
MySQL中的FIELD
函数允许您指定自定义排序顺序,在这种情况下,它确保如果存在具有您想要的国家/地区代码的结果,它将始终首先返回。你可以在这里阅读更多关于FIELD
的信息。
以下是完整的测试代码:
from sqlalchemy_app import Base, Session, engine # you need to create these
from sqlalchemy import Column, Integer, String, func
class Bar(Base):
id = Column(Integer, primary_key=True)
foo = Column(String(32))
country_code = Column(String(32))
if __name__ == "__main__":
Base.metadata.create_all(engine)
s = Session()
s.add_all(
[
Bar(foo="hello", country_code="ES"),
Bar(foo="hello", country_code="ZIM"),
Bar(foo="hello", country_code="RoW"),
]
)
s.commit()
for foo_filter, country_code_filter, exp_res in (
("hello", "ES", "ES"),
("hello", "ZIM", "ZIM"),
("hello", "AUS", "RoW"),
("goodbye", "GBR", None),
):
result = (
s.query(Bar)
.filter(
Bar.foo == foo_filter,
Bar.country_code.in_([country_code_filter, "RoW"]),
)
.order_by(func.FIELD(Bar.country_code, country_code_filter, "RoW"))
.limit(1)
.first()
)
assert getattr(result, "country_code", None) == exp_res
正如您所说,在查询返回None
的情况下,您希望引发异常,您可以将.first()
访问方法交换为.one()
,因为查询将结果集限制为1个结果服务器端,您永远不会因为返回超过1个结果而遇到异常,并且当没有行与查询匹配时,您将引发NoResultFound
异常。