Litestar

Strawberry comes with an integration for Litestar by providing a make_graphql_controller function that can be used to create a GraphQL controller.

See the example below for integrating Litestar with Strawberry:

import strawberry
from litestar import Litestar
from strawberry.litestar import make_graphql_controller
@strawberry.type
class Query:
@strawberry.field
def hello(self) -> str:
return "Hello World"
schema = strawberry.Schema(Query)
GraphQLController = make_graphql_controller(
schema,
path="/graphql",
)
app = Litestar(route_handlers=[GraphQLController])

Options

The make_graphql_controller function accepts the following options:

context_getter

The context_getter option allows you to provide a Litestar dependency that return a custom context object that can be used in your resolver.

import strawberry
from litestar import Request, Litestar
from strawberry.litestar import make_graphql_controller
from strawberry.types.info import Info
async def custom_context_getter():
return {"custom": "context"}
@strawberry.type
class Query:
@strawberry.field
def hello(self, info: strawberry.Info[dict, None]) -> str:
return info.context["custom"]
schema = strawberry.Schema(Query)
GraphQLController = make_graphql_controller(
schema,
path="/graphql",
context_getter=custom_context_getter,
)
app = Litestar(route_handlers=[GraphQLController])

The context_getter is a standard Litestar dependency and can receive any existing dependency:

import strawberry
from litestar import Request, Litestar
from strawberry.litestar import make_graphql_controller
from strawberry.types.info import Info
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import User
from sqlalchemy import select
async def custom_context_getter(request: Request, db_session: AsyncSession):
return {"user": request.user, "session": db_session}
@strawberry.type
class Query:
@strawberry.field
async def hello(self, info: strawberry.Info[dict, None]) -> str:
session: AsyncSession = info.context["session"]
user: User = info.context["user"]
query = select(User).where(User.id == user.id)
user = (await session.execute((query))).scalar_one()
return f"Hello {user.first_name}"
schema = strawberry.Schema(Query)
GraphQLController = make_graphql_controller(
schema,
path="/graphql",
context_getter=custom_context_getter,
)
app = Litestar(route_handlers=[GraphQLController])

You can also use a class-based custom context. To do this, you must inherit from BaseContext msgspec Struct or an InvalidCustomContext exception will be raised.

import strawberry
from litestar import Request, Litestar
from strawberry.litestar import make_graphql_controller, BaseContext
from strawberry.types.info import Info
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import User
from sqlalchemy import select
class CustomContext(BaseContext):
user: User
session: AsyncSession
async def custom_context_getter(
request: Request, db_session: AsyncSession
) -> CustomContext:
return CustomContext(user=request.user, session=db_session)
@strawberry.type
class Query:
@strawberry.field
async def hello(self, info: strawberry.Info[CustomContext, None]) -> str:
session: AsyncSession = info.context.session
user: User = info.context.user
query = select(User).where(User.id == user.id)
user = (await session.execute((query))).scalar_one()
return f"Hello {user.first_name}"
schema = strawberry.Schema(Query)
GraphQLController = make_graphql_controller(
schema,
path="/graphql",
context_getter=custom_context_getter,
)
app = Litestar(route_handlers=[GraphQLController])

Context typing

In our previous example using class based context, the actual runtime context a CustomContext type. Because it inherits from BaseContext , the request , socket and response attributes are typed as optional.

When inside a query/mutation resolver, request and response are always set and socket is only set in subscriptions.

To distinguish theses cases typing wise, the integration provides two classes that will help you to enforce strong typing:

from strawberry.litestar import HTTPContextType, WebSocketContextType

These classes does not actually exists at runtime, they are intended to be used to define a custom Info type with proper context typing. Taking over our previous example with class based custom context, here it how we can define two Info types for both queries/mutations and subscriptions:

import strawberry
from typing import Any
from litestar import Request, Litestar
from litestar.datastructures import State
from strawberry.litestar import (
make_graphql_controller,
BaseContext,
HTTPContextType,
WebSocketContextType,
)
from strawberry.types.info import Info
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import User
from sqlalchemy import select
class CustomContext(BaseContext, kw_only=True):
user: User
session: AsyncSession
class CustomHTTPContextType(HTTPContextType, CustomContext):
request: Request[User, Any, State]
class CustomWSContextType(WebSocketContextType, CustomContext):
socket: WebSocket[User, Token, State]
async def custom_context_getter(
request: Request, db_session: AsyncSession
) -> CustomContext:
return CustomContext(user=request.user, session=db_session)
@strawberry.type
class Query:
@strawberry.field
async def hello(self, info: strawberry.Info[CustomHTTPContextType, None]) -> str:
session: AsyncSession = info.context.session
user: User = info.context.user
query = select(User).where(User.id == user.id)
user = (await session.execute((query))).scalar_one()
return f"Hello {user.first_name}"
@strawberry.type
class Subscription:
@strawberry.subscription
async def count(
self, info: strawberry.Info[CustomWSContextType, None], target: int = 100
) -> AsyncGenerator[int, None]:
import devtools
devtools.debug(info.context)
devtools.debug(info.context.socket)
for i in range(target):
yield i
await asyncio.sleep(0.5)
schema = strawberry.Schema(Query, subscription=Subscription)
GraphQLController = make_graphql_controller(
schema,
path="/graphql",
context_getter=custom_context_getter,
)
app = Litestar(route_handlers=[GraphQLController])

root_value_getter

The root_value_getter option allows you to provide a custom root value that can be used in your resolver

import strawberry
from litestar import Request, Litestar
from strawberry.litestar import make_graphql_controller
@strawberry.type
class Query:
example: str = "Hello World"
@strawberry.field
def hello(self) -> str:
return self.example
def custom_get_root_value():
return Query()
schema = strawberry.Schema(Query)
GraphQLController = make_graphql_controller(
schema,
path="/graphql",
root_value_getter=custom_get_root_value,
)
app = Litestar(route_handlers=[GraphQLController])
Edit this page on GitHub