sqlalchemy core build dynamic query

Question

I need to build a search form based on the request form input using flask framework. The query should be dynamically built based on various input values of the parameters using sqlalchemy CORE ONLY. Some of the values can be null. Also need curl command for the same. I have come across sqlalchemy orm based approach using filter. How exactly should I approach? I tried the following: table:

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"

Here any of the parameter values are optional.

View:

@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

I have deleted some of the attributes as I was just trying out. Please revert as to how the query can be generated. I have many attribute values to consider in the query. Using flask 0.10, sqlalchemy core, python 2.7

On executing the above, the data is not consistently stored in the table. Sometimes it stores the record, but many times it throws up http 400 status code. How do I compare the sub_category_id to Null value which is optional?

Answer 1

It sounds like you want to conditionally build your where clause based on whether the value given for any filter is None. Does something like this work for you?:

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))

To ensure you have None values for the filters you wish to exclude, you could exclude them from the curl command line:

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

becomes:

-d "category_id=1"

Then in the Python script, you could use get whose default return value is None:

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

Another alternative is to use empty strings to indicate an excluded filter, which might be easier to pass around.