Query Helpers for SQLAlchemy Models
Making frequent database tasks easier with helpers for common SQL operations.
Series: How I Use Flask
6 min
Common database tasks involve a lot of verbose repetition
If there’s one thing I dislike when working on a Flask app, it’s the rather excessively verbose and repetitive code that is required to build up relatively simple—but exceedingly common—database queries and operations. Whether it’s getting an item by its primary key, filtering or defining a query based on certain model attributes, or even just saving items to the database, I find SQLAlchemy to be somewhat of an annoying intermediary for common operations.
To be clear, I’m still quite a fan of SQLAlchemy, and generally enjoy working with it when necessary—particularly when defining custom queries or raw SQL—I just think it could do with a bit of help being less annoying for common operations in any sizeable Flask application.
Remove repetition with custom base Model class
A custom base Model class is always the first place I start when building Flask apps. This base Model class—which all application models will subclass—usually defines a couple common columns, like primary key and timestamp fields I want all models to have.
# -*- coding: utf-8 -*-
from arrow import utcnow
from sqlalchemy.ext.declarative import declared_attr, has_inherited_table
from sqlalchemy_utils import ArrowType
from myapp.db import db
from .mixins.query import QueryMixin
__all__ = ('Model',)
class Model(db.Model, QueryMixin):
"""Abstract base class for all app models.
Provides a `pk` & `created_at` column to every model.
To define models, follow this example:
from .base import Model
class MyModel(Model):
# model definition
"""
__abstract__ = True
pk = db.Column(db.Integer, primary_key=True)
@declared_attr
def created_at(cls):
return db.Column(ArrowType, default=utcnow, nullable=False, index=True)
@property
def class_name(self):
"""Shortcut for returning class name."""
return unicode(self.__class__.__name__)
@classmethod
def __ignore__(cls):
"""Custom class attr that lets us control which models get ignored.
We are using this because knowing whether or not we're actually dealing
with an abstract base class is only possible late in the class's init
lifecycle.
This is used by the dynamic model loader to know if it should ignore.
"""
return cls.__name__ in ('Model',) # can add more abstract base classes here
def __repr__(self):
""""Returns a string representation of every object.
Useful for logging & error reporting.
Example:
>>> obj = MyModel()
>>> print obj
MyModel.123
Can be overridden by subclasses to customize string representation.
"""
return u"{}.{}".format(self.class_name, self.pk)
@declared_attr
def __tablename__(cls):
"""Generate a __tablename__ attr for every model that does not have
inherited tables.
Ensures table names match the model name without needing to declare it.
"""
if has_inherited_table(cls):
return None
return cls.__name__.lower()
Custom QueryMixin for friendlier database lookups
Eagle-eyed readers will have noticed the base Model class above included a QueryMixin in its class definition. This mixin is the real meat of the custom base Model class, providing a series of helper functions that make common database operations far easier to deal with. Let’s take a look at the QueryMixin now.
# -*- coding: utf-8 -*-
from flask import abort, current_app as app
from sqlalchemy.orm.exc import ObjectDeletedError
from myapp.db import db
__all__ = ('QueryMixin',)
class QueryMixin(object):
"""Mixin class for database queries."""
# CRUD methods
def save(self):
"""Save instance to database."""
db.session.add(self)
db.session.commit()
def delete(self):
"""Delete instance."""
db.session.delete(self)
db.session.commit()
# Query helpers
@classmethod
def exists(cls, **kwargs):
"""Checks if record matching kwargs exists in the database.
Returns True/False.
"""
return db.session.query(cls._and_query(kwargs).exists()).all()[0][0]
@classmethod
def find(cls, **kwargs):
"""Return filtered AND query results for passed in kwargs.
Example:
# find all instances of MyModel for first name 'John' AND last name 'Doe'
MyModel.find(first_name='John', last_name='Doe')
Returns result list or None.
"""
return cls._and_query(kwargs)
@classmethod
def find_or(cls, **kwargs):
"""Return filtered OR query results for passed in kwargs.
Example:
# find all instances of MyModel for first name 'John' OR last name 'Doe'
MyModel.find_or(first_name='John', last_name='Doe')
Returns result list or None.
"""
return cls._or_query(kwargs)
@classmethod
def find_in(cls, _or=False, **kwargs):
"""Return filtered query results for passed in attrs that match a list.
Query defaults to an AND query. If you want an OR query, pass _or=True.
"""
if _or:
return cls._or_in_query(kwargs)
else:
return cls._and_in_query(kwargs)
@classmethod
def find_not_in(cls, _or=False, **kwargs):
"""Return filtered query results for passed in attrs that do not match
a list.
Query defaults to an AND query. If you want an OR query, pass _or=True.
"""
if _or:
return cls._or_not_in_query(kwargs)
else:
return cls._and_not_in_query(kwargs)
@classmethod
def find_not_null(cls, *args):
"""Return filtered query results for passed in attrs that are not None.
Example:
# find all instances of MyModel where email and phone != null
MyModel.find_not_null('email', 'phone')
NOTE: Filtering for JSON types that are not NULL does not work. JSON
must be cast to at least text to check for a NULL value. You can verify
this yourself in the `psql` client like so:
# you will see null results show up
select * from form where custom_fields is not null;
# you will not see null results
select * from form where custom_fields::text != 'null';
Returns result list or None.
"""
filters = [getattr(cls, attr) != None for attr in args]
return cls.query.filter(*filters)
@classmethod
def first(cls, **kwargs):
"""Return first result for query.
Returns instance or None.
"""
return cls._and_query(kwargs).first()
@classmethod
def first_or_404(cls, **kwargs):
"""Get first item that matches kwargs or raise 404 error."""
item = cls._and_query(kwargs).first()
if item is None:
return abort(404)
else:
return item
@classmethod
def get(cls, pk):
"""Get item by primary key.
Returns instance or `None`.
"""
return cls.query.get(pk)
@classmethod
def get_active_or_404(cls, pk):
"""Get item by primary key or 404 only if it is active."""
item = cls.query.get_or_404(pk)
if item.active:
return item
else:
return abort(404)
@classmethod
def get_or_404(cls, pk):
"""Get item by primary key or 404."""
return cls.query.get_or_404(pk)
########################################
# Internal methods; Do not use directly
########################################
@classmethod
def _filters(cls, filters):
"""Return filter list from kwargs."""
return [getattr(cls, attr)==filters[attr] for attr in filters]
@classmethod
def _filters_in(cls, filters):
"""Return IN filter list from kwargs."""
return [getattr(cls, attr).in_(filters[attr]) for attr in filters]
@classmethod
def _filters_not_in(cls, filters):
"""Return NOT IN filter list from kwargs."""
return [getattr(cls, attr).notin_(filters[attr]) for attr in filters]
@classmethod
def _and_query(cls, filters):
"""Execute AND query.
Returns BaseQuery.
"""
return cls.query.filter(db.and_(*cls._filters(filters)))
@classmethod
def _and_in_query(cls, filters):
"""Execute AND query.
Returns BaseQuery.
"""
return cls.query.filter(db.and_(*cls._filters_in(filters)))
@classmethod
def _and_not_in_query(cls, filters):
"""Execute AND NOT IN query.
Returns BaseQuery.
"""
return cls.query.filter(db.and_(*cls._filters_not_in(filters)))
@classmethod
def _or_query(cls, filters):
"""Execute OR query.
Returns BaseQuery.
"""
return cls.query.filter(db.or_(*cls._filters(filters)))
@classmethod
def _or_in_query(cls, filters):
"""Execute OR IN query.
Returns BaseQuery.
"""
return cls.query.filter(db.or_(*cls._filters_in(filters)))
@classmethod
def _or_not_in_query(cls, filters):
"""Execute OR NOT IN query.
Returns BaseQuery.
"""
return cls.query.filter(db.or_(*cls._filters_not_in(filters)))
The QueryMixin in action
To give a quick preview of our QueryMixin in action, take a look at how much simpler common database operations are now:
# create & save objects
user = User(name=u'John Doe', age=32, email=u'johndoe@example.org', city=u'Los Angeles', state=u'CA')
user.save()
# Find some users
users_who_are_32 = User.find(age=32)
users_in_los_angeles = User.find(city=u'Los Angeles')
users_in_ca = User.find(state=u'CA')
los_angelinos_who_are_32 = User.find(city=u'Los Angeles', age=32)
users_in_30s = User.find_in(age=range(30,40))
users_not_in_ca_or_wa = User.find_not_in(state=(u'CA',u'WA'))
users_in_southeast = User.find_in(state=(u'GA', u'TN', u'AL', u'MS', u'NC', u'SC'))