graphene relay's `get_node` is not `async`

[wapiflapi]

I did not figure out how I could have the get_node classmethod of an ObectType be async. It would be nice if it was because otherwise it can not take full advantage of dataloaders.

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[wapiflapi]

Am I the only person having this issue? Is there a simple workaround?

[art1415926535]

It works as an asynchronous function in my code:

class MyObject(ObjectType):
    @classmethod
    async def get_node(cls, info, id):
        return await ...

[wapiflapi]

Thank you for your reply! That's useful and I'm happy we can engage the conversation about this issue.

The code calling get_node in the relay implementation isn't async and doesn't use await, see https://github.com/graphql-python/graphene/blob/60a9609b9a3b97e13a25ea0c1330ff35e2332156/graphene/relay/node.py#L121

More context:

class Node(AbstractNode):
    """An object with an ID"""

    @classmethod
    def get_node_from_global_id(cls, info, global_id, only_type=None):
        # some code omitted for clarity
        get_node = getattr(graphene_type, "get_node", None)
        if get_node:
            return get_node(info, _id)

So it is certainly possible to make our own async get_node() but then it wouldn't work with node_resolver / get_node_from_global_id etc.

Or am I miss understanding all of this? TBH I'm not working on a graphene project right now, so I might be miss-remembering exactly the issue I was having.

[art1415926535]

Nodes are almost never used (node field in query). I say this from my own experience.

However we can update Node interface and NodeField. I marked updated lines that are different from graphene source code.

from graphene import Field, ID, relay


class AsyncNodeField(Field):
    def __init__(self, node, type=False, deprecation_reason=None, name=None, **kwargs):
        assert issubclass(node, AsyncNode), "NodeField can only operate in Nodes"  # <- UPDATED
        self.node_type = node
        self.field_type = type

        super().__init__(    # <- UPDATED
            # If we don's specify a type, the field type will be the node
            # interface
            type or node,
            description="The ID of the object",
            id=ID(required=True),
            )

    def get_resolver(self, parent_resolver):
        return partial(self.node_type.node_resolver, get_type(self.field_type))


class AsyncNode(relay.Node):
    @classmethod
    def Field(cls, *args, **kwargs):  # noqa: N802
        return AsyncNodeField(cls, *args, **kwargs)  # <- UPDATED

    @classmethod
    def resolve_type(cls, instance, info):
        type_ = super().resolve_type(instance, info)
        if type_ is not None:
            return type_

        #  v UPDATED
        if isinstance(instance, models.User):
            return UserNode
        #  ^ UPDATED

    @classmethod
    async def get_node_from_global_id(cls, info, global_id, only_type=None):  # <- UPDATED
        try:
            _type, _id = cls.from_global_id(global_id)
            graphene_type = info.schema.get_type(_type).graphene_type
        except Exception:
            return None

        if only_type:
            assert graphene_type == only_type, ("Must receive a {} id.").format(
                only_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 await get_node(info, _id)  # <- UPDATED

And usage:

class UserNode(...):
    class Meta:
        interfaces = (AsyncNode, )


class Query(ObjectType):
    node = AsyncNode.Field()

Maybe there is an easier way... but this one works.

[wapiflapi]

That's a lot of work for something that I feel should work out of the box.

An example of something where this bothered me:

class Query(graphene.ObjectType):

    node = papi.relay.Node.Field()
    nodes = graphene.List(
        papi.relay.Node,
        required=True,
        ids=graphene.List(graphene.ID),
    )

    async def resolve_nodes(self, info, ids):
        """Resolve multiple nodes by ID at the same time."""

        # We can not do this in parallel because graphene.relay's
        # get_node is not async. Which prevents any possible
        # dataloaders from doing their jobs.

        return [
            graphene.relay.Node.get_node_from_global_id(info, node_id) for node_id in ids
        ]

I wish I could await papi.relay.Node.get_node_from_global_id() in that code, but that's not possible because down the line it requires a lot of rewriting of code and that's not very maintainable and is really something that should be handled by graphene.

[art1415926535]

I think the root of the problem is that graphene tries to work with both synchronous and asynchronous code.

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[ayyoubelamrani4]

@erikwrede should this still be labelled as wontfix?

[erikwrede]

@ayyoubelamrani4 thanks, removed the label