SQLAlchemy - “动态过滤器”答案

【问题标题】：SQLAlchemy - "Dynamic Filter"SQLAlchemy - “动态过滤器”
【发布时间】：2016-09-17 02:51:00
【问题描述】：

我刚刚开始使用 SQLAlchemy。我决定使用它是因为我在我的 sqlite 查询中使用了很多字符串表达式。

所以，这是我的问题。我有一张桌子，上面有很多设备，每个设备都有维护水平的日期。关键是用户可以选择他想在屏幕上看到的维护级别。所以，我应该为他选择的每个维护级别组合“调整”我的 SQLAlchemmy。

例如，在原始 SQLite 中。

SELECT * WHERE (设备 IN []) AND m_level1 = DATE AND m_level2 = DATE ....)

因此，每个 if 条件可以有多个组合，这取决于选中了哪些复选框。正如我所提到的，在原始 SQL 中，我使用了很多字符串来达到我的目标。但我想使用 SQLAlchemy 改进代码。

对不起，我现在没有代码！谢谢你们！

【问题讨论】：

欢迎来到 Stack Overflow！您能否发布足够的代码来帮助我们了解您的尝试，以便我们可以使用接近 Minimal, Complete, and Verifiable Example (MCVE) 的内容？
当然。请看我的下一个答案。它很有效，但我不确定这是否是正确的想法。关键是我希望条件是动态的，并且通过该代码我得到了它。

标签： python sqlite dynamic filter sqlalchemy

【解决方案1】：

我假设您正在使用 ORM。

在这种情况下，filter 函数返回一个查询对象。您可以通过执行以下操作有条件地构建查询

query = Session.query(schema.Object).filter_by(attribute=value)
if condition:
    query = query.filter_by(condition_attr=condition_val)
if another_condition:
    query = query.filter_by(another=another_val)

#then finally execute it

results = query.all()

【讨论】：

如何解决多个 OR 而不是 AND？

【解决方案2】：

函数filter(*criterion) 表示您可以使用元组作为它的参数，@Wolph 在这里有详细信息： SQLALchemy dynamic filter_by了解详情

【讨论】：

【解决方案3】：

如果我们谈到 SQLAlchemy 核心，还有另一种方式：

from sqlalchemy import and_


filters = [table.c.col1 == filter1, table.c.col2 > filter2]
query = table.select().where(and_(*filters))

如果您尝试根据传入的表单条件进行过滤：

form = request.form.to_dict()
filters = []
for col in form:
    sqlalchemybinaryexpression = (getattr(MODEL, col) == form[col])
    filters.append(sqlalchemybinaryexpression)
query = table.select().where(and_(*filters))

MODEL 是你的 SQLAlchemy 模型

【讨论】：

【解决方案4】：

这个问题的另一个解决方案，这个案例以更安全的方式提出，因为它验证模型中是否存在要过滤的字段。

将运算符添加到要过滤的值。并且不必在查询中添加新参数，我们可以在值之前添加一个运算符，例如 ?foo=>1, '?foo=,?foo=>=1, ?foo=?foo=!1,?foo=1，最后在它们之间会是这样的 `?foo=a, b'。

from sqlalchemy.orm import class_mapper
import re

# input parameters
filter_by = {
  "column1": "!1", # not equal to
  "column2": "1",   # equal to
  "column3": ">1",  # great to. etc...
}

def computed_operator(column, v):
  if re.match(r"^!", v):
    """__ne__"""
    val = re.sub(r"!", "", v)
    return column.__ne__(val)
  if re.match(r">(?!=)", v):
    """__gt__"""
    val = re.sub(r">(?!=)", "", v)
    return column.__gt__(val)
  if re.match(r"<(?!=)", v):
    """__lt__"""
    val = re.sub(r"<(?!=)", "", v)
    return column.__lt__(val)
  if re.match(r">=", v):
    """__ge__"""
    val = re.sub(r">=", "", v)
    return column.__ge__(val)
  if re.match(r"<=", v):
    """__le__"""
    val = re.sub(r"<=", "", v)
    return column.__le__(val)
  if re.match(r"(\w*),(\w*)", v):
    """between"""
    a, b = re.split(r",", v)
    return column.between(a, b)
  """ default __eq__ """
  return column.__eq__(v)

query = Table.query
filters = []
for k, v in filter_by.items():
  mapper = class_mapper(Table)
  if not hasattr(mapper.columns, k):
    continue
  filters.append(computed_operator(mapper.columns[k], "{}".format(v))
query = query.filter(*filters)
query.all()

【讨论】：