FastAPI

Strawberry provides support for FastAPI with a custom APIRouter called GraphQLRouter .

Before using Strawberry’s FastAPI support make sure you install all the required dependencies by running:

Terminal window
pip install 'strawberry-graphql[fastapi]'

See the example below for integrating FastAPI with Strawberry: 

import strawberry


from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter




@strawberry.type
class Query:
    @strawberry.field
    def hello(self) -> str:
        return "Hello World"




schema = strawberry.Schema(Query)


graphql_app = GraphQLRouter(schema)


app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

Options

The GraphQLRouter accepts the following options:

context_getter

The context_getter option allows you to provide a custom context object that can be used in your resolver. context_getter is a FastAPI dependency and can inject other dependencies if you so wish.

There are two options at your disposal here:

  1. Define your custom context as a dictionary,
  2. Define your custom context as a class.

If no context is supplied, then the default context returned is a dictionary containing the request, the response, and any background tasks.

However, you can define a class-based custom context inline with FastAPI practice . If you choose to do this, you must ensure that your custom context class inherits from BaseContext or an InvalidCustomContext exception is raised.

For dictionary-based custom contexts, an example might look like the following. 

import strawberry


from fastapi import FastAPI, Depends, Request, WebSocket, BackgroundTasks
from strawberry.fastapi import GraphQLRouter




def custom_context_dependency() -> str:
    return "John"




async def get_context(
    custom_value=Depends(custom_context_dependency),
):
    return {
        "custom_value": custom_value,
    }




@strawberry.type
class Query:
    @strawberry.field
    def example(self, info: strawberry.Info) -> str:
        return f"Hello {info.context['custom_value']}"




schema = strawberry.Schema(Query)


graphql_app = GraphQLRouter(
    schema,
    context_getter=get_context,
)


app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

Here we are returning a custom context dictionary that contains one extra item called “customvalue”, which is injected from custom_context_dependency . This value exists alongside request , response , and background_tasks in the info.context _dictionary and so it requires ['request'] indexing.

Then we use the context in a resolver. The resolver will return “Hello John” in this case.

For class-based custom contexts, an example might look like the following. 

import strawberry


from fastapi import FastAPI, Depends, Request, WebSocket, BackgroundTasks
from strawberry.fastapi import BaseContext, GraphQLRouter




class CustomContext(BaseContext):
    def __init__(self, greeting: str, name: str):
        self.greeting = greeting
        self.name = name




def custom_context_dependency() -> CustomContext:
    return CustomContext(greeting="you rock!", name="John")




async def get_context(
    custom_context=Depends(custom_context_dependency),
):
    return custom_context




@strawberry.type
class Query:
    @strawberry.field
    def example(self, info: strawberry.Info) -> str:
        return f"Hello {info.context.name}, {info.context.greeting}"




schema = strawberry.Schema(Query)


graphql_app = GraphQLRouter(
    schema,
    context_getter=get_context,
)


app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

In this case, we are returning a custom context class that inherits from BaseContext with fields name and greeting , which is also injected by custom_context_dependency . These custom values exist alongside request , response , and background_tasks in the info.context class and so it requires .request indexing.

Then we use the context in a resolver. The resolver will return “Hello John, you rock!” in this case.

Setting background tasks

Similarly, background tasks can be added via the context: 

import strawberry


from fastapi import FastAPI, BackgroundTasks
from strawberry.fastapi import GraphQLRouter




async def notify_new_flavour(name: str):
    print(name)




@strawberry.type
class Query:
    @strawberry.field
    def hello(self) -> str:
        return "Hello World"




@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_flavour(self, name: str, info: strawberry.Info) -> bool:
        info.context["background_tasks"].add_task(notify_new_flavour, name)
        return True




schema = strawberry.Schema(Query, Mutation)


graphql_app = GraphQLRouter(schema)


app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

If using a custom context class, then background tasks should be stored within the class object as .background_tasks .

root_value_getter

The root_value_getter option allows you to provide a custom root value for your schema. This is most likely a rare usecase but might be useful in certain situations.

Here’s an example: 

import strawberry


from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter




async def get_root_value():
    return Query(name="Patrick")




@strawberry.type
class Query:
    name: str




schema = strawberry.Schema(Query)


graphql_app = GraphQLRouter(
    schema,
    root_value_getter=get_root_value,
)


app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

Here we are returning a Query where the name is “Patrick”, so when we request the field name we’ll return “Patrick”.

process_result

The process_result option allows you to customize and/or process results before they are sent to the clients. This can be useful for logging errors or hiding them (for example to hide internal exceptions).

It needs to return a GraphQLHTTPResponse object and accepts the request and execution results. 

from fastapi import Request
from strawberry.fastapi import GraphQLRouter
from strawberry.http import GraphQLHTTPResponse
from strawberry.types import ExecutionResult




class MyGraphQLRouter(GraphQLRouter):
    async def process_result(
        self, request: Request, result: ExecutionResult
    ) -> GraphQLHTTPResponse:
        data: GraphQLHTTPResponse = {"data": result.data}


        if result.errors:
            data["errors"] = [err.formatted for err in result.errors]


        return data

In this case we are doing the default processing of the result, but it can be tweaked based on your needs.

encode_json

encode_json allows to customize the encoding of the JSON response. By default we use json.dumps but you can override this method to use a different encoder. For example, the orjson library from pypi has blazing fast speeds. 

class MyGraphQLRouter(GraphQLRouter):
    def encode_json(self, data: GraphQLHTTPResponse) -> bytes:
        return orjson.dumps(data)

Contributors:

Edit this page on GitHub