Relay

apischema provides some facilities to implement a GraphQL server following Relay GraphQL server specification. They are included in the module apischema.graphql.relay.

Note

These facilities are independent of each others — you could keep only the mutations part and use your own identification and connection system for example.

(Global) Object Identification

apischema defines a generic relay.Node[Id] interface which can be used which can be used as base class of all identified resources. This class contains a unique generic field of type Id, which will be automatically converted into an ID! in the schema. The Id type chosen has to be serializable into a string-convertible value (it can register conversions if needed).

Each node has to implement the classmethod get_by_id(cls: type[T], id: Id, info: graphql.GraphQLResolveInfo=None) -> T.

All nodes defined can be retrieved using relay.nodes, while the node query is defined as relay.node. relay.nodes() can be passed to graphql_schema types parameter in order to add them in the schema even if they don't appear in any resolvers.

from dataclasses import dataclass
from uuid import UUID

import graphql
from graphql.utilities import print_schema

from apischema.graphql import graphql_schema, relay


@dataclass
class Ship(relay.Node[UUID]):  # Let's use an UUID for Ship id
    name: str

    @classmethod
    async def get_by_id(cls, id: UUID, info: graphql.GraphQLResolveInfo = None):
        ...


@dataclass
class Faction(relay.Node[int]):  # Nodes can have different id types
    name: str

    @classmethod
    def get_by_id(cls, id: int, info: graphql.GraphQLResolveInfo = None) -> "Faction":
        ...


schema = graphql_schema(query=[relay.node], types=relay.nodes())
schema_str = """\
type Ship implements Node {
  id: ID!
  name: String!
}

interface Node {
  id: ID!
}

type Faction implements Node {
  id: ID!
  name: String!
}

type Query {
  node(id: ID!): Node!
}"""
assert print_schema(schema) == schema_str

Warning

For now, even if its result is not used, relay.nodes must be called before generating the schema.

Global ID

apischema defines a relay.GlobalId type with the following signature :

@dataclass
class GlobalId(Generic[Node]):
    id: str
    node_class: type[Node]

In fact, it is GlobalId type which is serialized and deserialized as an ID!, not the Id parameter of the Node class; apischema automatically add a field converter to make the conversion between the Id (for example an UUID) of a given node and the corresponding GlobalId.

Node instance global id can be retrieved with global_id property.

from dataclasses import dataclass

import graphql

from apischema import serialize
from apischema.graphql import graphql_schema, relay


@dataclass
class Faction(relay.Node[int]):
    name: str

    @classmethod
    def get_by_id(cls, id: int, info: graphql.GraphQLResolveInfo = None) -> "Faction":
        return [Faction(0, "Empire"), Faction(1, "Rebels")][id]


schema = graphql_schema(query=[relay.node], types=relay.nodes())
some_global_id = Faction.get_by_id(0).global_id  # Let's pick a global id ...
assert some_global_id == relay.GlobalId("0", Faction)
query = """
query factionName($id: ID!) {
    node(id: $id) {
        ... on Faction {
            name
        }
    }
}"""
assert graphql.graphql_sync(  # ... and use it in a query
    schema, query, variable_values={"id": serialize(relay.GlobalId, some_global_id)}
).data == {"node": {"name": "Empire"}}

Id encoding

Relay specifications encourage the use of base64 encoding, so apischema defines a relay.base64_encoding that you can pass to graphql_schema id_encoding parameter.

Connections

apischema provides a generic relay.Connection[Node, Cursor, Edge] type, which can be used directly without subclassing it; it's also possible to subclass it to add fields to a given connection (or to all the connection which will subclass the subclass). relay.Edge[Node, Cursor] can also be subclassed to add fields to the edges.

Connection dataclass has the following declaration:

@dataclass
class Connection(Generic[Node, Cursor, Edge]):
    edges: Optional[Sequence[Optional[Edge]]]
    has_previous_page: bool = field(default=False, metadata=skip)
    has_next_page: bool = field(default=False, metadata=skip)
    start_cursor: Optional[Cursor] = field(default=None, metadata=skip)
    end_cursor: Optional[Cursor] = field(default=None, metadata=skip)

    @resolver
    def page_info(self) -> PageInfo[Cursor]:
        ...

The pageInfo field is computed by a resolver; it uses the cursors of the first and the last edge when they are not provided.

Here is an example of Connection use:

from dataclasses import dataclass
from typing import Optional, TypeVar

import graphql
from graphql.utilities import print_schema

from apischema.graphql import graphql_schema, relay, resolver

Cursor = int  # let's use an integer cursor in all our connection
Node = TypeVar("Node")
Connection = relay.Connection[Node, Cursor, relay.Edge[Node, Cursor]]
# Connection can now be used just like Connection[Ship] or Connection[Faction | None]


@dataclass
class Ship:
    name: str


@dataclass
class Faction:
    @resolver
    def ships(
        self, first: int | None, after: Cursor | None
    ) -> Connection[Optional[Ship]] | None:
        edges = [relay.Edge(Ship("X-Wing"), 0), relay.Edge(Ship("B-Wing"), 1)]
        return Connection(edges, relay.PageInfo.from_edges(edges))


def faction() -> Faction | None:
    return Faction()


