sqlalchemy 核心构建动态查询

Question

我需要使用flask框架根据请求表单输入构建一个搜索表单。应该使用 sqlalchemy CORE ONLY 根据参数的各种输入值动态构建查询。某些值可以为空。同样需要 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 -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 核心、python 2.7

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

Answer 1

听起来您想根据为任何过滤器提供的值是否为 有条件地构建您的 where 子句None。像这样的事情对你有用吗？：

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 脚本中，您可以使用get其默认返回值为None：

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

另一种替代方法是使用空字符串来指示排除的过滤器，这可能更容易传递。