Building relatively complex dynamic queries in SQLAlchemy

[Nikola Jankovic]

I have an API endpoint that handles searches from the frontend. A search can have a dynamic amount of filters applied to it, including (1) sizes, (2) colors, (3) price, and (4) category that are passed through query parameters. Sizes and colors are passed as comma separated values (i.e. colors=Red,Blue,Orange) and are converted to lists on the backend. Price and Category are single values (i.e. price=20-300&category=Shirt). 

I'm having trouble constructing a query based on these filters. If multiple values could not be passed for any parameters I wouldn't have much of an issue - I could construct a dictionary based on the parameter name and unpack it in a .filter() call. That's not the case though - I need to implement an 'or' operator to accomodate the multiple possible colors and sizes. 

Additionally, each of these filters is optional, meaning some queries will comprise only some of them, which again complicates the query. 

And if that wasn't complicated enough - properties like sizes and colors are represented as a one to many relationship not simple categories. I have hybrid properties that return a list of possible sizes but using those in queries containing filter_by() raises issues. 

Does anyone know how I could even begin to structure a dynamic query like this?

[Simon King]

Just build it up one step at a time, something like this:

q = session.query(YourObject)

if price is not None:
q = q.filter(YourObject.price == price)

if category is not None:
q = q.filter(YourObject.category == category)

if sizes:
q = q.join(Size).filter(Size.size.in_(sizes))

if colors:
q = q.join(Color).filter(Color.name.in_(colors))

Hope that helps,

Simon

[Jonathan Vanasco]

I have a few of these and generally use a variant of Simon's method where I wrap everything in a custom class.  I stash all the requirements/options in the custom class instance, then invoke a `count()` or `paginated()` method that builds/stashes the core query and runs `count()` or `all()` on it.

        class CustomQuery():
                _core_query = None

                @property
                def core_query(self):
                        if self._core_query is None:
                                # simon's code
                                self._core_query = XXX
                        return self._core_query

                def count(self):
                        return self.core_query.count()

                def paginated(self, offset=0, limit=None):
                        return self.core_query.offset(offset).limit(limit).all()