Existing Relay API needs modifications to be usable in practical applications

[aryaniyaps]

While using the relay API provided by strawberry GraphQL, I've been noticing a few quirks that make the existing API patterns unusable in the real world.

Quirk 1- Relay Connections API

The documentation shows that we can use the relay Connections API 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()

However, there are a few issues with this approach:

• there is no way to introduce dependency injection over here
• pagination usually occurs at the repository layer, when using a layered architecture, which is very common. This logic is not used in the resolver layer commonly.

Here's how I've setup Relay Connections with a layered architecture instead:

from typing import Generic, Self, TypeVar

import strawberry
from strawberry import relay

from app.database.base import Base

ModelType = TypeVar("ModelType", bound=Base)


@strawberry.type
class BaseNodeType(Generic[ModelType], relay.Node):
    id: relay.NodeID[int]

    @classmethod
    def from_orm(cls, model: ModelType) -> Self:
        """Construct a node from an ORM instance."""
        raise NotImplementedError


@strawberry.interface(name="Error")
class BaseErrorType:
    message: str


NodeType = TypeVar("NodeType", bound=BaseNodeType[Base])

@strawberry.type(name="Answer")
class AnswerType(BaseNodeType[Answer]):
    content: str
    created_at: datetime
    updated_at: datetime | None

    @classmethod
    def from_orm(cls, answer: Answer) -> Self:
        """Construct a node from an ORM instance."""
        return cls(
            id=answer.id,
            content=answer.content,
            created_at=answer.created_at,
            updated_at=answer.updated_at,
        )

    @classmethod
    async def resolve_nodes(  # type: ignore[no-untyped-def] # noqa: ANN206
        cls,
        *,
        info: Info,
        node_ids: Iterable[str],
        required: bool = False,  # noqa: ARG003
    ):
        answers = await info.context["loaders"].answer_by_id.load_many(node_ids)
        return [
            cls.from_orm(answer) if answer is not None else answer for answer in answers
        ]


@strawberry.type(name="AnswerConnection")
class AnswerConnectionType(relay.Connection[AnswerType]):
    @classmethod
    def from_paginated_result(
        cls, paginated_result: PaginatedResult[Answer, int]
    ) -> Self:
        return cls(
            page_info=relay.PageInfo(
                has_next_page=paginated_result.page_info.has_next_page,
                has_previous_page=paginated_result.page_info.has_previous_page,
                start_cursor=relay.to_base64(
                    AnswerType,
                    paginated_result.page_info.start_cursor,
                )
                if paginated_result.page_info.start_cursor
                else None,
                end_cursor=relay.to_base64(
                    AnswerType,
                    paginated_result.page_info.end_cursor,
                )
                if paginated_result.page_info.end_cursor
                else None,
            ),
            edges=[
                relay.Edge(
                    node=AnswerType.from_orm(answer),
                    cursor=relay.to_base64(AnswerType, answer.id),
                )
                for answer in paginated_result.entities
            ],
        )

...

   @strawberry.field
    @inject
    async def answers(
        self,
        answer_repo: Annotated[
            AnswerRepo,
            Inject,
        ],
        before: relay.GlobalID | None = None,
        after: relay.GlobalID | None = None,
        first: int | None = None,
        last: int | None = None,
    ) -> AnswerConnectionType:
        """Return a paginated connection of answers for the question."""
        # Retrieve paginated answers for this question
        paginated_answers = await answer_repo.get_all(
            question_id=self.id,
            after=(int(after.node_id) if after else None),
            before=(int(before.node_id) if before else None),
            first=first,
            last=last,
        )

        # Convert to AnswerConnectionType
        return AnswerConnectionType.from_paginated_result(paginated_answers)

Quirk 2- Global ID API

Whenever I declare a NodeID like this:

@strawberry.type
class BaseNodeType(Generic[ModelType], relay.Node):
    id: relay.NodeID[int]

    @classmethod
    def from_orm(cls, model: ModelType) -> Self:
        """Construct a node from an ORM instance."""
        raise NotImplementedError

A GlobalID is automatically being generated in the schema. (referencing #3551 )

My understanding is that this dataclass was initially meant to help conversion to and from base64 opaque ids, but ended up being a custom scalar in the output GraphQL schema as well.

This probably doesn't matter much, until when you're facing errors from the relay compiler which strictly requires the ID fields to be named as ID, and not GlobalID, as described in #3551 ,

As a result, Im using a hack like this in the meantime:

import strawberry
from strawberry.relay import GlobalID

# temporary hack until strawberry fixes relay ID scalar generation
# https://github.com/strawberry-graphql/strawberry/issues/3551
ID = strawberry.scalar(
    strawberry.ID,
    serialize=str,
    parse_value=GlobalID.from_id,
)

I think that the GlobalID field should be an internal helper and not a custom scalar type.

Making these changes would probably break some stuff, but I wanted to put forth these DX issues to consider them before making things stable!