python类QuerySet()的实例源码

def get_queryset(self):
        if self.queryset is not None:
            queryset = self.queryset
            if isinstance(queryset, QuerySet):
                queryset = queryset.all()
        elif self.model is not None:
            queryset = self.model._default_manager.all()
        else:
            raise ImproperlyConfigured(
                "%(cls)s is missing a QuerySet. Define "
                "%(cls)s.model, %(cls)s.queryset, or override "
                "%(cls)s.get_queryset()." % {
                    'cls': self.__class__.__name__
                }
            )
        ordering = self.get_ordering()
        if ordering:
            if isinstance(ordering, six.string_types):
                ordering = (ordering,)
            queryset = queryset.order_by(*ordering)
        return queryset

def _get_queryset(klass):
    """
    Returns a QuerySet from a Model, Manager, or QuerySet. Created to make
    get_object_or_404 and get_list_or_404 more DRY.

    Raises a ValueError if klass is not a Model, Manager, or QuerySet.
    """
    if isinstance(klass, QuerySet):
        return klass
    elif isinstance(klass, Manager):
        manager = klass
    elif isinstance(klass, ModelBase):
        manager = klass._default_manager
    else:
        if isinstance(klass, type):
            klass__name = klass.__name__
        else:
            klass__name = klass.__class__.__name__
        raise ValueError("Object is of type '%s', but must be a Django Model, "
                         "Manager, or QuerySet" % klass__name)
    return manager.all()

def get_object_or_404(klass, *args, **kwargs):
    """
    Uses get() to return an object, or raises a Http404 exception if the object
    does not exist.

    klass may be a Model, Manager, or QuerySet object. All other passed
    arguments and keyword arguments are used in the get() query.

    Note: Like with get(), an MultipleObjectsReturned will be raised if more than one
    object is found.
    """
    queryset = _get_queryset(klass)
    try:
        return queryset.get(*args, **kwargs)
    except queryset.model.DoesNotExist:
        raise Http404('No %s matches the given query.' % queryset.model._meta.object_name)

def process_rhs(self, compiler, connection):
        value = self.rhs
        if self.bilateral_transforms:
            if self.rhs_is_direct_value():
                # Do not call get_db_prep_lookup here as the value will be
                # transformed before being used for lookup
                value = QueryWrapper("%s",
                    [self.lhs.output_field.get_db_prep_value(value, connection)])
            value = self.apply_bilateral_transforms(value)
        # Due to historical reasons there are a couple of different
        # ways to produce sql here. get_compiler is likely a Query
        # instance, _as_sql QuerySet and as_sql just something with
        # as_sql. Finally the value can of course be just plain
        # Python value.
        if hasattr(value, 'get_compiler'):
            value = value.get_compiler(connection=connection)
        if hasattr(value, 'as_sql'):
            sql, params = compiler.compile(value)
            return '(' + sql + ')', params
        if hasattr(value, '_as_sql'):
            sql, params = value._as_sql(connection=connection)
            return '(' + sql + ')', params
        else:
            return self.get_db_prep_lookup(value, connection)

def _eval(self):
        results = self.aggregate(**self._params)
        try:
            values = isinstance(results, ValuesQuerySet)
        except NameError:       # django >= 1.9
            values = results.__class__ is not ModelIterable
        if isinstance(results, QuerySet) and not values:
            self._data_type = 'qs'
        elif pnd and isinstance(results, DataFrame):
            self._data_type = 'df'
        self._split_totals(results)
        self._evaluated = True

def get_list_or_404(klass, *args, **kwargs):
    """
    Uses filter() to return a list of objects, or raise a Http404 exception if
    the list is empty.

    klass may be a Model, Manager, or QuerySet object. All other passed
    arguments and keyword arguments are used in the filter() query.
    """
    queryset = _get_queryset(klass)
    obj_list = list(queryset.filter(*args, **kwargs))
    if not obj_list:
        raise Http404('No %s matches the given query.' % queryset.model._meta.object_name)
    return obj_list

def __init__(self, lhs, rhs):
        self.lhs, self.rhs = lhs, rhs
        self.rhs = self.get_prep_lookup()
        if hasattr(self.lhs, 'get_bilateral_transforms'):
            bilateral_transforms = self.lhs.get_bilateral_transforms()
        else:
            bilateral_transforms = []
        if bilateral_transforms:
            # We should warn the user as soon as possible if he is trying to apply
            # a bilateral transformation on a nested QuerySet: that won't work.
            # We need to import QuerySet here so as to avoid circular
            from django.db.models.query import QuerySet
            if isinstance(rhs, QuerySet):
                raise NotImplementedError("Bilateral transformations on nested querysets are not supported.")
        self.bilateral_transforms = bilateral_transforms

def get_queryset(self):
        """
        Get the list of items for this view.
        This must be an iterable, and may be a queryset.
        Defaults to using `self.queryset`.

        This method should always be used rather than accessing `self.queryset`
        directly, as `self.queryset` gets evaluated only once, and those results
        are cached for all subsequent requests.

        You may want to override this if you need to provide different
        querysets depending on the incoming request.

        (Eg. return a list of items that is specific to the user)
        """
        assert self.queryset is not None, (
            "'%s' should either include a `queryset` attribute, "
            "or override the `get_queryset()` method."
            % self.__class__.__name__
        )

        queryset = self.queryset
        if isinstance(queryset, QuerySet):
            # Ensure queryset is re-evaluated on each request.
            queryset = queryset.all()
        return queryset

