Merged in PassThroughManager from Paul McLanahan.

This commit is contained in:
Carl Meyer 2011-03-29 13:10:50 -04:00
commit 36b99304c2
5 changed files with 200 additions and 6 deletions

View file

@ -2,5 +2,7 @@ Carl Meyer <carl@dirtcircle.com>
Jannis Leidel <jannis@leidel.info>
Gregor Müllegger <gregor@muellegger.de>
Jeff Elmore <jeffelmore.org>
Paul McLanahan <paul@mclanahan.net>
zyegfryed
sayane

View file

@ -392,3 +392,87 @@ same additional methods::
.. _created by George Sakkis: http://djangosnippets.org/snippets/2117/
PassThroughManager
==================
The ``PassThroughManager`` class (`contributed by Paul McLanahan`_) solves
the same problem as the above ``manager_from`` function. This class, however,
accomplishes it in a different way. The reason it exists is that the dynamically
generated ``QuerySet`` classes created by the ``manager_from`` function are
not picklable. It's probably not often that a ``QuerySet`` is pickled, but
it is a documented feature of the Django ``QuerySet`` class, and this method
maintains that functionality.
``PassThroughManager`` is a subclass of ``django.db.models.manager.Manager``,
so all that is required is that you change your custom managers to inherit from
``PassThroughManager`` instead of Django's built-in ``Manager`` class. Once you
do this, create your custom ``QuerySet`` class, and have your manager's
``get_query_set`` method return instances of said class, then all of the
methods you add to your custom ``QuerySet`` class will be available from your
manager as well::
from datetime import datetime
from django.db import models
from django.db.models.query import QuerySet
class PostQuerySet(QuerySet):
def by_author(self, user):
return self.filter(user=user)
def published(self):
return self.filter(published__lte=datetime.now())
def unpublished(self):
return self.filter(published__gte=datetime.now())
class PostManager(PassThroughManager):
def get_query_set(self):
PostQuerySet(self.model, using=self._db)
def get_stats(self):
return {
'published_count': self.published().count(),
'unpublished_count': self.unpublished().count(),
}
class Post(models.Model):
user = models.ForeignKey(User)
published = models.DateTimeField()
objects = PostManager()
Post.objects.get_stats()
Post.objects.published()
Post.objects.by_author(user=request.user).unpublished()
Alternatively, if you don't need any methods on your manager that shouldn't also
be on your queryset, a shortcut is available. ``PassThroughManager``'s
constructor takes an optional argument. If you pass it a ``QuerySet`` subclass
it will automatically use that class when creating querysets for the manager::
from datetime import datetime
from django.db import models
from django.db.models.query import QuerySet
class PostQuerySet(QuerySet):
def by_author(self, user):
return self.filter(user=user)
def published(self):
return self.filter(published__lte=datetime.now())
def unpublished(self):
return self.filter(published__gte=datetime.now())
class Post(models.Model):
user = models.ForeignKey(User)
published = models.DateTimeField()
objects = PassThroughManager(PostQuerySet)
Post.objects.published()
Post.objects.by_author(user=request.user).unpublished()
.. _contributed by Paul McLanahan: http://paulm.us/post/3717466639/passthroughmanager-for-django

View file

