Building queries with Flask-SQLAlchemy

I guess it's probably about time I wrote about a bit of code on this little blog of mine.

SQLAlchemy is a 'Python SQL Toolkit and Object Relational Mapper'. As we know Flask is pretty barebones but there's plenty of extensions including one for Flask-SQLAlchemy. And it's pretty handy. When I first started playing around with this app I tried writing my own basic Model class that used the built in database functions for Flask. But it wasn't long before all roads were pointing to SQLAlchemy - especially if I ever wanted to upgrade from SQLite at some stage with a minimum of fuss.

It's powerful and allows you to create models easily and do just about everything you want straight out of the box.

The hard part though, it the rather limited documentation on the Flask-SQLAlchemy site. You're therefore made to cross reference with the standard SQLAlchemy site where the implementation can be just different enough to be quite confusing.

Given that, I found it quite difficult to find figure out how to build a site with dynamically generated queries based on URL strings - you know, the sort that are central to APIs.

So I thought I'd just post my code here and step through it to explain what I've done:

from flask import Blueprint, abort  
from yvih import db  
from yvih.models import Member  
from sqlalchemy.sql.expression import or_

....

@members_blueprint.route('/')
@members_blueprint.route('/<path:conditions>')
def members( conditions=None ):  
    # if there are parameters
    if conditions != None:

        # get query and turn it into a dictionary
        conditions = conditions.split('/')
        query = zip(conditions[0::2], conditions[1::2])

        # check that all fields are valid
        if not parameter_accepted(query):
            abort(400)

        # build query
        filters = []
        for conditions in query:
            if conditions[0] == 'id':
                filters.append(Member.id.in_(conditions[1].split(',')))
            else:
                for term in conditions[1].split(','):
                    filters.append(Member.__dict__[conditions[0]].ilike('%' + term + '%'))
        members = Member.query.filter( or_(*filters) )
...

I'm obviously using Blueprints here but more about that later.

The important bit is the query building. Firstly I take the 'path' that forms the query and split it into a list of tuple that form key-value pairs using the zip() function. It needs to be a tuple rather than a dictionary because it's possible that a the query could contain two of the same keys. The url could be /members/id/1/second_name/Smith/id/20. If it were a dictionary, the id 20 would override id 1.

Then there's a function that tests the parameters are correct and throws a 'Bad request 400' error if, for example, a non-number has been passed as an id or one of the keys isn't in the list of possible keys.

I then create an empty filter list and loop over the list of tuples I created from the url query. If it's an id that has been passed, I split it into a list (ids can be in csv format) using the in_() method. Otherwise I loop over the other values (which can also be comma separated).

This time I use the __dict__ method of Member using the 'key' in the tuple as they key for the dictionary. Then ilike which is essentially a LIKE SQL condition.

Finally these rules are all joined together with the or_ method and filters is passed in with the asterisk to denote that it's an argument list (rather than just one argument). You'll notice I had to import or_ from sqlalchemy.sql.expression which was quite a revelation for me. Not surprisingly, it means that all the filters are joined with an OR condition which is what we want in this case.

Anyway, hope that helps someone. If I could have done it better, please let me know in the comments. This query business has a long way to go in my app development but this is a good start.