GraphQL Relay contrib module

[pkucmus]

Today we're starting the work to give Ariadne a bunch of tooling around GraphQL Relay.

• GraphQL Relay Specification
• Another GraphQL Relay Spec

I'd like to stick to the simplicity that Ariadne follows and NOT make a fully blown Relay package. Rather than that I'd like a set of tools that would make it easier and more reproducible to make Relay style GraphQL APIs with Ariadne. This would mean an ariadne.contrib.relay module that would deliver a RelayObjectType enabled to resolve a schema that applies to the Relay spec.

I'll shortly paste how I imagine the module to be used shortly, until then, any thoughts so far?

[pkucmus]

Here's the code (in case of a bigger discussion let's call this approach no.1) - and by code I mean - a rough idea of how one would use it (schema from here):

from ariadne import QueryType, make_executable_schema
from ariadne.asgi import GraphQL
from ariadne.contrib.relay import RelayObjectType, Connection, bindables

...

query = QueryType()

@query.field("rebels")
async def resolve_rebels(*_):
    return FACTIONS[0]

@query.field("empire")
async def resolve_empire(*_):
    return FACTIONS[1]

faction = RelayObjectType("Faction")

@faction.connection("ships")
async def resolve_ships(
	faction,
	info,
    first: int | None = None,
    after: str | None = None,
    last: int | None = None,
    before: str | None = None,
):
	total = await fetch_all_count()
	ships_slice = await fetch_slice(first, after)

    return Connection(
        ships_slice,
        {
            "total": total,
            "hasNextPage": ...,
            "hasPreviousPage": ...,
            "startCursor": ...,
            "endCursor": ...,
        },
    )

app = GraphQL(
    make_executable_schema(schema, query, faction, *bindables),
    debug=True,
)

The special RelayObjectType.connection decorator would expect a Connection (or a different name, not to confuse the decorator with this result type) to be returned. Behind the scene the connection decorator would construct the edges and pageInfo types.

The node query would be provided or documented, depending on how much utility we could deliver here outside of what is already possible.

This is to illustrate that a lot would remain in the hands of the developer, like the need to generate the slice - the performance or lack thereof is up to the developer implementing the server and very much dependent on the resource (SQL DB, other GraphQL server or otherwise).

Feedback welcome.

[pkucmus]

Made the PR: #1214

[rafalp]

Here's my idea, inspired by what I've did in past:

class ConnectionEdge:
    def __init__(
        self,
        query,
        first: int | None=None,
        last: int | None=None,
        after: int | None=None,
        before: str | None=None,
    ):
        self.query = query

        assert not all((first, last))
        assert not all((before, after))

        # unpack Relay ID here

        self.first = first
        self.last = last
        self.after = after
        self.before = before

        # eager caches for look ahead/look back
        self._has_next_page = None
        self._has_previous_page = None

    def get_nodes(self):  # this can be cached per instance after first call
        if self.after:
            if self.first:
                results = self.query.filter(id__gt=self.after).order_by("id")[self.first + 1:]
            else:
                results = self.query.filter(id__gt=self.after).order_by("-id")[self.last + 1:]
        elif self.before:
            if self.first:
                results = self.query.filter(id__lt=self.after).order_by("-id")[self.first + 1:]
            else:
                results = self.query.filter(id__lt=self.after).order_by("id")[self.last + 1:]
        else:
            if self.first:
                results = self.query.order_by("id")[self.first + 1:]
            else:
                results = self.query.order_by("-id")[self.last + 1:]
        
        # we can skip next page lookup using extra result of query
        if self.after:
            if self.first and len(results) > self.first:
                self._has_next_page = True
                results = results[:1]
            if self.last and len(results) > self.last:
                self._has_previous_page = True
                results = results[1:]
        if self.before:
            if self.first and len(results) > self.first:
                self._has_previous_page = True
                results = results[1:]
            if self.last and len(results) > self.last:
                self._has_next_page = True
                results = results[:1]
        else:
            ...

        return results  # wrap results into nodes?

    def get_page_info(self):
        return ConnectionPageInfo(
            query=self.query,
            first=self.first,
            last=self.last,
            after=self.after,
            before=self.before,
            has_next_page=self._has_next_page,
            has_previous_page=self._has_previous_page,
            .... # pass extra options like first and last cursor
        )


class ConnectionPageInfo:
    def __init__(
        self,
        query,
        first: int | None=None,
        last: int | None=None,
        after: int | None=None,
        before: str | None=None,
        has_next_page: bool | None=None,
        has_previous_page: bool | None=None,
    ):
        # store params
        self.query = query

    def resolve_total(self):
        return self.query.count()


faction = RelayObjectType("Faction", ConnectionEdge)


@faction.connection("ships")
def resolve_connection(
    obj,
    info,
    first: int | None = None,
    last: int | None = None,
    after: str | None = None,
    before: str | None = None,
):
    return query # Decorator packs Query and first/last/after/before into ConnectionEdge

My idea is to be as lazy as possible with resolving properties from relay connection objects, and leave resolvers only with filtering queryset if needed.

[pkucmus]

Thanks Rafał. This example being tied to Django, does this approach assume one would need to create their own get_nodes, based on what is returned by the connection resolver?

[rafalp]

You will need to create ConnectionEdge that knows how to paginate and count your data source, yes.

[reallistic]

I'm looking forward to having relay support built-in to Ariadne!

Would be awesome to include some tools for global object identification for servers with multiple types coming from different datasources.

Consider the following:

from ariadne import QueryType

query = QueryType()

@query.field("node")
def resolve_global_object(_, info, global_id):
    node_type, obj_id = b63_decode(global_id).split(":", 1)
    if node_type == "Faction":
        return factions_backend.find_first(obj_id)
    elif node_type == "Ship":
       return ships_backend.find_first(obj_id)
    return None

This can become cumbersome with many objects and could use a simple helper. Something like:

from ariadne import QueryType
from ariadne.contrib.relay import (
    RelayQueryType,
    RelayObjectType,
    from_global_id,
    to_global_id,
    resolve_global_node
)

query = RelayQueryType()

@query.node
def resolve_global_object(_, info, global_id):
    return resolve_global_node(info, global_id, from_global_id=from_global_id)


faction = RelayObjectType("Faction")

@faction.node
def resolve_instance(info, obj_id):
    return factions_backend.find_first(obj_id)

The resolve_global_node function and RelayObjectType.node function could look something like this:

def resolve_global_node(info, global_id, from_global_id=default_from_global_id):
    type_name, obj_id = from_global_id(global_id)
    gql_obj = info.schema.type_map.get(type_name)
    if not gql_obj:
        return None
    resolver = gql_obj.extensions.get("NODE_RESOLVER")
    return resolver(obj_id)


class RelayObjectType(ObjectType):
    def node(self, f):
        self._node_resolver = f
        return f

    def bind_resolvers_to_graphql_type(self, graphql_type, ...):
        super().bind_resolvers_to_graphql_type(graphql_type, ...)
        graphql_type.extensions["NODE_RESOLVER"] = self._node_resolver