Improved Node get_node_from_global_id

This introduces a breaking changes for custom Nodes implementations

Author: syrusakbary

docs/relay/nodes.rst

@@ -55,8 +55,13 @@ Example of a custom node:
             return '{}:{}'.format(type, id)
 
         @staticmethod
-        def get_node_from_global_id(global_id, context, info):
+        def get_node_from_global_id(global_id, context, info, only_type=None):
             type, id = global_id.split(':')
+            if only_node:
+                # We assure that the node type that we want to retrieve
+                # is the same that was indicated in the field type
+                assert type == only_node._meta.name, 'Received not compatible node.'
+
             if type == 'User':
                 return get_user(id)
             elif type == 'Photo':
@@ -66,6 +71,17 @@ Example of a custom node:
 The ``get_node_from_global_id`` method will be called when ``CustomNode.Field`` is resolved.
 
 
+Accessing node types
+--------------------
+
+If we want to retrieve node instances from a ``global_id`` (scalar that identifies an instance by it's type name and id),
+we can simply do ``Node.get_node_from_global_id(global_id, contet, info)``.
+
+In the case we want to restric the instnance retrieval to an specific type, we can do:
+``Node.get_node_from_global_id(global_id, contet, info, only_type=Ship)``. This will raise an error
+if the global_id doesn't correspond to a Ship type.
+
+
 Node Root field
 ---------------
 

graphene/relay/node.py

@@ -63,12 +63,16 @@ class NodeField(Field):
     def __init__(self, node, type=False, deprecation_reason=None,
                  name=None, **kwargs):
         assert issubclass(node, Node), 'NodeField can only operate in Nodes'
-        type = type or node
+        self.node_type = node
+
+        # If we don's specify a type, the field type will be the node interface
+        field_type = type or node
+
         super(NodeField, self).__init__(
-            type,
+            field_type,
             description='The ID of the object',
             id=ID(required=True),
-            resolver=node.node_resolver
+            resolver=partial(node.node_resolver, only_type=type)
         )
 
 
@@ -80,18 +84,26 @@ def Field(cls, *args, **kwargs):  # noqa: N802
         return NodeField(cls, *args, **kwargs)
 
     @classmethod
-    def node_resolver(cls, root, args, context, info):
-        return cls.get_node_from_global_id(args.get('id'), context, info)
+    def node_resolver(cls, root, args, context, info, only_type=None):
+        return cls.get_node_from_global_id(args.get('id'), context, info, only_type)
 
     @classmethod
-    def get_node_from_global_id(cls, global_id, context, info):
+    def get_node_from_global_id(cls, global_id, context, info, only_type=None):
         try:
             _type, _id = cls.from_global_id(global_id)
             graphene_type = info.schema.get_type(_type).graphene_type
-            # We make sure the ObjectType implements the "Node" interface
-            assert cls in graphene_type._meta.interfaces
         except:
             return None
+
+        if only_type:
+            assert graphene_type == only_type, (
+                'Must receive an {} id.'
+            ).format(graphene_type._meta.name)
+
+        # We make sure the ObjectType implements the "Node" interface
+        if cls not in graphene_type._meta.interfaces:
+            return None
+
         get_node = getattr(graphene_type, 'get_node', None)
         if get_node:
             return get_node(_id, context, info)

graphene/relay/tests/test_node.py

@@ -44,6 +44,7 @@ def get_node(id, *_):
 class RootQuery(ObjectType):
     first = String()
     node = Node.Field()
+    only_node = Node.Field(MyNode)
 
 schema = Schema(query=RootQuery, types=[MyNode, MyOtherNode])
 
@@ -63,7 +64,7 @@ def test_node_get_connection_dont_duplicate():
 
 def test_node_query():
     executed = schema.execute(
-        '{ node(id:"%s") { ... on MyNode { name } } }' % to_global_id("MyNode", 1)
+        '{ node(id:"%s") { ... on MyNode { name } } }' % Node.to_global_id("MyNode", 1)
     )
     assert not executed.errors
     assert executed.data == {'node': {'name': '1'}}
@@ -86,6 +87,35 @@ def test_node_query_incorrect_id():
     assert executed.data == {'node': None}
 
 
+def test_node_field():
+    node_field = Node.Field()
+    assert node_field.type == Node
+    assert node_field.node_type == Node
+
+
+def test_node_field_custom():
+    node_field = Node.Field(MyNode)
+    assert node_field.type == MyNode
+    assert node_field.node_type == Node
+
+
+def test_node_field_only_type():
+    executed = schema.execute(
+        '{ onlyNode(id:"%s") { __typename, name } } ' % Node.to_global_id("MyNode", 1)
+    )
+    assert not executed.errors
+    assert executed.data == {'onlyNode': {'__typename': 'MyNode', 'name': '1'}}
+
+
+def test_node_field_only_type_wrong():
+    executed = schema.execute(
+        '{ onlyNode(id:"%s") { __typename, name } } ' % Node.to_global_id("MyOtherNode", 1)
+    )
+    assert len(executed.errors) == 1
+    assert str(executed.errors[0]) == 'Must receive an MyOtherNode id.'
+    assert executed.data == { 'onlyNode': None }
+
+
 def test_str_schema():
     assert str(schema) == """
 schema {
@@ -111,5 +141,6 @@ def test_str_schema():
 type RootQuery {
   first: String
   node(id: ID!): Node
+  onlyNode(id: ID!): MyNode
 }
 """.lstrip()

graphene/relay/tests/test_node_custom.py

@@ -15,7 +15,7 @@ def to_global_id(type, id):
         return id
 
     @staticmethod
-    def get_node_from_global_id(id, context, info):
+    def get_node_from_global_id(id, context, info, only_type=None):
         assert info.schema == schema
         if id in user_data:
             return user_data.get(id)