Strawberry GraphQL•7mo ago

AioInject errors while overriding resolve_connection method (relay)

I wanted to create my own relay connection class like so: I get this mypy error which shows an error in the method signature:

Signature of "resolve_connection" incompatible with supertype "Connection"Mypyoverride
     Superclass:
         def resolve_connection(cls, nodes: Iterator[T] | Iterable[T] | AsyncIterator[T] | AsyncIterable[T], *, info: Info[Any, Any], before: str | None = ..., after: str | None = ..., first: int | None = ..., last: int | None = ..., **kwargs: Any) -> Awaitable[KeysetConnection[T]] | KeysetConnection[T]
     Subclass:
         def resolve_connection(nodes: Iterator[T] | Iterable[T] | AsyncIterator[T] | AsyncIterable[T], session: AsyncSession, *, info: Info[Context, None], before: str | None = ..., after: str | None = ..., first: int | None = ..., last: int | None = ..., **kwargs: Any) -> Coroutine[Any, Any, KeysetConnection[T]]

Signature of "resolve_connection" incompatible with supertype "Connection"Mypyoverride
     Superclass:
         def resolve_connection(cls, nodes: Iterator[T] | Iterable[T] | AsyncIterator[T] | AsyncIterable[T], *, info: Info[Any, Any], before: str | None = ..., after: str | None = ..., first: int | None = ..., last: int | None = ..., **kwargs: Any) -> Awaitable[KeysetConnection[T]] | KeysetConnection[T]
     Subclass:
         def resolve_connection(nodes: Iterator[T] | Iterable[T] | AsyncIterator[T] | AsyncIterable[T], session: AsyncSession, *, info: Info[Context, None], before: str | None = ..., after: str | None = ..., first: int | None = ..., last: int | None = ..., **kwargs: Any) -> Coroutine[Any, Any, KeysetConnection[T]]

How could this be fixed? I need to get my session within the resolver

3 Replies

Aryan IyappanOP•7mo ago

from typing import (
    Annotated,
    Any,
    AsyncIterable,
    AsyncIterator,
    Iterable,
    Iterator,
    List,
    Optional,
    TypeVar,
)

import strawberry
from aioinject import Inject
from aioinject.ext.strawberry import inject
from sqlakeyset import Page, unserialize_bookmark
from sqlakeyset.asyncio import select_page
from sqlalchemy.ext.asyncio import AsyncSession
from strawberry import relay

from app.context import Info

T = TypeVar("T", bound=relay.Node)


@strawberry.type(name="Connection", description="A connection to a list of items.")
class KeysetConnection(relay.Connection[T]):
    edges: List[relay.Edge[T]] = strawberry.field(
        description="Contains the nodes in this connection",
    )

    @classmethod
    @inject
    async def resolve_connection(
        cls,
        nodes: Iterator[T] | Iterable[T] | AsyncIterator[T] | AsyncIterable[T],
        session: Annotated[AsyncSession, Inject],
        *,
        info: Info,
        before: Optional[str] = None,
        after: Optional[str] = None,
        first: Optional[int] = None,
        last: Optional[int] = None,
        **kwargs: Any,
    ) -> "KeysetConnection[T]":
        if first is not None and last is not None:
            raise ValueError("Cannot specify both first and last")

        max_results = info.schema.config.relay_max_results
        per_page = first or last or max_results

        if per_page > max_results:
            raise ValueError(f"Cannot request more than {max_results} results")

        # Use sqlakeyset to paginate
        page: Page = await select_page(
            session,
            nodes,
            per_page=per_page,
            after=unserialize_bookmark(after) if after else None,
            before=unserialize_bookmark(before) if before else None,
        )

from typing import (
    Annotated,
    Any,
    AsyncIterable,
    AsyncIterator,
    Iterable,
    Iterator,
    List,
    Optional,
    TypeVar,
)

import strawberry
from aioinject import Inject
from aioinject.ext.strawberry import inject
from sqlakeyset import Page, unserialize_bookmark
from sqlakeyset.asyncio import select_page
from sqlalchemy.ext.asyncio import AsyncSession
from strawberry import relay

from app.context import Info

T = TypeVar("T", bound=relay.Node)


@strawberry.type(name="Connection", description="A connection to a list of items.")
class KeysetConnection(relay.Connection[T]):
    edges: List[relay.Edge[T]] = strawberry.field(
        description="Contains the nodes in this connection",
    )

    @classmethod
    @inject
    async def resolve_connection(
        cls,
        nodes: Iterator[T] | Iterable[T] | AsyncIterator[T] | AsyncIterable[T],
        session: Annotated[AsyncSession, Inject],
        *,
        info: Info,
        before: Optional[str] = None,
        after: Optional[str] = None,
        first: Optional[int] = None,
        last: Optional[int] = None,
        **kwargs: Any,
    ) -> "KeysetConnection[T]":
        if first is not None and last is not None:
            raise ValueError("Cannot specify both first and last")

        max_results = info.schema.config.relay_max_results
        per_page = first or last or max_results

        if per_page > max_results:
            raise ValueError(f"Cannot request more than {max_results} results")

        # Use sqlakeyset to paginate
        page: Page = await select_page(
            session,
            nodes,
            per_page=per_page,
            after=unserialize_bookmark(after) if after else None,
            before=unserialize_bookmark(before) if before else None,
        )

        # Create edges
        # (Overriding Edge.resolve_edge as we already have base64 encoded cursors from
        # sqlakeyset available!)
        edges = [
            relay.Edge(node=node, cursor=page.paging.get_bookmark_at(i))
            for i, node in enumerate(page)
        ]

        return cls(
            edges=edges,
            page_info=relay.PageInfo(
                has_next_page=page.paging.has_next,
                has_previous_page=page.paging.has_previous,
                start_cursor=page.paging.get_bookmark_at(0) if page else None,
                end_cursor=page.paging.get_bookmark_at(-1) if page else None,
            ),
        )

        # Create edges
        # (Overriding Edge.resolve_edge as we already have base64 encoded cursors from
        # sqlakeyset available!)
        edges = [
            relay.Edge(node=node, cursor=page.paging.get_bookmark_at(i))
            for i, node in enumerate(page)
        ]

        return cls(
            edges=edges,
            page_info=relay.PageInfo(
                has_next_page=page.paging.has_next,
                has_previous_page=page.paging.has_previous,
                start_cursor=page.paging.get_bookmark_at(0) if page else None,
                end_cursor=page.paging.get_bookmark_at(-1) if page else None,
            ),
        )

Unknown User•7mo ago

Message Not Public

Aryan IyappanOP•6mo ago

I'm revamping the way I'm using connections, actually. So I'm having a repo layer as well as a service layer behind the resolver layer. The repo layer handles the pagination and returns a data class called PaginatedResult. My resolver inturn converts it into a connection by just type casting the right fields. I think this approach is more scalable and aligns with separation of concerns principles better So IMO this connection abstraction via the decorator doesn't really scale. Or atleast it should allow the field to return any object (like the PaginatedResult object I mentioned above), rather than just a list of nodes What do you think?? Should I propose a new feature about this?🤔🤔

Gaming

Programming

AioInject errors while overriding resolve_connection method (relay)