GraphQL Integration with Nexios and Strawberry: Building Flexible APIs
GraphQL has revolutionized how we build APIs, giving clients the power to request exactly what they need. When combined with Nexios’s high-performance ASGI architecture and Strawberry’s elegant Python integration, you get a powerful combination for building modern, flexible APIs.
Traditional REST APIs often suffer from over-fetching or under-fetching data. GraphQL solves this by allowing clients to specify their data requirements in a single query. Here’s why this combination shines:
- Type Safety: Strawberry provides automatic type generation from Python classes
- Performance: Nexios’s async-first architecture handles concurrent GraphQL requests efficiently
- Developer Experience: Built-in GraphiQL interface for API exploration
- Real-time Capabilities: GraphQL subscriptions work seamlessly with Nexios WebSockets
First, install the required packages:
pip install nexios nexios_contrib strawberry-graphqlLet’s start with a simple GraphQL server:
import strawberryfrom nexios import NexiosAppfrom nexios_contrib.graphql import GraphQL
@strawberry.typeclass Query: @strawberry.field def hello(self) -> str: return "Hello from Nexios GraphQL!"
@strawberry.field def version(self) -> str: return "3.5.0"
@strawberry.typeclass Mutation: @strawberry.mutation def say_hello(self, name: str) -> str: return f"Hello, {name}!"
schema = strawberry.Schema(query=Query, mutation=Mutation)app = NexiosApp()
# Add GraphQL endpoint at /graphql with GraphiQL interfaceGraphQL(app, schema)
if __name__ == "__main__": app.run()Visit http://localhost:8000/graphql to explore your API with the GraphiQL interface!
Let’s build a more practical example - a blog API with users, posts, and comments.
from typing import List, Optionalimport strawberryfrom datetime import datetime
@strawberry.typeclass User: id: int username: str email: str created_at: datetime
@strawberry.typeclass Comment: id: int content: str author: User created_at: datetime
@strawberry.typeclass Post: id: int title: str content: str author: User comments: List[Comment] created_at: datetime updated_at: datetimeNexios shines with async operations. Here’s how to implement async resolvers:
@strawberry.typeclass Query: @strawberry.field async def posts(self) -> List[Post]: # Simulate database query posts = await get_posts_from_db() return posts
@strawberry.field async def post(self, id: int) -> Optional[Post]: # Get single post by ID post = await get_post_by_id(id) return post
@strawberry.field async def user(self, id: int) -> Optional[User]: user = await get_user_by_id(id) return user
@strawberry.typeclass Mutation: @strawberry.mutation async def create_post( self, title: str, content: str, author_id: int ) -> Post: # Create new post post = await create_post_in_db(title, content, author_id) return post
@strawberry.mutation async def add_comment( self, post_id: int, content: str, author_id: int ) -> Comment: comment = await add_comment_to_post(post_id, content, author_id) return commentGraphQL resolvers often need access to request data like authentication information. Nexios provides this through the GraphQL context:
@strawberry.typeclass Query: @strawberry.field def current_user(self, info: strawberry.Info) -> Optional[User]: # Access request from context request = info.context["request"]
# Get user from authentication token auth_header = request.headers.get("Authorization") if auth_header and auth_header.startswith("Bearer "): token = auth_header[7:] user = await verify_jwt_token(token) return user
return None
@strawberry.field def user_agent(self, info: strawberry.Info) -> str: request = info.context["request"] return request.headers.get("user-agent", "Unknown")Define custom scalar types for specialized data:
@strawberry.scalarclass DateTime: @staticmethod def serialize(value: datetime) -> str: return value.isoformat()
@staticmethod def parse_value(value: str) -> datetime: return datetime.fromisoformat(value)
@strawberry.scalarclass Upload: # Handle file uploads passCombine GraphQL subscriptions with Nexios WebSockets for real-time features:
import asynciofrom typing import AsyncGenerator
@strawberry.typeclass Subscription: @strawberry.subscription async def post_updates(self) -> AsyncGenerator[Post, None]: # Stream real-time post updates while True: new_post = await wait_for_new_post() yield new_post await asyncio.sleep(1)
# Enable subscriptions in GraphQL setupGraphQL(app, schema, enable_subscriptions=True)Implement proper error handling in your resolvers:
from strawberry.types import GraphQLResolveInfo
@strawberry.fieldasync def post(self, id: int, info: GraphQLResolveInfo) -> Post: try: post = await get_post_by_id(id) if not post: raise strawberry.GraphQLError("Post not found") return post except DatabaseError as e: raise strawberry.GraphQLError(f"Database error: {str(e)}")Prevent N+1 query problems with DataLoader:
from strawberry.dataloader import DataLoader
async def load_users(keys: List[int]) -> List[User]: # Batch load users users = await get_users_by_ids(keys) return {user.id: user for user in users}
# Create dataloaderuser_loader = DataLoader(load_users)
@strawberry.typeclass Post: id: int title: str author_id: strawberry.Private[int]
@strawberry.field async def author(self, info: strawberry.Info) -> User: return await info.context["user_loader"].load(self.author_id)Implement caching for frequently accessed data:
from functools import lru_cacheimport redis
cache = redis.Redis()
@strawberry.fieldasync def popular_posts(self) -> List[Post]: # Check cache first cached = cache.get("popular_posts") if cached: return json.loads(cached)
# Fetch from database posts = await get_popular_posts()
# Cache for 5 minutes cache.setex("popular_posts", 300, json.dumps(posts)) return postsImplement JWT-based authentication:
import jwtfrom strawberry.types import Info
def get_current_user(info: Info) -> User: request = info.context["request"] auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "): raise strawberry.GraphQLError("Authentication required")
token = auth_header[7:] try: payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"]) user_id = payload["user_id"] return get_user_by_id(user_id) except jwt.InvalidTokenError: raise strawberry.GraphQLError("Invalid token")
@strawberry.typeclass Mutation: @strawberry.mutation async def create_post( self, title: str, content: str, info: Info ) -> Post: user = get_current_user(info) return await create_post(title, content, user.id)Add role-based access control:
def require_role(required_role: str): def decorator(func): async def wrapper(self, info: Info, *args, **kwargs): user = get_current_user(info) if user.role != required_role: raise strawberry.GraphQLError("Insufficient permissions") return await func(self, info, *args, **kwargs) return wrapper return decorator
@strawberry.typeclass Mutation: @strawberry.mutation @require_role("admin") async def delete_post(self, post_id: int, info: Info) -> bool: return await delete_post_by_id(post_id)Test your resolvers individually:
import pytestfrom strawberry.testing import GraphQLTestClient
@pytest.mark.asyncioasync def test_create_post(): schema = strawberry.Schema(query=Query, mutation=Mutation) client = GraphQLTestClient(schema)
result = await client.query( """ mutation { createPost(title: "Test Post", content: "Test content", authorId: 1) { id title author { username } } } """ )
assert result.data["createPost"]["title"] == "Test Post"Test the full Nexios integration:
from nexios.testing import TestClient
@pytest.mark.asyncioasync def test_graphql_endpoint(): app = create_app() # Your app creation function
async with TestClient(app) as client: response = await client.post( "/graphql", json={ "query": """ query { posts { id title } } """ } )
assert response.status_code == 200 data = response.json() assert "data" in data- Use specific types: Avoid generic JSON objects
- Implement pagination: Use cursor-based pagination for large datasets
- Version your schema: Use namespaces for different API versions
- Document your schema: Use descriptions for types and fields
@strawberry.type(description="A blog post with author and comments")class Post: id: strawberry.ID(description="Unique identifier") title: str(description="Post title") content: str(description="Post content in markdown")
@strawberry.field(description="Post author information") async def author(self) -> User: return await get_author(self.author_id)- Use async resolvers: Leverage Nexios’s async architecture
- Implement DataLoader: Prevent N+1 query problems
- Add caching: Cache expensive operations
- Monitor query complexity: Prevent expensive queries
@strawberry.fieldasync def posts( self, limit: int = 10, offset: int = 0) -> List[Post]: # Validate pagination parameters if limit > 100: raise strawberry.GraphQLError("Limit cannot exceed 100")
return await get_posts_with_pagination(limit, offset)Configure GraphQL for production:
# Disable GraphiQL in productionGraphQL( app, schema, graphiql=False, # Disable in production path="/api/graphql" # Custom endpoint)Add query monitoring and logging:
import loggingimport time
logger = logging.getLogger(__name__)
async def log_query_middleware(request, call_next): start_time = time.time()
# Log query details query = request.json.get("query", "") logger.info(f"GraphQL Query: {query[:100]}...")
response = await call_next()
# Log execution time duration = time.time() - start_time logger.info(f"Query executed in {duration:.3f}s")
return response
app.add_middleware(log_query_middleware)Combining GraphQL with Nexios and Strawberry gives you a powerful stack for building modern APIs:
- Type Safety: Automatic schema generation from Python types
- Performance: Async-first architecture handles concurrent requests efficiently
- Flexibility: Clients request exactly what they need
- Developer Experience: Built-in tools for exploration and testing
- Real-time: Subscriptions work seamlessly with WebSockets
This combination is perfect for applications that need flexible data access patterns, real-time updates, and high performance. Whether you’re building a blog API, e-commerce platform, or real-time dashboard, Nexios + GraphQL provides the tools you need.
Ready to start building? Check out the Nexios GraphQL documentation (opens in a new window) and explore the nexios-contrib (opens in a new window) package for more integrations.