Initial commit

.gitignore

@@ -0,0 +1,9 @@
+__pycache__
+*.pyc
+
+.coverage
+
+_build
+
+*.swp
+*.swo

.readthedocs.yml

@@ -0,0 +1,14 @@
+version: 2
+
+sphinx:
+  configuration: docs/conf.py
+
+# Optionally build your docs in additional formats such as PDF and ePub
+formats: all
+
+python:
+  version: 3.7
+  install:
+  - requirements: docs/requirements.txt
+  - method: pip
+    path: .

README.rst

@@ -0,0 +1,18 @@
+django-flexquery
+================
+
+This library aims to provide a new way of declaring reusable QuerySet filtering
+logic in your Django project, incorporating the DRY principle and maximizing user
+experience and performance by allowing you to decide between sub-queries and JOINs.
+
+Its strengths are, among others:
+
+* Easy to learn in minutes
+* Cleanly integrates with Django's ORM
+* Small code footprint, less bugs - ~150 lines of code (LoC)
+* 100% test coverage
+* Fully documented code, formatted using the excellent `Black Code Formatter
+  <https://github.com/python/black>`_.
+
+See the `documentation at Read The Docs <https://django-flexquery.readthedocs.org>`_
+to convince yourself.

django_flexquery/__init__.py

@@ -0,0 +1,16 @@
+"""
+django-flexquery
+
+Reusable QuerySet filtering logic for Django.
+"""
+
+try:
+    from .flexquery import Manager, Q, FlexQuery, QuerySet
+except ImportError:
+    # Django is missing, just provide version
+    __all__ = []
+else:
+    __all__ = ["FlexQuery", "Manager", "Q", "QuerySet"]
+
+
+__version__ = "0.0.0"

django_flexquery/flexquery.py

