aiohttp + strawberry + aiodataloader example

app.py

from __future__ import annotations

from aiohttp import web
from strawberry.http import process_result

import loader
import graph

routes = web.RouteTableDef()


@routes.post('/graph')
async def hello(request):
    data = await request.json()
    query = data["query"]
    variables = data.get("variables")
    operation_name = data.get("operationName")

    root_value = None
    context = {"request": request}
    async with loader.context():
        result = await graph.schema.execute(
            query,
            variable_values=variables,
            context_value=context,
            operation_name=operation_name,
            root_value=root_value,
        )
        result = process_result(result)

    return web.json_response(result)


app = web.Application()
app.add_routes(routes)
web.run_app(app)

graph.py

from __future__ import annotations

from typing import List

import strawberry
from graphql import GraphQLResolveInfo

import loader


async def get_books():
    return [
        Book(id="1", title='The Great Gatsby'),
        Book(id="2", title='The Great Gatsby'),
        Book(id="3", title='The Great Gatsby'),
        Book(id="4", title='The Great Gatsby'),
        Book(id="5", title='The Great Gatsby'),
        Book(id="3", title='The Great Gatsby'),
    ]


async def get_book_author(info, ids) -> List[Author]:
    print('Fetch authors: ', ids)
    return [Author(id=id_, name=f'Author {id_}') for id_ in ids]


@strawberry.type
class Author:
    id: str
    name: str


@strawberry.type
class Book:
    id: str
    title: str

    @strawberry.field
    async def author(self, info: GraphQLResolveInfo) -> Author:
        return await loader.load_key(
            info=info,
            resolver=get_book_author,
            key=self.id,
        )


@strawberry.type
class Query:

    @strawberry.field
    async def books(self, info: GraphQLResolveInfo) -> List[Book]:
        return await get_books()


schema = strawberry.Schema(query=Query)

loader.py

from __future__ import annotations

from contextlib import asynccontextmanager
from contextvars import ContextVar
from typing import Any, AsyncIterator, Awaitable, List

from aiodataloader import DataLoader
from graphql import GraphQLResolveInfo

context_loaders = ContextVar('loaders')


def load_key(info: GraphQLResolveInfo, resolver, key: Any) -> Awaitable[Any]:
    # For null key return nullable awaitable
    if key is None:
        async def _nothing(*args: Any, **kwargs: Any) -> None:
            return None

        return _nothing()

    # Try to select loader from dict of loaders
    # of the current execution context
    loaders = context_loaders.get()
    loader = loaders.get(resolver)

    # If loader not exist in current context, create new one
    if not loader:

        async def _resolver(ids: List[Any]) -> List[Any]:
            return await resolver(info, ids)

        loader = DataLoader(batch_load_fn=_resolver)
        loaders[resolver] = loader
        context_loaders.set(loaders)

    return loader.load(key)


@asynccontextmanager
async def context() -> AsyncIterator[None]:
    token = context_loaders.set({})
    try:
        yield
    finally:
        context_loaders.reset(token)