GraphQL FastAPI Integration Guide
FastAPI is Python's fastest web framework for building REST and async APIs. Strawberry integrates seamlessly with FastAPI, letting you run GraphQL and REST endpoints on the same server, share dependencies (database, auth), and leverage FastAPI's features (validation, OpenAPI docs, background tasks). This article teaches you to mount Strawberry on FastAPI, unify auth across both protocols, and design a hybrid API.
In 2024, I maintained a REST API and a newer GraphQL API on separate servers—painful to keep auth synchronized, expensive to run both. Combining them on FastAPI simplified everything: one auth system, one database connection pool, one deployment. This article shows that integration.
Basic FastAPI + Strawberry Integration
Install dependencies:
pip install strawberry-graphql fastapi uvicorn starlette
Create a hybrid API:
import strawberry
from fastapi import FastAPI, Depends
from strawberry.asgi import GraphQL
# Define your Strawberry schema.
@strawberry.type
class User:
id: int
name: str
@strawberry.type
class Query:
@strawberry.field
def user(self, id: int) -> User:
return User(id=id, name="Alice")
schema = strawberry.Schema(query=Query)
# Create FastAPI app.
app = FastAPI()
# Mount GraphQL on /graphql route.
graphql_app = GraphQL(schema)
app.add_route("/graphql", graphql_app)
app.add_websocket_route("/graphql", graphql_app)
# Add REST endpoints.
@app.get("/users/{user_id}")
async def get_user(user_id: int):
return {"id": user_id, "name": "Alice"}
# Run with: uvicorn main:app --reload
Now you have:
GET /users/1→ REST endpointPOST /graphql→ GraphQL queries and mutationsWebSocket /graphql→ GraphQL subscriptions
Sharing Database and Context Between REST and GraphQL
Share a database connection pool:
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from fastapi import Depends, Request
import strawberry
# Create async SQLAlchemy engine.
engine = create_async_engine("postgresql+asyncpg://user:password@localhost/dbname")
async_session_maker = sessionmaker(engine, class_=AsyncSession, expire_on_fetch=False)
# Dependency for FastAPI endpoints.
async def get_db():
async with async_session_maker() as session:
yield session
# Context getter for Strawberry.
async def get_context(request: Request) -> dict:
session = async_session_maker()
return {
'db': session,
'request': request,
'user_id': extract_user_id(request), # Covered in auth article.
}
# Strawberry schema with context.
@strawberry.type
class Query:
@strawberry.field
async def user(self, id: int, info: strawberry.Info) -> Optional[User]:
db = info.context['db']
# Query the database.
result = await db.execute(
select(UserModel).filter(UserModel.id == id)
)
return result.scalar_one_or_none()
schema = strawberry.Schema(query=Query)
# FastAPI app.
app = FastAPI()
# GraphQL route with context.
graphql_app = GraphQL(schema, context_getter=get_context)
app.add_route("/graphql", graphql_app)
# REST route with same database.
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
result = await db.execute(
select(UserModel).filter(UserModel.id == user_id)
)
user = result.scalar_one_or_none()
return {"id": user.id, "name": user.name}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Both GraphQL and REST endpoints now share the same database session maker. Authentication logic can also be shared:
from jose import JWTError, jwt
def extract_user_id(request: Request) -> Optional[int]:
"""Extract user ID from JWT token in Authorization header."""
auth_header = request.headers.get('Authorization', '')
if auth_header.startswith('Bearer '):
token = auth_header[7:]
try:
payload = jwt.decode(token, SECRET, algorithms=['HS256'])
return payload['user_id']
except JWTError:
return None
return None
# Use in both REST and GraphQL.
@app.get("/users/{user_id}")
async def get_user(user_id: int, request: Request):
current_user_id = extract_user_id(request)
if not current_user_id:
raise HTTPException(status_code=401)
# Fetch and return...
Structured Exceptions: REST and GraphQL Errors
Define custom exceptions that work with both REST and GraphQL:
from fastapi import HTTPException
import strawberry
class AppException(Exception):
"""Base app exception."""
def __init__(self, code: str, message: str, status_code: int = 400):
self.code = code
self.message = message
self.status_code = status_code
super().__init__(message)
class UserNotFound(AppException):
def __init__(self, user_id: int):
super().__init__(
code="USER_NOT_FOUND",
message=f"User {user_id} not found.",
status_code=404
)
# In REST endpoint.
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
try:
result = await db.execute(select(UserModel).filter(UserModel.id == user_id))
user = result.scalar_one_or_none()
if not user:
raise UserNotFound(user_id)
return user
except AppException as e:
raise HTTPException(status_code=e.status_code, detail=e.message)
# In GraphQL resolver.
@strawberry.type
class Query:
@strawberry.field
async def user(self, id: int, info: strawberry.Info) -> User:
try:
db = info.context['db']
result = await db.execute(select(UserModel).filter(UserModel.id == id))
user = result.scalar_one_or_none()
if not user:
raise UserNotFound(id)
return User(id=user.id, name=user.name)
except AppException as e:
raise strawberry.errors.GraphQLError(
message=e.message,
extensions={"code": e.code}
)
Both REST and GraphQL return consistent error formats.
Middleware: CORS, Logging, Rate Limiting
FastAPI middleware applies to all routes (REST and GraphQL):
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import logging
# Setup logging.
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# CORS for cross-origin requests (browsers).
app.add_middleware(
CORSMiddleware,
allow_origins=["https://example.com", "https://app.example.com"],
allow_methods=["GET", "POST"],
allow_headers=["Authorization", "Content-Type"],
)
# Trusted hosts.
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["example.com"])
# Custom logging middleware.
@app.middleware("http")
async def log_requests(request: Request, call_next):
logger.info(f"{request.method} {request.url.path}")
response = await call_next(request)
logger.info(f"Response status: {response.status_code}")
return response
# Rate limiting (using slowapi library).
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
@app.get("/users/{user_id}")
@limiter.limit("100/minute")
async def get_user(user_id: int, request: Request):
# ...
pass
Middleware applies to both REST and GraphQL routes.
Combining REST and GraphQL: Which to Use?
| Use Case | Recommendation |
|---|---|
| Public API for mobile/web clients | GraphQL (flexible, efficient) |
| Simple CRUD operations | REST (simpler to understand) |
| Internal service-to-service API | REST or gRPC (lower overhead) |
| Real-time updates | GraphQL Subscriptions |
| File uploads | REST (GraphQL file support is newer) |
| Existing REST API | Keep REST, add GraphQL alongside |
Complete Production Setup
Tie everything together:
import strawberry
from fastapi import FastAPI, Depends, Request, HTTPException
from strawberry.asgi import GraphQL
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
import logging
import os
# Configuration.
DB_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://localhost/myapp")
SECRET = os.getenv("SECRET", "dev-secret")
ALLOWED_ORIGINS = os.getenv("ALLOWED_ORIGINS", "http://localhost:3000").split(",")
# Setup.
engine = create_async_engine(DB_URL)
async_session_maker = sessionmaker(engine, class_=AsyncSession, expire_on_fetch=False)
logger = logging.getLogger(__name__)
async def get_context(request: Request) -> dict:
session = async_session_maker()
user_id = extract_user_id(request)
return {
'db': session,
'request': request,
'user_id': user_id,
}
@strawberry.type
class Query:
@strawberry.field
async def me(self, info: strawberry.Info) -> Optional[User]:
user_id = info.context.get('user_id')
if not user_id:
raise strawberry.errors.GraphQLError("Not authenticated")
db = info.context['db']
# Fetch user...
return User(id=user_id, name="Alice")
schema = strawberry.Schema(query=Query)
# FastAPI.
app = FastAPI(title="API", version="1.0.0")
# Middleware.
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["GET", "POST"],
allow_headers=["*"],
)
# Routes.
graphql_app = GraphQL(schema, context_getter=get_context)
app.add_route("/graphql", graphql_app)
app.add_websocket_route("/graphql", graphql_app)
@app.get("/health")
async def health():
return {"status": "ok"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Key Takeaways
- Mount Strawberry GraphQL on a FastAPI app with
GraphQL(schema)andapp.add_route(). - Share database sessions, auth logic, and config between REST and GraphQL endpoints.
- Use custom exceptions to provide consistent error responses to both REST and GraphQL clients.
- FastAPI middleware (CORS, logging, rate limiting) applies to all routes.
- Choose GraphQL for flexible client queries; use REST for simple, well-defined endpoints.
Frequently Asked Questions
Can I have both /api/v1/users (REST) and /graphql (GraphQL) on the same server?
Yes. They coexist on the same FastAPI app. Document which protocol to use for which use case.
How do I serve the GraphQL playground (GraphiQL) in production?
Strawberry includes GraphiQL by default. For security, disable it in production by passing debug=False to the GraphQL constructor.
Can FastAPI middleware access GraphQL requests?
Yes. Middleware sees all requests, including GraphQL requests to /graphql. However, GraphQL uses POST bodies, so you can't inspect the query in middleware without parsing JSON.
Should I use FastAPI for just GraphQL?
Yes, FastAPI is lightweight and handles GraphQL well. Even if you only expose GraphQL, FastAPI provides validation, dependency injection, and middleware out of the box.