SQLAlchemy核心构建动态查询

Question

SQLAlchemy核心构建动态查询

4

我需要使用flask框架基于请求表单输入构建一个搜索表单。查询应该是根据参数的各种输入值动态构建的，只使用sqlalchemy CORE。有些值可以为空。同时还需要curl命令。我了解到可以使用基于过滤器的sqlalchemy orm方法。我应该如何处理？我尝试了以下内容：表格：

docs = Table("docs", metadata,   
                      Column("id", Integer, primary_key=True),
                      Column("file_name", String),
                      Column("custom_tag", String, nullable=True),
                      Column("description", String, nullable=True),
                      Column("up_date", DateTime(timezone=True),
                             default=datetime.datetime.utcnow),
                      Column("category_id", Integer, 
                             ForeignKey("categories.id"), nullable=True),
                      Column("sub_category_id", Integer,
                             ForeignKey("sub_categories.id"), nullable=True),
                      Column("alert_date", DateTime(timezone=True)))

curl:

curl -b cookies.txt -X POST http://localhost:5000/api/v1.0/docs -d "category_id=1&sub_category_id=None&alert_date=None"

这里的任何参数值都是可选的。

视图：

@app.route('/api/v1.0/docs', methods=['POST'])
def search_docs():
        if request.method == 'POST':
            doc_list = []
            category_id = request.form['category_id']
            sub_category_id = request.form['sub_category_id']
            custom_tag = request.form['custom_tag']
            alert_date = request.form['alert_date']

            f_dict = {
                'category_id': category_id,
                'sub_category_id': sub_category_id, 
                'alert_date': alert_date
                }
            for f in f_dict:
                if f_dict[f] == "None":
                    f_dict[f] = None

            s = select([docs]).\
                where(
                and_(
                     docs.c.category_id == category_id,
                     docs.c.sub_category_id == sub_category_id,
                     docs.c.alert_date == alert_date
                     )
                )
            print str(s)
            rs = conn.execute(s)
            for r in rs:
                doc_list.append(dict(r))

            return jsonify({'doc_list': doc_list}), 200

我删除了一些属性，因为我只是在尝试。请告诉我如何生成查询。我有许多属性值要考虑。使用flask 0.10、sqlalchemy core和python 2.7。

执行上述操作后，数据没有始终存储在表中。有时它会存储记录，但很多时候会抛出http 400状态码。如何将sub_category_id与可选的Null值进行比较？

- user956424

你能详细说明一下你所说的“如何生成查询”的意思吗？从你的例子中，你已经在创建并执行SELECT查询了。你是指要查看原始的SQL语句吗？ - Warren Spencer

@WarrenSpencer 我已经编辑了对上述查询的注释。我只想让 SQL 每次执行而不会出现 HTTP 状态码 400。 - user956424

@WarrenSpencer，通过在curl命令中添加与传递的字典中完全相同数量的参数，纠正了400状态。 - user956424

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- quornian · Accepted Answer

听起来您想根据任何过滤器给定的值是否为 None，有条件地构建where子句。类似下面这样的东西是否适合您？：

conditions = []
if category_id is not None:
    conditions.append(docs.c.category_id == category_id)
if sub_category_id is not None:
    conditions.append(docs.c.sub_category_id == sub_category_id)
if alert_date is not None:
    conditions.append(docs.c.alert_date == alert_date)

s = select([docs]).where(and_(*conditions))

为了确保您在排除筛选器时拥有None值，您可以从curl命令行中将它们排除掉：

-d "category_id=1&sub_category_id=None&alert_date=None"

成为：

-d "category_id=1"

在Python脚本中，您可以使用默认返回值为None的get：

category_id = request.form.get('category_id')
sub_category_id = request.form.get('sub_category_id')
...

另一种选择是使用空字符串来表示已排除的过滤器，这可能更容易传递。