def get_queryset(self):
        queryset = self.queryset
        if isinstance(queryset, (QuerySet, Manager)):
            # Ensure queryset is re-evaluated whenever used.
            # Note that actually a `Manager` class may also be used as the
            # queryset argument. This occurs on ModelSerializer fields,
            # as it allows us to generate a more expressive 'repr' output
            # for the field.
            # Eg: 'MyRelationship(queryset=ExampleModel.objects.all())'
            queryset = queryset.all()
        return queryset

def test_model(self):
        """The model should be an instance of Book."""
        # The replaced model should be on both the QuerySet and Query.
        self.assertIs(self.all.query._querysets[0].model,
                      self.all.query._querysets[0].query.model)

        # It's still an instance of the original model.
        first = self.all[0]
        self.assertIsInstance(first, Book)
        # But it also has a new superclass.
        self.assertIn('queryset_sequence.QuerySequenceModel',
                      map(lambda cls: cls.__module__ + '.' + cls.__name__,
                          first.__class__.__mro__))

        # Note that a bunch of meta properties get re-labeled. This is OK.
        options = first._meta
        self.assertTrue(
            options.app_label.startswith('queryset_sequence.'))
        self.assertEquals(options.model_name, 'querysequencemodel')
        self.assertEquals(options.object_name, 'QuerySequenceModel')

        # Django >= 1.9 the label attribute exists. Otherwise, cast to a string.
        object_name = 'QuerySequenceModel'
        try:
            label = options.label
        except AttributeError:
            label = str(options)
            object_name = object_name.lower()
        self.assertTrue(label.startswith('queryset_sequence'))
        self.assertTrue(label.endswith(object_name))

def test_queryset_number(self):
        """Ensure that the QuerySet number is correct on the model."""
        data = list(map(attrgetter('#'), self.all._clone()))
        self.assertEqual([0, 0, 1, 1, 1], data)

def test_queryset_number_filter(self):
        """The QuerySet number shouldn't change after filtering, etc."""
        data = list(map(attrgetter('#'), self.all.filter(**{'#': 1})))
        self.assertEqual([1, 1, 1], data)

def test_len(self):
        qss = self.all._clone()

        # Calling len() evaluates the QuerySet.
        self.assertEqual(len(qss), 5)
        self.assertIsNotNone(qss._result_cache)

        # Count should still work (and not hit the database) by using the cache.
        qss.query = None
        self.assertEqual(qss.count(), 5)

def test_iter_cache(self):
        """Ensure that iterating the QuerySet caches."""
        qss = self.all._clone()

        with self.assertNumQueries(2):
            data = [it.title for it in qss]
            self.assertEqual(data, TestIterator.EXPECTED)

        # So the second call does nothing.
        with self.assertNumQueries(0):
            data = [it.title for it in qss]
            self.assertEqual(data, TestIterator.EXPECTED)

def test_empty(self):
        """
        Ensure that filter() works when it results in an empty QuerySet.
        """
        # Filter to nothing.
        with self.assertNumQueries(0):
            qss = self.all.filter(title='')
        self.assertEqual(qss.count(), 0)
        self.assertIsInstance(qss, QuerySetSequence)

        # This should not throw an exception.
        data = list(qss)
        self.assertEqual(len(data), 0)

def test_simplify(self):
        """
        Ensure that filter() properly filters the children QuerySets and
        simplifies to a single child QuerySet when all others become empty.
        """
        # Filter to just Alice's work.
        with self.assertNumQueries(0):
            alice_qss = self.all.exclude(author=self.bob)
        self.assertEqual(alice_qss.count(), 2)
        # TODO
        # self.assertIsNone(alice_qss._result_cache)

        # Since we've now filtered down to a single QuerySet, we shouldn't be a
        # QuerySetSequence any longer.
        self.assertIsInstance(alice_qss, QuerySet)

def test_empty(self):
        """
        Ensure that filter() works when it results in an empty QuerySet.
        """
        # Filter to nothing.
        with self.assertNumQueries(0):
            qss = self.all.exclude(author__in=[self.alice, self.bob])
        self.assertEqual(qss.count(), 0)
        self.assertIsInstance(qss, QuerySetSequence)

        # This should not throw an exception.
        data = list(qss)
        self.assertEqual(len(data), 0)

def test_order_by_queryset(self):
        """Ensure we can order by QuerySet and then other fields."""
        # Order by title, but don't interleave each QuerySet.
        with self.assertNumQueries(0):
            qss = self.all.order_by('#', 'title')
        self.assertEqual(qss.query.order_by, ['#', 'title'])
        self.assertEqual(
            qss.query._querysets[0].query.order_by,
            ('title',) if DJANGO_VERSION >= (2,) else ['title'],
        )

        # Ensure that _ordered_iterator isn't called.
        with patch('queryset_sequence.QuerySequence._ordered_iterator',
                   side_effect=AssertionError('_ordered_iterator should not be called')):
            # Check the titles are properly ordered.
            data = [it.title for it in qss]
            expected = [
                # First the Books, in order.
                'Biography',
                'Fiction',
                # Then the Articles, in order.
                'Alice in Django-land',
                'Django Rocks',
                'Some Article',
            ]
            self.assertEqual(data, expected)

def test_single_element(self):
        """Single element."""
        qss = self.all._clone()
        result = qss[0]
        self.assertEqual(result.title, 'Fiction')
        self.assertIsInstance(result, Book)
        # qss never gets evaluated since the underlying QuerySet is used.
        self.assertIsNone(qss._result_cache)

def test_one_QuerySet(self):
        """Test slicing only from one QuerySet."""
        qss = self.all._clone()
        result = qss[0:2]
        self.assertIsInstance(result, QuerySet)
        # qss never gets evaluated since the underlying QuerySet is used.
        self.assertIsNone(qss._result_cache)
        # Check the data.
        for element in result:
            self.assertIsInstance(element, Book)