@ -81,8 +81,45 @@ class QueryManager(models.Manager):
return qs
class PassThroughManager(models.Manager):
"""
Inherit from this Manager to enable you to call any methods from your
custom QuerySet class from your manager. Simply define your QuerySet
class, and return an instance of it from your manager's `get_query_set`
method.
Alternately, if you don't need any extra methods on your manager that
aren't on your QuerySet, then just pass your QuerySet class to this
class' constructer.
class PostQuerySet(QuerySet):
def enabled(self):
return self.filter(disabled=False)
class Post(models.Model):
objects = PassThroughManager(PostQuerySet)
"""
# pickling causes recursion errors
_deny_methods = ['__getstate__', '__setstate__']
def __init__(self, queryset_cls=None):
self._queryset_cls = queryset_cls
super(PassThroughManager, self).__init__()
def __getattr__(self, name):
if name in self._deny_methods:
raise AttributeError(name)
return getattr(self.get_query_set(), name)
def get_query_set(self):
if self._queryset_cls is not None:
return self._queryset_cls(self.model, using=self._db)
return super(PassThroughManager, self).get_query_set()
def manager_from(*mixins, **kwds):
'''
"""
Returns a Manager instance with extra methods, also available and
chainable on generated querysets.
@ -98,7 +135,8 @@ def manager_from(*mixins, **kwds):
:keyword manager_cls: The base manager class to extend from
(``django.db.models.manager.Manager`` by default).
'''
"""
# collect separately the mixin classes and methods
bases = [kwds.get('queryset_cls', QuerySet)]
methods = {}

View file

@ -1,9 +1,8 @@
from datetime import datetime
from django.db import models
from django.utils.translation import ugettext_lazy as _
from model_utils.models import InheritanceCastModel, TimeStampedModel, StatusModel, TimeFramedModel
from model_utils.managers import QueryManager, manager_from, InheritanceManager
from model_utils.managers import QueryManager, manager_from, InheritanceManager, PassThroughManager
from model_utils.fields import SplitField, MonitorField
from model_utils import Choices
@ -124,9 +123,40 @@ class Entry(models.Model):
author = models.CharField(max_length=20)
published = models.BooleanField()
feature = models.BooleanField(default=False)
objects = manager_from(AuthorMixin, PublishedMixin, unpublished)
broken = manager_from(PublishedMixin, manager_cls=FeaturedManager)
featured = manager_from(PublishedMixin,
manager_cls=FeaturedManager,
queryset_cls=ByAuthorQuerySet)
class DudeQuerySet(models.query.QuerySet):
def abiding(self):
return self.filter(abides=True)
def rug_positive(self):
return self.filter(has_rug=True)
def rug_negative(self):
return self.filter(has_rug=False)
def by_name(self, name):
return self.filter(name__iexact=name)
class AbidingManager(PassThroughManager):
def get_query_set(self):
return DudeQuerySet(self.model, using=self._db).abiding()
def get_stats(self):
return {
'abiding_count': self.count(),
'rug_count': self.rug_positive().count(),
}
class Dude(models.Model):
abides = models.BooleanField(default=True)
name = models.CharField(max_length=20)
has_rug = models.BooleanField()
objects = PassThroughManager(DudeQuerySet)
abiders = AbidingManager()

View file

@ -1,3 +1,8 @@
try:
import cPickle as pickle
except ImportError:
import pickle
from datetime import datetime, timedelta
import django
@ -17,7 +22,7 @@ from model_utils.tests.models import (
InheritParent, InheritChild, InheritChild2, InheritanceManagerTestParent,
InheritanceManagerTestChild1, InheritanceManagerTestChild2,
TimeStamp, Post, Article, Status, StatusPlainTuple, TimeFrame, Monitored,
StatusManagerAdded, TimeFrameManagerAdded, Entry)
StatusManagerAdded, TimeFrameManagerAdded, Entry, Dude)
class GetExcerptTests(TestCase):
@ -413,6 +418,7 @@ if 'south' in settings.INSTALLED_APPS:
NoRendered._meta.get_field,
'_body_excerpt')
class ManagerFromTests(TestCase):
def setUp(self):
Entry.objects.create(author='George', published=True)
@ -434,3 +440,37 @@ class ManagerFromTests(TestCase):
def test_cant_reconcile_qs_class(self):
self.assertRaises(TypeError, Entry.broken.all)
def test_queryset_pickling_fails(self):
qs = Entry.objects.all()
def dump_load():
pqs = pickle.dumps(qs)
upqs = pickle.loads(pqs)
self.assertRaises(pickle.PicklingError, dump_load)
class PassThroughManagerTests(TestCase):
def setUp(self):
Dude.objects.create(name='The Dude', abides=True, has_rug=False)
Dude.objects.create(name='His Dudeness', abides=False, has_rug=True)
Dude.objects.create(name='Duder', abides=False, has_rug=False)
Dude.objects.create(name='El Duderino', abides=True, has_rug=True)
def test_chaining(self):
self.assertEqual(Dude.objects.by_name('Duder').count(), 1)
self.assertEqual(Dude.objects.all().by_name('Duder').count(), 1)
self.assertEqual(Dude.abiders.rug_positive().count(), 1)
self.assertEqual(Dude.abiders.all().rug_positive().count(), 1)
def test_manager_only_methods(self):
stats = Dude.abiders.get_stats()
self.assertEqual(stats['rug_count'], 1)
def notonqs():
Dude.abiders.all().get_stats()
self.assertRaises(AttributeError, notonqs)
def test_queryset_pickling(self):
qs = Dude.objects.all()
saltyqs = pickle.dumps(qs)
unqs = pickle.loads(saltyqs)
self.assertEqual(unqs.by_name('The Dude').count(), 1)