schema = graphql_schema(query=[faction])
schema_str = """\
type Query {
  faction: Faction
}

type Faction {
  ships(first: Int, after: Int): ShipConnection
}

type ShipConnection {
  edges: [ShipEdge]
  pageInfo: PageInfo!
}

type ShipEdge {
  node: Ship
  cursor: Int!
}

type Ship {
  name: String!
}

type PageInfo {
  hasPreviousPage: Boolean!
  hasNextPage: Boolean!
  startCursor: Int
  endCursor: Int
}"""
assert print_schema(schema) == schema_str
query = """
{
    faction {
        ships {
            pageInfo {
                endCursor
                hasNextPage
            }
            edges {
                cursor
                node {
                    name
                }
            }
        }
    }
}"""
assert graphql.graphql_sync(schema, query).data == {
    "faction": {
        "ships": {
            "pageInfo": {"endCursor": 1, "hasNextPage": False},
            "edges": [
                {"cursor": 0, "node": {"name": "X-Wing"}},
                {"cursor": 1, "node": {"name": "B-Wing"}},
            ],
        }
    }
}

Custom connections/edges

Connections can be customized by simply subclassing relay.Connection class and adding the additional fields.

For the edges, relay.Edge can be subclassed too, and the subclass has then to be passed as type argument to the generic connection.

from dataclasses import dataclass
from typing import Optional, TypeVar

from graphql import print_schema

from apischema.graphql import graphql_schema, relay, resolver

Cursor = int
Node = TypeVar("Node")
Edge = TypeVar("Edge", bound=relay.Edge)


@dataclass
class MyConnection(relay.Connection[Node, Cursor, Edge]):
    connection_field: bool


@dataclass
class MyEdge(relay.Edge[Node, Cursor]):
    edge_field: int | None


Connection = MyConnection[Node, MyEdge[Node]]


@dataclass
class Ship:
    name: str


@dataclass
class Faction:
    @resolver
    def ships(
        self, first: int | None, after: Cursor | None
    ) -> Connection[Optional[Ship]] | None:
        ...


def faction() -> Faction | None:
    return Faction()


schema = graphql_schema(query=[faction])
schema_str = """\
type Query {
  faction: Faction
}

type Faction {
  ships(first: Int, after: Int): ShipConnection
}

type ShipConnection {
  edges: [ShipEdge]
  pageInfo: PageInfo!
  connectionField: Boolean!
}

type ShipEdge {
  node: Ship
  cursor: Int!
  edgeField: Int
}

type Ship {
  name: String!
}

type PageInfo {
  hasPreviousPage: Boolean!
  hasNextPage: Boolean!
  startCursor: Int
  endCursor: Int
}"""
assert print_schema(schema) == schema_str

Mutations

Relay compliant mutations can be declared with a dataclass subclassing the relay.Mutation class; its fields will be put in the payload type of the mutation.

This class must implement a classmethod/staticmethod name mutate; it can be synchronous or asynchronous. The arguments of the method will correspond to the input type fields.

The mutation will be named after the name of the mutation class.

All the mutations declared can be retrieved with relay.mutations, in order to be passed to graphql_schema.

from dataclasses import dataclass

from graphql.utilities import print_schema

from apischema.graphql import graphql_schema, relay


@dataclass
class Ship:
    ...


@dataclass
class Faction:
    ...


@dataclass
class IntroduceShip(relay.Mutation):
    ship: Ship
    faction: Faction

    @staticmethod
    def mutate(faction_id: str, ship_name: str) -> "IntroduceShip":
        ...


def hello() -> str:
    return "world"


schema = graphql_schema(query=[hello], mutation=relay.mutations())
schema_str = """\
type Query {
  hello: String!
}

type Mutation {
  introduceShip(input: IntroduceShipInput!): IntroduceShipPayload!
}

type IntroduceShipPayload {
  ship: Ship!
  faction: Faction!
  clientMutationId: String
}

type Ship

type Faction

input IntroduceShipInput {
  factionId: String!
  shipName: String!
  clientMutationId: String
}"""
assert print_schema(schema) == schema_str

ClientMutationId

As you can see in the previous example, the field named clientMutationId is automatically added to the input and the payload types.

The forward of the mutation id from the input to the payload is automatically handled. It's value can be accessed by declaring a parameter of type relay.ClientMutationId — even if the parameter is not named client_mutation_id, it will be renamed internally.

This feature is controlled by a Mutation class variable _client_mutation_id, with 3 possible values:

• None (automatic, the default): clientMutationId field will be nullable unless it's declared as a required parameter (without default value) in the mutate method.
• False: their will be no clientMutationId field added (having a dedicated parameter will raise an error)
• True: clientMutationId is added and forced to be non-null.

from dataclasses import dataclass

from graphql.utilities import print_schema

from apischema.graphql import graphql_schema, relay


@dataclass
class Ship:
    ...


@dataclass
class Faction:
    ...


@dataclass
class IntroduceShip(relay.Mutation):
    ship: Ship
    faction: Faction

    @staticmethod
    def mutate(
        # mut_id is required because no default value
        faction_id: str,
        ship_name: str,
        mut_id: relay.ClientMutationId,
    ) -> "IntroduceShip":
        ...


def hello() -> str:
    return "world"


schema = graphql_schema(query=[hello], mutation=relay.mutations())
# clientMutationId field becomes non nullable in introduceShip types
schema_str = """\
type Query {
  hello: String!
}

type Mutation {
  introduceShip(input: IntroduceShipInput!): IntroduceShipPayload!
}

type IntroduceShipPayload {
  ship: Ship!
  faction: Faction!
  clientMutationId: String!
}

type Ship

type Faction

input IntroduceShipInput {
  factionId: String!
  shipName: String!
  clientMutationId: String!
}"""
assert print_schema(schema) == schema_str

Error handling and other resolver arguments

Relay mutation are operations, so they can be configured with the same parameters. As they are declared as classes, parameters will be passed as class variables, prefixed by _ (error_handler becomes _error_handler)

Note

Because parameters are class variables, you can reuse them by setting their value in a base class; for example, to share a same error_handler in a group of mutations.