Overview
FastAPI can be integrated with GraphQL libraries to provide a GraphQL API alongside or instead of REST endpoints. GraphQL offers a flexible query language that allows clients to request exactly the data they need.
The two most popular GraphQL libraries for Python are:
Strawberry - Modern, type-hint based GraphQL library (recommended)
Graphene - Mature GraphQL library with extensive features
Strawberry is recommended for FastAPI projects because it uses Python type hints similar to FastAPI, providing a consistent development experience.
Strawberry GraphQL
Installation
pip install strawberry-graphql[fastapi]
Basic Setup
Here’s a complete example integrating Strawberry with FastAPI:
import strawberry
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
@strawberry.type
class User :
name: str
age: int
@strawberry.type
class Query :
@strawberry.field
def user ( self ) -> User:
return User( name = "Patrick" , age = 100 )
schema = strawberry.Schema( query = Query)
graphql_app = GraphQLRouter(schema)
app = FastAPI()
app.include_router(graphql_app, prefix = "/graphql" )
Now you can access:
GraphQL endpoint: http://localhost:8000/graphql
GraphQL Playground: http://localhost:8000/graphql (in browser)
Queries with Arguments
@strawberry.type
class Query :
@strawberry.field
def user ( self , user_id : int ) -> User:
# Fetch user from database
return User( name = f "User { user_id } " , age = 25 )
@strawberry.field
def users ( self , limit : int = 10 ) -> list[User]:
# Fetch multiple users
return [User( name = f "User { i } " , age = 20 + i) for i in range (limit)]
Query example:
query {
user ( userId : 1 ) {
name
age
}
users ( limit : 5 ) {
name
}
}
Mutations
Add mutations for creating or updating data:
@strawberry.type
class CreateUserInput :
name: str
age: int
email: str
@strawberry.type
class Mutation :
@strawberry.mutation
def create_user ( self , input : CreateUserInput) -> User:
# Save user to database
return User( name = input .name, age = input .age)
@strawberry.mutation
def update_user ( self , user_id : int , name : str ) -> User:
# Update user in database
return User( name = name, age = 25 )
schema = strawberry.Schema( query = Query, mutation = Mutation)
Mutation example:
mutation {
createUser ( input : {
name : "Alice" ,
age : 30 ,
email : "[email protected] "
}) {
name
age
}
}
Async Resolvers
Strawberry supports async/await for database operations:
import strawberry
from typing import Optional
@strawberry.type
class Query :
@strawberry.field
async def user ( self , user_id : int ) -> Optional[User]:
# Async database query
user_data = await database.fetch_user(user_id)
if user_data:
return User( name = user_data[ "name" ], age = user_data[ "age" ])
return None
@strawberry.field
async def users ( self ) -> list[User]:
users_data = await database.fetch_all_users()
return [User( ** user) for user in users_data]
Use async resolvers when performing I/O operations like database queries or API calls to maintain FastAPI’s async performance benefits.
Relationships and Data Loaders
Handle nested relationships efficiently:
import strawberry
from strawberry.dataloader import DataLoader
from typing import List
@strawberry.type
class Post :
id : int
title: str
author_id: int
@strawberry.field
async def author ( self , info ) -> "User" :
# Use DataLoader to avoid N+1 queries
return await info.context[ "user_loader" ].load( self .author_id)
@strawberry.type
class User :
id : int
name: str
age: int
@strawberry.field
async def posts ( self , info ) -> List[Post]:
return await info.context[ "post_loader" ].load( self .id)
async def load_users ( keys : List[ int ]) -> List[User]:
# Batch load users
users = await database.fetch_users_by_ids(keys)
return users
async def load_posts ( keys : List[ int ]) -> List[List[Post]]:
# Batch load posts for multiple users
posts = await database.fetch_posts_by_user_ids(keys)
return posts
# Configure context with data loaders
async def get_context ():
return {
"user_loader" : DataLoader( load_fn = load_users),
"post_loader" : DataLoader( load_fn = load_posts),
}
graphql_app = GraphQLRouter(schema, context_getter = get_context)
Without DataLoaders, nested queries can cause N+1 query problems. Always use DataLoaders when resolving relationships to batch database queries efficiently.
Authentication
Integrate with FastAPI’s dependency injection:
from fastapi import Depends, HTTPException, status
from strawberry.fastapi import GraphQLRouter
from strawberry.types import Info
def get_current_user ( token : str = Header( ... )):
# Verify token and return user
if not token:
raise HTTPException( status_code = 401 , detail = "Not authenticated" )
return { "id" : 1 , "name" : "User" }
async def get_context (
current_user : dict = Depends(get_current_user)
):
return { "user" : current_user}
@strawberry.type
class Query :
@strawberry.field
def me ( self , info : Info) -> User:
current_user = info.context[ "user" ]
return User(
id = current_user[ "id" ],
name = current_user[ "name" ],
age = 25
)
graphql_app = GraphQLRouter(
schema,
context_getter = get_context
)
Graphene Integration
Installation
pip install graphene
pip install starlette-graphene3
Basic Setup
import graphene
from starlette_graphene3 import GraphQLApp, make_graphiql_handler
from fastapi import FastAPI
class User ( graphene . ObjectType ):
id = graphene.Int()
name = graphene.String()
age = graphene.Int()
class Query ( graphene . ObjectType ):
user = graphene.Field(User, user_id = graphene.Int())
def resolve_user ( self , info , user_id ):
return User( id = user_id, name = f "User { user_id } " , age = 25 )
class CreateUser ( graphene . Mutation ):
class Arguments :
name = graphene.String( required = True )
age = graphene.Int( required = True )
user = graphene.Field(User)
def mutate ( self , info , name , age ):
user = User( id = 1 , name = name, age = age)
return CreateUser( user = user)
class Mutation ( graphene . ObjectType ):
create_user = CreateUser.Field()
schema = graphene.Schema( query = Query, mutation = Mutation)
app = FastAPI()
app.mount( "/graphql" , GraphQLApp(schema, on_get = make_graphiql_handler()))
Combining REST and GraphQL
You can use both REST and GraphQL in the same FastAPI application:
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
import strawberry
app = FastAPI()
# REST endpoints
@app.get ( "/api/health" )
async def health_check ():
return { "status" : "healthy" }
@app.get ( "/api/users/ {user_id} " )
async def get_user_rest ( user_id : int ):
return { "id" : user_id, "name" : "User" }
# GraphQL endpoint
@strawberry.type
class Query :
@strawberry.field
def user ( self , user_id : int ) -> User:
return User( name = "User" , age = 25 )
schema = strawberry.Schema( query = Query)
graphql_app = GraphQLRouter(schema)
app.include_router(graphql_app, prefix = "/graphql" )
This approach works well when migrating from REST to GraphQL or when different clients have different needs.
Subscriptions (WebSocket)
Strawberry supports GraphQL subscriptions for real-time updates:
import asyncio
import strawberry
from typing import AsyncGenerator
@strawberry.type
class Subscription :
@strawberry.subscription
async def count ( self , target : int = 10 ) -> AsyncGenerator[ int , None ]:
for i in range (target):
yield i
await asyncio.sleep( 1 )
@strawberry.subscription
async def user_updates ( self , user_id : int ) -> AsyncGenerator[User, None ]:
# Subscribe to user updates from a message queue
async for update in message_queue.subscribe( f "user: { user_id } " ):
yield User( ** update)
schema = strawberry.Schema(
query = Query,
mutation = Mutation,
subscription = Subscription
)
graphql_app = GraphQLRouter(schema)
Subscription example:
subscription {
count ( target : 5 )
}
subscription {
userUpdates ( userId : 1 ) {
name
age
}
}
Error Handling
Handle errors gracefully in GraphQL:
import strawberry
from typing import Optional
@strawberry.type
class UserError :
message: str
code: str
@strawberry.type
class UserResult :
user: Optional[User] = None
error: Optional[UserError] = None
@strawberry.type
class Query :
@strawberry.field
async def user ( self , user_id : int ) -> UserResult:
try :
user_data = await database.fetch_user(user_id)
if not user_data:
return UserResult(
error = UserError(
message = "User not found" ,
code = "USER_NOT_FOUND"
)
)
return UserResult( user = User( ** user_data))
except Exception as e:
return UserResult(
error = UserError(
message = "Internal server error" ,
code = "INTERNAL_ERROR"
)
)
Testing GraphQL Endpoints
from fastapi.testclient import TestClient
import json
def test_graphql_query ():
client = TestClient(app)
query = """
query {
user(userId: 1) {
name
age
}
}
"""
response = client.post(
"/graphql" ,
json = { "query" : query}
)
assert response.status_code == 200
data = response.json()
assert "data" in data
assert data[ "data" ][ "user" ][ "name" ] == "User 1"
def test_graphql_mutation ():
client = TestClient(app)
mutation = """
mutation {
createUser(input: {
name: "Alice",
age: 30,
email: "[email protected] "
}) {
name
age
}
}
"""
response = client.post(
"/graphql" ,
json = { "query" : mutation}
)
assert response.status_code == 200
data = response.json()
assert data[ "data" ][ "createUser" ][ "name" ] == "Alice"
Best Practices
Use DataLoaders Always use DataLoaders for relationships to avoid N+1 query problems and batch database operations.
Schema Design Design your schema around client needs, not database structure. Think in terms of the graph of relationships.
Rate Limiting Implement query complexity analysis and rate limiting to prevent abuse of expensive queries.
Monitoring Monitor GraphQL query performance and identify slow resolvers to optimize database queries.
# Enable query depth limiting
from strawberry.extensions import QueryDepthLimiter
schema = strawberry.Schema(
query = Query,
extensions = [
QueryDepthLimiter( max_depth = 10 ),
]
)
# Add query caching
from strawberry.extensions import AddValidationRules
from graphql import ValidationRule
schema = strawberry.Schema(
query = Query,
extensions = [
AddValidationRules([ ... ]),
]
)
Consider implementing persisted queries in production to reduce bandwidth and improve security by only allowing pre-approved queries.