Get queryset (#528)

* first attempt at adding get_queryset

* add queryset_resolver to DjangoConnectionField and fix test failures

* cleanup get_queryset API to match proposal as close as possible

* pep8 fix: W293

* document get_queryset usage

* add test for when get_queryset is defined on DjangoObjectType

Author: zbyte64

docs/authorization.rst

@@ -96,6 +96,29 @@ schema is simple.
 
     result = schema.execute(query, context_value=request)
 
+
+Global Filtering
+----------------
+
+If you are using ``DjangoObjectType`` you can define a custom `get_queryset`.
+
+.. code:: python
+
+    from graphene import relay
+    from graphene_django.types import DjangoObjectType
+    from .models import Post
+
+    class PostNode(DjangoObjectType):
+        class Meta:
+            model = Post
+
+        @classmethod
+        def get_queryset(cls, queryset, info):
+            if info.context.user.is_anonymous:
+                return queryset.filter(published=True)
+            return queryset
+
+
 Filtering ID-based Node Access
 ------------------------------
 

graphene_django/fields.py

@@ -67,6 +67,10 @@ def get_manager(self):
         else:
             return self.model._default_manager
 
+    @classmethod
+    def resolve_queryset(cls, connection, queryset, info, args):
+        return connection._meta.node.get_queryset(queryset, info)
+
     @classmethod
     def merge_querysets(cls, default_queryset, queryset):
         if default_queryset.query.distinct and not queryset.query.distinct:
@@ -135,7 +139,8 @@ def connection_resolver(
                 args["last"] = min(last, max_limit)
 
         iterable = resolver(root, info, **args)
-        on_resolve = partial(cls.resolve_connection, connection, default_manager, args)
+        queryset = cls.resolve_queryset(connection, default_manager, info, args)
+        on_resolve = partial(cls.resolve_connection, connection, queryset, args)
 
         if Promise.is_thenable(iterable):
             return Promise.resolve(iterable).then(on_resolve)

graphene_django/tests/test_query.py

@@ -1007,3 +1007,47 @@ class Query(graphene.ObjectType):
 
     result = schema.execute(query)
     assert result.errors
+
+
+def test_should_resolve_get_queryset_connectionfields():
+    reporter_1 = Reporter.objects.create(
+        first_name="John", last_name="Doe", email="johndoe@example.com", a_choice=1
+    )
+    reporter_2 = CNNReporter.objects.create(
+        first_name="Some",
+        last_name="Guy",
+        email="someguy@cnn.com",
+        a_choice=1,
+        reporter_type=2,  # set this guy to be CNN
+    )
+
+    class ReporterType(DjangoObjectType):
+        class Meta:
+            model = Reporter
+            interfaces = (Node,)
+
+        @classmethod
+        def get_queryset(cls, queryset, info):
+            return queryset.filter(reporter_type=2)
+
+    class Query(graphene.ObjectType):
+        all_reporters = DjangoConnectionField(ReporterType)
+
+    schema = graphene.Schema(query=Query)
+    query = """
+        query ReporterPromiseConnectionQuery {
+            allReporters(first: 1) {
+                edges {
+                    node {
+                        id
+                    }
+                }
+            }
+        }
+    """
+
+    expected = {"allReporters": {"edges": [{"node": {"id": "UmVwb3J0ZXJUeXBlOjI="}}]}}
+
+    result = schema.execute(query)
+    assert not result.errors
+    assert result.data == expected

graphene_django/types.py

@@ -133,9 +133,14 @@ def is_type_of(cls, root, info):
         model = root._meta.model._meta.concrete_model
         return model == cls._meta.model
 
+    @classmethod
+    def get_queryset(cls, queryset, info):
+        return queryset
+
     @classmethod
     def get_node(cls, info, id):
+        queryset = cls.get_queryset(cls._meta.model.objects, info)
         try:
-            return cls._meta.model.objects.get(pk=id)
+            return queryset.get(pk=id)
         except cls._meta.model.DoesNotExist:
             return None