Relay Guide

What is Relay?

The relay spec defines some interfaces that GraphQL servers can follow to allow clients to interact with them in a more efficient way. The spec makes two core assumptions about a GraphQL server:

1. It provides a mechanism for refetching an object
2. It provides a description of how to page through connections.

You can read more about the relay spec here.

Relay implementation example

Suppose we have the following type:

@strawberry.type
class Fruit:
    name: str
    weight: str

We want it to have a globally unique ID, a way to retrieve a paginated results list of it and a way to refetch if if necessary. For that, we need to inherit it from the Node interface, annotate its attribute that will be used for the ID field with relay.NodeID and implement its resolve_nodes abstract method.

import strawberry
from strawberry import relay


@strawberry.type
class Fruit(relay.Node):
    code: relay.NodeID[int]
    name: str
    weight: float

    @classmethod
    def resolve_nodes(
        cls,
        *,
        info: strawberry.Info,
        node_ids: Iterable[str],
        required: bool = False,
    ):
        return [
            all_fruits[int(nid)] if required else all_fruits.get(nid)
            for nid in node_ids
        ]


# In this example, assume we have a dict mapping the fruits code to the Fruit
# object itself
all_fruits: Dict[int, Fruit]

Explaining what we did here:

• We annotated code using relay.NodeID[int]. This makes code a Private type, which will not be exposed to the GraphQL API, and also tells the Node interface that it should use its value to generate its id: ID! for the Fruit type.

• We also implemented the resolve_nodes abstract method. This method is responsible for retrieving the Fruit instances given its id. Because code is our id, node_ids will be a list of codes as a string.

Now we can expose it in the schema for retrieval and pagination like:

@strawberry.type
class Query:
    node: relay.Node = relay.node()

    @relay.connection(relay.ListConnection[Fruit])
    def fruits(self) -> Iterable[Fruit]:
        # This can be a database query, a generator, an async generator, etc
        return all_fruits.values()

This will generate a schema like this:

interface Node {
  id: ID!
}

type PageInfo {
  hasNextPage: Boolean!
  hasPreviousPage: Boolean!
  startCursor: String
  endCursor: String
}

type Fruit implements Node {
  id: ID!
  name: String!
  weight: Float!
}

type FruitEdge {
  cursor: String!
  node: Fruit!
}

type FruitConnection {
  pageInfo: PageInfo!
  edges: [FruitEdge!]!
}

type Query {
  node(id: ID!): Node!
  fruits(
    before: String = null
    after: String = null
    first: Int = null
    last: Int = null
  ): FruitConnection!
}

With only that we have a way to query node to retrieve any Node implemented type in our schema (which includes our Fruit type), and also a way to retrieve a list of fruits with pagination.

For example, to retrieve a single fruit given its unique ID:

query {
  node(id: "<some id>") {
    id
    ... on Fruit {
      name
      weight
    }
  }
}

Or to retrieve the first 10 fruits available:

query {
  fruits(first: 10) {
    pageInfo {
      firstCursor
      endCursor
      hasNextPage
      hasPreviousPage
    }
    edges {
      # node here is the Fruit type
      node {
        id
        name
        weight
      }
    }
  }
}

The connection resolver for relay.ListConnection should return one of those:

• List[<NodeType>]
• Iterator[<NodeType>]
• Iterable[<NodeType>]
• AsyncIterator[<NodeType>]
• AsyncIterable[<NodeType>]
• Generator[<NodeType>, Any, Any]
• AsyncGenerator[<NodeType>, Any]

The node field

As demonstrated above, the Node field can be used to retrieve/refetch any object in the schema that implements the Node interface.

It can be defined in the Query objects in 4 ways:

• node: Node: This will define a field that accepts a ID! and returns a Node instance. This is the most basic way to define it.
• node: Optional[Node]: The same as Node, but if the given object doesn't exist, it will return null.
• node: List[Node]: This will define a field that accepts [ID!]! and returns a list of Node instances. They can even be from different types.
• node: List[Optional[Node]]: The same as List[Node], but the returned list can contain null values if the given objects don't exist.

Max results for connections

The implementation of relay.ListConnection will limit the number of results to the relay_max_results configuration in the schema's config (which defaults to 100).

That can also be configured on a per-field basis by passing max_results to the @connection decorator. For example:

@strawerry.type
class Query:
    fruits: ListConnection[Fruit] = relay.connection(max_results=10_000)

Custom connection pagination

The default relay.Connection class doesn't implement any pagination logic, and should be used as a base class to implement your own pagination logic. All you need to do is implement the resolve_connection classmethod.

The integration provides relay.ListConnection, which implements a limit/offset approach to paginate the results. This is a basic approach and might be enough for most use cases.

Now, suppose we want to implement a custom cursor-based pagination for our previous example. We can do something like this:

import strawberry
from strawberry import relay


