mirror of
https://github.com/Hopiu/django-model-utils.git
synced 2026-03-16 20:00:23 +00:00
Remove PassThroughManager
As of Django 1.7, QuerySet.as_manager() achieves the same result.
This commit is contained in:
jarekwg
parent
7a33e14f4b
commit
2824ec2e48
3 changed files with 2 additions and 270 deletions
|
|
@ -224,91 +224,3 @@ class QueryManagerMixin(object):
|
|||
|
||||
class QueryManager(QueryManagerMixin, models.Manager):
|
||||
pass
|
||||
|
||||
|
||||
class PassThroughManagerMixin(object):
|
||||
"""
|
||||
A mixin that enables you to call custom QuerySet methods from your manager.
|
||||
"""
|
||||
|
||||
# pickling causes recursion errors
|
||||
_deny_methods = ['__getstate__', '__setstate__', '__getinitargs__',
|
||||
'__getnewargs__', '__copy__', '__deepcopy__', '_db',
|
||||
'__slots__']
|
||||
|
||||
def __init__(self, queryset_cls=None):
|
||||
self._queryset_cls = queryset_cls
|
||||
super(PassThroughManagerMixin, self).__init__()
|
||||
|
||||
def __getattr__(self, name):
|
||||
if name in self._deny_methods:
|
||||
raise AttributeError(name)
|
||||
if django.VERSION < (1, 6, 0):
|
||||
return getattr(self.get_query_set(), name)
|
||||
return getattr(self.get_queryset(), name)
|
||||
|
||||
def __dir__(self):
|
||||
"""
|
||||
Allow introspection via dir() and ipythonesque tab-discovery.
|
||||
|
||||
We do dir(type(self)) because to do dir(self) would be a recursion
|
||||
error.
|
||||
We call dir(self.get_query_set()) because it is possible that the
|
||||
queryset returned by get_query_set() is interesting, even if
|
||||
self._queryset_cls is None.
|
||||
"""
|
||||
my_values = frozenset(dir(type(self)))
|
||||
my_values |= frozenset(dir(self.get_query_set()))
|
||||
return list(my_values)
|
||||
|
||||
def get_queryset(self):
|
||||
try:
|
||||
qs = super(PassThroughManagerMixin, self).get_queryset()
|
||||
except AttributeError:
|
||||
qs = super(PassThroughManagerMixin, self).get_query_set()
|
||||
if self._queryset_cls is not None:
|
||||
qs = qs._clone(klass=self._queryset_cls)
|
||||
return qs
|
||||
|
||||
get_query_set = get_queryset
|
||||
|
||||
@classmethod
|
||||
def for_queryset_class(cls, queryset_cls):
|
||||
return create_pass_through_manager_for_queryset_class(
|
||||
cls, queryset_cls)
|
||||
|
||||
|
||||
class PassThroughManager(PassThroughManagerMixin, 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_queryset`
|
||||
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 the
|
||||
``for_queryset_class`` class method.
|
||||
|
||||
class PostQuerySet(QuerySet):
|
||||
def enabled(self):
|
||||
return self.filter(disabled=False)
|
||||
|
||||
class Post(models.Model):
|
||||
objects = PassThroughManager.for_queryset_class(PostQuerySet)()
|
||||
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
def create_pass_through_manager_for_queryset_class(base, queryset_cls):
|
||||
class _PassThroughManager(base):
|
||||
def __init__(self, *args, **kwargs):
|
||||
return super(_PassThroughManager, self).__init__(*args, **kwargs)
|
||||
|
||||
def get_queryset(self):
|
||||
qs = super(_PassThroughManager, self).get_queryset()
|
||||
return qs._clone(klass=queryset_cls)
|
||||
|
||||
get_query_set = get_queryset
|
||||
|
||||
return _PassThroughManager
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ from django.utils.translation import ugettext_lazy as _
|
|||
|
||||
from model_utils.models import TimeStampedModel, StatusModel, TimeFramedModel
|
||||
from model_utils.tracker import FieldTracker, ModelTracker
|
||||
from model_utils.managers import QueryManager, InheritanceManager, PassThroughManager
|
||||
from model_utils.managers import QueryManager, InheritanceManager
|
||||
from model_utils.fields import SplitField, MonitorField, StatusField
|
||||
from model_utils.tests.fields import MutableField
|
||||
from model_utils import Choices
|
||||
|
|
@ -201,76 +201,6 @@ class FeaturedManager(models.Manager):
|
|||
get_query_set = get_queryset
|
||||
|
||||
|
||||
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_queryset(self):
|
||||
return DudeQuerySet(self.model).abiding()
|
||||
|
||||
get_query_set = get_queryset
|
||||
|
||||
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(default=False)
|
||||
|
||||
objects = PassThroughManager(DudeQuerySet)
|
||||
abiders = AbidingManager()
|
||||
|
||||
|
||||
class Car(models.Model):
|
||||
name = models.CharField(max_length=20)
|
||||
owner = models.ForeignKey(Dude, related_name='cars_owned')
|
||||
|
||||
objects = PassThroughManager(DudeQuerySet)
|
||||
|
||||
|
||||
class SpotManager(PassThroughManager):
|
||||
def get_queryset(self):
|
||||
return super(SpotManager, self).get_queryset().filter(secret=False)
|
||||
|
||||
get_query_set = get_queryset
|
||||
|
||||
|
||||
class SpotQuerySet(models.query.QuerySet):
|
||||
def closed(self):
|
||||
return self.filter(closed=True)
|
||||
|
||||
def secured(self):
|
||||
return self.filter(secure=True)
|
||||
|
||||
|
||||
class Spot(models.Model):
|
||||
name = models.CharField(max_length=20)
|
||||
secure = models.BooleanField(default=True)
|
||||
closed = models.BooleanField(default=False)
|
||||
secret = models.BooleanField(default=False)
|
||||
owner = models.ForeignKey(Dude, related_name='spots_owned')
|
||||
|
||||
objects = SpotManager.for_queryset_class(SpotQuerySet)()
|
||||
|
||||
|
||||
class Tracked(models.Model):
|
||||
name = models.CharField(max_length=20)
|
||||
number = models.IntegerField()
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
from __future__ import unicode_literals
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
import pickle
|
||||
try:
|
||||
from unittest import skipUnless
|
||||
except ImportError: # Python 2.6
|
||||
|
|
@ -25,7 +24,7 @@ from model_utils.tests.models import (
|
|||
InheritanceManagerTestParent, InheritanceManagerTestChild1,
|
||||
InheritanceManagerTestChild2, TimeStamp, Post, Article, Status,
|
||||
StatusPlainTuple, TimeFrame, Monitored, MonitorWhen, MonitorWhenEmpty, StatusManagerAdded,
|
||||
TimeFrameManagerAdded, Dude, SplitFieldAbstractParent, Car, Spot,
|
||||
TimeFrameManagerAdded, SplitFieldAbstractParent,
|
||||
ModelTracked, ModelTrackedFK, ModelTrackedNotDefault, ModelTrackedMultiple, InheritedModelTracked,
|
||||
Tracked, TrackedFK, TrackedNotDefault, TrackedNonFieldAttr, TrackedMultiple,
|
||||
InheritedTracked, StatusFieldDefaultFilled, StatusFieldDefaultNotFilled,
|
||||
|
|
@ -1206,115 +1205,6 @@ class SouthFreezingTests(TestCase):
|
|||
self.assertEqual(kwargs['no_check_for_status'], 'True')
|
||||
|
||||
|
||||
|
||||
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)
|
||||
with self.assertRaises(AttributeError):
|
||||
Dude.abiders.all().get_stats()
|
||||
|
||||
|
||||
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)
|
||||
|
||||
|
||||
def test_queryset_not_available_on_related_manager(self):
|
||||
dude = Dude.objects.by_name('Duder').get()
|
||||
Car.objects.create(name='Ford', owner=dude)
|
||||
self.assertFalse(hasattr(dude.cars_owned, 'by_name'))
|
||||
|
||||
|
||||
def test_using_dir(self):
|
||||
# make sure introspecing via dir() doesn't actually cause queries,
|
||||
# just as a sanity check.
|
||||
with self.assertNumQueries(0):
|
||||
querysets_to_dir = (
|
||||
Dude.objects,
|
||||
Dude.objects.by_name('Duder'),
|
||||
Dude.objects.all().by_name('Duder'),
|
||||
Dude.abiders,
|
||||
Dude.abiders.rug_positive(),
|
||||
Dude.abiders.all().rug_positive()
|
||||
)
|
||||
for qs in querysets_to_dir:
|
||||
self.assertTrue('by_name' in dir(qs))
|
||||
self.assertTrue('abiding' in dir(qs))
|
||||
self.assertTrue('rug_positive' in dir(qs))
|
||||
self.assertTrue('rug_negative' in dir(qs))
|
||||
# some standard qs methods
|
||||
self.assertTrue('count' in dir(qs))
|
||||
self.assertTrue('order_by' in dir(qs))
|
||||
self.assertTrue('select_related' in dir(qs))
|
||||
# make sure it's been de-duplicated
|
||||
self.assertEqual(1, dir(qs).count('distinct'))
|
||||
|
||||
# manager only method.
|
||||
self.assertTrue('get_stats' in dir(Dude.abiders))
|
||||
# manager only method shouldn't appear on the non AbidingManager
|
||||
self.assertFalse('get_stats' in dir(Dude.objects))
|
||||
# standard manager methods
|
||||
self.assertTrue('get_query_set' in dir(Dude.abiders))
|
||||
self.assertTrue('contribute_to_class' in dir(Dude.abiders))
|
||||
|
||||
|
||||
|
||||
class CreatePassThroughManagerTests(TestCase):
|
||||
def setUp(self):
|
||||
self.dude = Dude.objects.create(name='El Duderino')
|
||||
self.other_dude = Dude.objects.create(name='Das Dude')
|
||||
|
||||
def test_reverse_manager(self):
|
||||
Spot.objects.create(
|
||||
name='The Crib', owner=self.dude, closed=True, secure=True,
|
||||
secret=False)
|
||||
self.assertEqual(self.dude.spots_owned.closed().count(), 1)
|
||||
Spot.objects.create(
|
||||
name='The Crux', owner=self.other_dude, closed=True, secure=True,
|
||||
secret=False
|
||||
)
|
||||
self.assertEqual(self.dude.spots_owned.closed().all().count(), 1)
|
||||
self.assertEqual(self.dude.spots_owned.closed().count(), 1)
|
||||
|
||||
def test_related_queryset_pickling(self):
|
||||
Spot.objects.create(
|
||||
name='The Crib', owner=self.dude, closed=True, secure=True,
|
||||
secret=False)
|
||||
qs = self.dude.spots_owned.closed()
|
||||
pickled_qs = pickle.dumps(qs)
|
||||
unpickled_qs = pickle.loads(pickled_qs)
|
||||
self.assertEqual(unpickled_qs.secured().count(), 1)
|
||||
|
||||
def test_related_queryset_superclass_method(self):
|
||||
Spot.objects.create(
|
||||
name='The Crib', owner=self.dude, closed=True, secure=True,
|
||||
secret=False)
|
||||
Spot.objects.create(
|
||||
name='The Secret Crib', owner=self.dude, closed=False, secure=True,
|
||||
secret=True)
|
||||
self.assertEqual(self.dude.spots_owned.count(), 1)
|
||||
|
||||
def test_related_manager_create(self):
|
||||
self.dude.spots_owned.create(name='The Crib', closed=True, secure=True)
|
||||
|
||||
|
||||
class FieldTrackerTestCase(TestCase):
|
||||
|
||||
tracker = None
|
||||
|
|
|
|||
Loading…
Reference in a new issue