@@ -0,0 +1,194 @@
+"""
+This module provides a convenient way to declare custom filtering logic with Django's
+model managers in a reusable fashion using Q objects.
+"""
+
+import inspect
+
+from django.core.exceptions import ImproperlyConfigured
+from django.db.models import Manager as _Manager, Q as _Q, QuerySet as _QuerySet
+from django.db.models.constants import LOOKUP_SEP
+from django.utils.decorators import classonlymethod
+
+
+class FlexQueryType(type):
+    """
+    Custom metaclass for FlexQuery that implements the descriptor pattern.
+    """
+
+    def __get__(self, instance, owner):
+        if isinstance(instance, (_Manager, _QuerySet)):
+            # Return a new FlexQuery instance bound to the holding object
+            return self(instance)
+        # Otherwise, preserve the unbound FlexQuery type
+        return self
+
+    def __repr__(self):
+        return "<Unbound %s %r>" % (
+            self.__name__,
+            (self.filter_func if self.q_func is None else self.q_func).__name__,
+        )
+
+
+class FlexQuery(metaclass=FlexQueryType):
+    """
+    Flexibly provides model-specific query constraints as an attribute of Manager
+    or QuerySet objects.
+
+    When a sub-type of FlexQuery is accessed as class attribute of a Manager or
+    QuerySet object, its metaclass, which is implemented as a descriptor, will
+    automatically initialize and return an instance of the FlexQuery type bound to
+    the holding Manager or QuerySet.
+
+    .. automethod:: __call__
+    .. automethod:: __init__
+    """
+
+    # Will be set when creating sub-types
+    filter_func = None
+    q_func = None
+
+    def __call__(self, *args, **kwargs):
+        """Filters the base QuerySet with the provided filter function or Q object.
+
+        All arguments are passed through to the FlexQuery's custom function.
+
+        :returns QuerySet:
+        """
+        if self.filter_func is not None:
+            return self.filter_func(self.base.all(), *args, **kwargs)
+        return self.base.filter(self.q_func(*args, **kwargs))
+
+    def __init__(self, base):
+        """Initializes a FlexQuery that will perform its filtering on the given base.
+
+        :param base: Instance to use as base for filtering.
+        :type  base: Manager, QuerySet
+        :raises ImproperlyConfigured:
+            If this FlexQuery type wasn't created by one of the from_*() classmethods.
+        :raises TypeError: If base is of unsupported type.
+        """
+        if self.filter_func is None and self.q_func is None:
+            raise ImproperlyConfigured(
+                "Use one of the from_*() classmethods to create a FlexQuery type."
+            )
+        if not isinstance(base, (_Manager, _QuerySet)):
+            raise TypeError(
+                "Can only bind %s to a Manager or QuerySet, but not to %r."
+                % (self.__class__.__name__, base)
+            )
+        self.base = base
+
+    def __repr__(self):
+        return "<Bound %s %r, base=%r>" % (
+            self.__class__.__name__,
+            (self.filter_func if self.q_func is None else self.q_func).__name__,
+            self.base,
+        )
+
+    def as_q(self, *args, **kwargs):
+        """Returns a Q object representing the custom filters.
+
+        The Q object is either retrieved from the configured Q function or, if a
+        filter function was provided instead, will contain a sub-query ensuring the
+        object is in the base QuerySet filtered by that filter function.
+
+        All arguments are passed through to the FlexQuery's custom function.
+
+        :returns Q:
+        """
+        if self.filter_func is not None:
+            return Q(pk__in=self.filter_func(self.base.all(), *args, **kwargs))
+        return self.q_func(*args, **kwargs)
+
+    @classonlymethod
+    def from_filter(cls, func):
+        """Creates a FlexQuery type from a filter function.
+
+        :param func:
+            Callable taking a QuerySet as first positional argument, applying some
+            filtering on it and returning it back.
+        :type  func: callable
+        :returns FlexQueryType:
+        """
+        return type(cls)(
+            "%sFromFilterFunction" % cls.__name__,
+            (cls,),
+            {"filter_func": staticmethod(func)},
+        )
+
+    @classonlymethod
+    def from_q(cls, func):
+        """Creates a FlexQuery type from a Q function.
+
+        :param func: Callable returning a Q object.
+        :type  func: callable
+        :returns FlexQueryType:
+        """
+        return type(cls)(
+            "%sFromQFunction" % cls.__name__, (cls,), {"q_func": staticmethod(func)}
+        )
+
+
+class Manager(_Manager):
+    """
+    Use this Manager class' from_queryset() method if you want to derive a Manager from
+    a QuerySet that has FlexQuery's defined. If Django's native Manager.from_queryset()
+    was used instead, all FlexQuery's would be lost.
+    """
+
+    @classmethod
+    def _get_queryset_methods(cls, queryset_class):
+        methods = super()._get_queryset_methods(queryset_class)
+        methods.update(
+            inspect.getmembers(
+                queryset_class,
+                predicate=lambda member: isinstance(member, FlexQueryType)
+                and not getattr(member, "queryset_only", None),
+            )
+        )
+        return methods
+
+
+class Q(_Q):
+    """
+    A custom Q implementation that allows prefixing existing Q objects with some
+    related field name dynamically.
+    """
+
+    def prefix(self, prefix):
+        """Recursively copies the Q object, prefixing all lookup keys.
+
+        The prefix and the existing filter key are delimited by the lookup separator __.
+        Use this feature to delegate existing query constraints to a related field.
+
+        :param prefix:
+            Name of the related field to prepend to existing lookup keys. This isn't
+            restricted to a single relation level, something like "tree__fruit"
+            is perfectly valid as well.
+        :type  prefix: str
+        :returns Q:
+        """
+        return type(self)(
+            *(
+                child.prefix(prefix)
+                if isinstance(child, Q)
+                else (prefix + LOOKUP_SEP + child[0], child[1])
+                for child in self.children
+            ),
+            _connector=self.connector,
+            _negated=self.negated,
+        )
+
+
+class QuerySet(_QuerySet):
+    """
+    Adds support for deriving a Manager from QuerySet via as_manager(), preserving
+    the FlexQuery's.
+    """
+
+    @classmethod
+    def as_manager(cls):
+        manager = Manager.from_queryset(cls)()
+        manager._built_with_as_manager = True
+        return manager

django_flexquery_tests/models.py

@@ -0,0 +1 @@
+from .tests import AModel

django_flexquery_tests/settings.py

