Django queryset convenience filters I wrote to make Locu data quality go round, but probably useful anywhere your DB schema gets unnecessarily complicated

django_filters.py

"""
Reusable Django queryset filters. Useful for filtering on complex rule sets,
and determining independent reasons why an object fails to pass said rules.
"""

from collections import namedtuple


class FilterSpec(
  namedtuple('FilterSpec', ['message', 'kwargs', 'exclude', 'distinct'])):
  """
  Specify filters to be applied to querysets.
  message: returned in get_message when queryset *fails* filter.
  kwargs: Will be used to filter the queryset, constructing AND clause
    between the conditions.
  exclude: Kwargs will be applied using `exclude` instead of `filter`,
    constructing OR clause between the conditions.
  distinct: Enforce distinctness after other conditions have been applied.
  """
  def __new__(cls, message, kwargs, exclude=False, distinct=False):
    return (super(FilterSpec, cls)
            .__new__(cls, message, kwargs, exclude, distinct))

  def apply(cls, queryset):
    """
    Filters the queryset using the defined conditions and returns the result.
    """
    if cls.exclude:
      # OR together exclude kwargs by chaining
      for kwarg, value in cls.kwargs.iteritems():
        queryset = queryset.exclude(**{kwarg: value})
    else:
      queryset = queryset.filter(**cls.kwargs)
    if cls.distinct:
      queryset = queryset.distinct()
    return queryset
    
  def get_message(cls, queryset):
    """If no objects passed the filter, return the failure message."""
    if not cls.apply(queryset).exists():
      return cls.message
    return ''


class ChainedFilterSpec(namedtuple('ChainedFilterSpec', ['filters'])):
  """A filter that chains together child filters."""
  def apply(cls, queryset):
    for child_filter in cls.filters:
      queryset = child_filter.apply(queryset)
    return queryset

  def get_message(cls, queryset):
    messages = []
    for child_filter in cls.filters:
      # Individually run each child filter on the original queryset.
      messages.append(child_filter.get_message(queryset))
    # Join and deduplicate messages.
    return '\n'.join(msg for msg in set(messages) if msg)


class FilterExamples(object):
  """
  Dummy example filters.

  Usage:
    print FilterExamples.EXCLUDE_BLANK.get_message(some_django_queryset)
    result_queryset = FilterExamples.EXCLUDE_BLANK.apply(some_django_queryset)
  """

  DUMMY1 = FilterSpec(
    u"This object was created before 2015, so we don't want it",
    {'date__created__gte': '2015-01-01'}
  )

  DUMMY2 = FilterSpec(
    u"Object has not been approved",
    {'acceptable': False}
  )

  # This applies the child filters to a queryset in order.
  COMBO = ChainedFilterSpec([
    DUMMY1,
    DUMMY2,
  ])

  # If any exclude condition is independently true, an object will not pass.
  EXCLUDE_BLANK = FilterSpec(
    u'Some required fields were left blank',
    {field: None for field in ['field1', 'field2', 'field3']},
    exclude=True
  )

  # If selecting on a many-to-one/many-to-many foreign relation, use DISTINCT
  # to prevent duplicates.
  DISTINCT_EXAMPLE = FilterSpec(
    u"Associated with foreign objects that have some property we don't want",
    {'foreignobj__someproperty__in': ['badness1', 'badness2']},
    exclude=True,
    distinct=True
  )