feat: add node query object resolving helpers

ariadne/contrib/relay/__init__.py

@@ -3,15 +3,18 @@
 )
 from ariadne.contrib.relay.connection import RelayConnection
 from ariadne.contrib.relay.objects import (
+    RelayNodeInterfaceType,
     RelayObjectType,
     RelayQueryType,
 )
-from ariadne.contrib.relay.types import ConnectionResolver
+from ariadne.contrib.relay.types import ConnectionResolver, GlobalIDTuple
 
 __all__ = [
     "ConnectionArguments",
+    "RelayNodeInterfaceType",
     "RelayConnection",
     "RelayObjectType",
     "RelayQueryType",
     "ConnectionResolver",
+    "GlobalIDTuple",
 ]

ariadne/contrib/relay/objects.py

@@ -1,4 +1,6 @@
+from base64 import b64decode
 from inspect import iscoroutinefunction
+from typing import Dict, Optional, Tuple
 
 from graphql.pyutils import is_awaitable
 
@@ -7,20 +9,78 @@
     ConnectionArguments,
     ConnectionArgumentsTypeUnion,
 )
-from ariadne.contrib.relay.types import ConnectionResolver
+from ariadne.contrib.relay.types import (
+    ConnectionResolver,
+    GlobalIDDecoder,
+    GlobalIDTuple,
+)
+from ariadne.types import Resolver
+
+
+def decode_global_id(kwargs) -> GlobalIDTuple:
+    return GlobalIDTuple(*b64decode(kwargs["id"]).decode().split(":"))
+
+
+class RelayNodeInterfaceType(InterfaceType):
+    def __init__(
+        self,
+        type_resolver: Optional[Resolver] = None,
+        global_id_decoder: Optional[GlobalIDDecoder] = decode_global_id,
+    ) -> None:
+        super().__init__("Node", type_resolver)
+        self._object_resolvers: Dict[str, Resolver] = {}
+        self.global_id_decoder = global_id_decoder
+
+    def node_resolver(self, name: str):
+        def decorator(resolver):
+            self.set_node_resolver(name, resolver)
+            return resolver
+
+        return decorator
+
+    def set_node_resolver(self, name: str, resolver):
+        self._object_resolvers[name] = resolver
+
+    def get_node_resolver(self, type_name: str):
+        try:
+            return self._object_resolvers[type_name]
+        except KeyError as exc:
+            raise ValueError(f"No object resolver for type {type_name}") from exc
 
 
 class RelayQueryType(QueryType):
     def __init__(
-        self, *args, node_type_resolver=None, node_field_resolver=None, **kwargs
+        self,
+        *args,
+        node=None,
+        node_field_resolver=None,
+        **kwargs,
     ) -> None:
         super().__init__(*args, **kwargs)
-        self.node = InterfaceType("Node", node_type_resolver)
+        if not node:
+            node = RelayNodeInterfaceType()
+        self.node = node
+        if not node_field_resolver:
+            node_field_resolver = self.default_resolve_node
         self.set_field("node", node_field_resolver)
 
     @property
-    def bindables(self):
-        return [self, self.node]
+    def bindables(self) -> Tuple["RelayQueryType", "RelayNodeInterfaceType"]:
+        return (self, self.node)
+
+    def default_resolve_node(self, obj, info, *args, **kwargs):
+        type_name, _ = self.node.global_id_decoder(kwargs)
+        resolver = self.node.get_node_resolver(type_name)
+        if iscoroutinefunction(resolver):
+
+            async def async_my_extension():
+                result = await resolver(obj, info, *args, **kwargs)
+                if is_awaitable(result):
+                    result = await result
+                return result
+
+            return async_my_extension()
+        return resolver(obj, info, *args, **kwargs)
 
 
 class RelayObjectType(ObjectType):

ariadne/contrib/relay/types.py

@@ -1,7 +1,10 @@
-from typing import Callable
+from collections import namedtuple
+from typing import Any, Callable, Dict
 
 from typing_extensions import TypeVar
 
 from ariadne.contrib.relay.connection import RelayConnection
 
 ConnectionResolver = TypeVar("ConnectionResolver", bound=Callable[..., RelayConnection])
+GlobalIDTuple = namedtuple("GlobalIDTuple", ["type", "id"])
+GlobalIDDecoder = Callable[[Dict[str, Any]], GlobalIDTuple]