@@ -0,0 +1,8 @@
+"""
+Django settings for tests of the django-flexquery project.
+"""
+
+SECRET_KEY = "om-l3b3%$jutfygk%pyg#i0hp@g$v4b=-jcsa_)+%^7nig@up4"
+DEBUG = True
+INSTALLED_APPS = ["django_flexquery_tests"]
+DATABASES = {"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}}

django_flexquery_tests/tests.py

@@ -0,0 +1,120 @@
+import re
+
+from django.core.exceptions import ImproperlyConfigured
+from django.db import models
+from django.test import SimpleTestCase, TestCase
+
+from django_flexquery import *
+
+
+filter_func = lambda qs: qs.filter(a=42)
+q_func = lambda: Q(a=42)
+
+fq_from_filter = FlexQuery.from_filter(filter_func)
+fq_from_q = FlexQuery.from_q(q_func)
+
+
+class QS(QuerySet):
+    fq_from_filter = fq_from_filter
+    fq_from_q = fq_from_q
+
+
+class AModel(models.Model):
+    objects = QS.as_manager()
+    a = models.IntegerField()
+
+
+class FlexQueryTestCase(TestCase):
+    def setUp(self):
+        AModel.objects.create(a=24)
+        AModel.objects.create(a=42)
+
+    # Derive Manager from QuerySet
+
+    def test_manager_from_queryset(self):
+        class Man(Manager.from_queryset(QS)):
+            pass
+
+        self.assertTrue(hasattr(Man, "fq_from_q"))
+
+    def test_queryset_as_manager(self):
+        self.assertTrue(hasattr(AModel.objects, "fq_from_q"))
+
+    # Sub-type creation
+
+    def test_from_filter(self):
+        self.assertIs(fq_from_filter.filter_func, filter_func)
+        self.assertIs(fq_from_filter.q_func, None)
+
+    def test_from_q(self):
+        self.assertIs(fq_from_q.filter_func, None)
+        self.assertIs(fq_from_q.q_func, q_func)
+
+    # Invalid initialization
+
+    def test_invalid_base(self):
+        Some = type("Some", (), {"fq_from_q": fq_from_q})
+        with self.assertRaises(TypeError):
+            fq_from_q(object)
+
+    def test_uninitialized_type(self):
+        with self.assertRaises(ImproperlyConfigured):
+            FlexQuery(AModel.objects)
+
+    # Access FlexQuery type as attribute
+
+    def test_class_access(self):
+        self.assertIs(QS.fq_from_q, fq_from_q)
+
+    def test_object_access(self):
+        self.assertTrue(isinstance(AModel.objects.fq_from_q, fq_from_q))
+
+    # __repr__()
+
+    def test_class_repr(self):
+        self.assertRegex(
+            repr(fq_from_q),
+            r"^<Unbound FlexQueryFromQFunction %s>$"
+            % re.escape(repr(fq_from_q.q_func.__name__)),
+        )
+
+    def test_object_repr(self):
+        bound = AModel.objects.fq_from_q
+        self.assertRegex(
+            repr(bound),
+            r"^<Bound FlexQueryFromQFunction %s, base=%s>$"
+            % (re.escape(repr(bound.q_func.__name__)), re.escape(repr(bound.base))),
+        )
+
+    # FlexQuery.__call__()
+
+    def test_filter_native(self):
+        self.assertTrue(AModel.objects.fq_from_filter().count(), 1)
+
+    def test_q_to_filter(self):
+        self.assertTrue(AModel.objects.fq_from_q().count(), 1)
+
+    # FlexQuery.as_q()
+
+    def test_filter_to_q(self):
+        q = AModel.objects.fq_from_filter.as_q()
+        self.assertEqual(len(q), 1)
+        self.assertEqual(q.children[0][0], "pk__in")
+        self.assertTrue(isinstance(q.children[0][1], QuerySet))
+
+    def test_q_native(self):
+        self.assertEqual(AModel.objects.fq_from_q.as_q(), Q(a=42))
+
+
+class QTestCase(SimpleTestCase):
+    def setUp(self):
+        self.q = (~Q(a=42) | Q(b=43) & Q(c=44)).prefix("x")
+
+    def test_prefix_copies_connector(self):
+        self.assertEqual(self.q.connector, Q.OR)
+
+    def test_prefix_copies_negated(self):
+        self.assertTrue(self.q.children[0].negated, True)
+
+    def test_prefix_keys(self):
+        self.assertEqual(self.q.children[0].children[0][0], "x__a")