@strawberry.type
class FruitCustomPaginationConnection(relay.Connection[Fruit]):
    @classmethod
    def resolve_connection(
        cls,
        nodes: Iterable[Fruit],
        *,
        info: Optional[Info] = None,
        before: Optional[str] = None,
        after: Optional[str] = None,
        first: Optional[int] = None,
        last: Optional[int] = None,
    ):
        # NOTE: This is a showcase implementation and is far from
        # being optimal performance wise
        edges_mapping = {
            relay.to_base64("fruit_name", n.name): relay.Edge(
                node=n,
                cursor=relay.to_base64("fruit_name", n.name),
            )
            for n in sorted(nodes, key=lambda f: f.name)
        }
        edges = list(edges_mapping.values())
        first_edge = edges[0] if edges else None
        last_edge = edges[-1] if edges else None

        if after is not None:
            after_edge_idx = edges.index(edges_mapping[after])
            edges = [e for e in edges if edges.index(e) > after_edge_idx]

        if before is not None:
            before_edge_idx = edges.index(edges_mapping[before])
            edges = [e for e in edges if edges.index(e) < before_edge_idx]

        if first is not None:
            edges = edges[:first]

        if last is not None:
            edges = edges[-last:]

        return cls(
            edges=edges,
            page_info=strawberry.relay.PageInfo(
                start_cursor=edges[0].cursor if edges else None,
                end_cursor=edges[-1].cursor if edges else None,
                has_previous_page=(
                    first_edge is not None and bool(edges) and edges[0] != first_edge
                ),
                has_next_page=(
                    last_edge is not None and bool(edges) and edges[-1] != last_edge
                ),
            ),
        )


@strawberry.type
class Query:
    @relay.connection(FruitCustomPaginationConnection)
    def fruits(self) -> Iterable[Fruit]:
        # This can be a database query, a generator, an async generator, etc
        return all_fruits.values()

In the example above we specialized the FruitCustomPaginationConnection by inheriting it from relay.Connection[Fruit]. We could still keep it generic by inheriting it from relay.Connection[relay.NodeType] and then specialize it when defining the field, making it possible to use our custom pagination logic with more than one type.

Custom connection arguments

By default the connection will automatically insert some arguments for it to be able to paginate the results. Those are:

• before: Returns the items in the list that come before the specified cursor
• after: Returns the items in the list that come after the " "specified cursor
• first: Returns the first n items from the list
• last: Returns the items in the list that come after the " "specified cursor

You can still define extra arguments to be used by your own resolver or custom pagination logic. For example, suppose we want to return the pagination of all fruits whose name starts with a given string. We could do that like this:

@strawberry.type
class Query:
    @relay.connection(relay.ListConnection[Fruit])
    def fruits_with_filter(
        self,
        info: strawberry.Info,
        name_endswith: str,
    ) -> Iterable[Fruit]:
        for f in fruits.values():
            if f.name.endswith(name_endswith):
                yield f

This will generate a schema like this:

type Query {
  fruitsWithFilter(
    nameEndswith: String!
    before: String = null
    after: String = null
    first: Int = null
    last: Int = null
  ): FruitConnection!
}

Convert the node to its proper type when resolving the connection

The connection expects that the resolver will return a list of objects that is a subclass of its NodeType. But there may be situations where you are resolving something that needs to be converted to the proper type, like an ORM model.

In this case you can subclass the relay.Connection/relay.ListConnection and provide a custom resolve_node method to it, which by default returns the node as is. For example:

import strawberry
from strawberry import relay

from db import models


@strawberry.type
class Fruit(relay.Node):
    code: relay.NodeID[int]
    name: str
    weight: float


@strawberry.type
class FruitDBConnection(relay.ListConnection[Fruit]):
    @classmethod
    def resolve_node(cls, node: FruitDB, *, info: strawberry.Info, **kwargs) -> Fruit:
        return Fruit(
            code=node.code,
            name=node.name,
            weight=node.weight,
        )


@strawberry.type
class Query:
    @relay.connection(FruitDBConnection)
    def fruits_with_filter(
        self,
        info: strawberry.Info,
        name_endswith: str,
    ) -> Iterable[models.Fruit]:
        return models.Fruit.objects.filter(name__endswith=name_endswith)

The main advantage of this approach instead of converting it inside the custom resolver is that the Connection will paginate the QuerySet first, which in case of django will make sure that only the paginated results are fetched from the database. After that, the resolve_node function will be called for each result to retrieve the correct object for it.

We used django for this example, but the same applies to any other other similar use case, like SQLAlchemy, etc.

The GlobalID type

The GlobalID type is a special object that contains all the info necessary to identify and retrieve a given object that implements the Node interface.

It can for example be useful in a mutation, to receive an object and retrieve it in its resolver. For example:

@strawberry.type
class Mutation:
    @strawberry.mutation
    async def update_fruit_weight(
        self,
        info: strawberry.Info,
        id: relay.GlobalID,
        weight: float,
    ) -> Fruit:
        # resolve_node will return an awaitable that returns the Fruit object
        fruit = await id.resolve_node(info, ensure_type=Fruit)
        fruit.weight = weight
        return fruit
 
    @strawberry.mutation
    def update_fruit_weight_sync(
        self,
        info: strawberry.Info,
        id: relay.GlobalID,
        weight: float,
    ) -> Fruit:
        # resolve_node will return the Fruit object
        fruit = id.resolve_node_sync(info, ensure_type=Fruit)
        fruit.weight = weight
        return fruit

In the example above, you can also access the type name directly with id.type_name, the raw node ID with id.id, or even resolve the type itself with id.resolve_type(info).

Edit on Github