Building APIs with GraphQL: Advanced Patterns and Complete Optimization Guide

# ❌ Without DataLoader - N+1 queries
def resolve_users(info):
    users = db.query("SELECT * FROM users")  # 1 query
    
    # For each user, get their posts - N queries!
    for user in users:
        user.posts = db.query(
            f"SELECT * FROM posts WHERE user_id = {user.id}"
        )
    
    return users

# Query: { users { name posts { title } } }
# Results in: 1 + N queries (where N = number of users)

from dataloader import DataLoader
import asyncio


class UserLoader(DataLoader):
    """Batch load users by IDs."""
    
    def __init__(self, db):
        super().__init__()
        self.db = db
    
    async def batch_load_fn(self, user_ids):
        """Batch load users in a single query."""
        # Single query for all users
        users = await self.db.query(
            "SELECT * FROM users WHERE id IN ($1)",
            user_ids
        )
        
        # Create lookup dictionary
        users_by_id = {user['id']: user for user in users}
        
        # Return in same order as requested
        return [users_by_id.get(uid) for uid in user_ids]


class PostLoader(DataLoader):
    """Batch load posts by user IDs."""
    
    def __init__(self, db):
        super().__init__()
        self.db = db
    
    async def batch_load_fn(self, user_ids):
        """Batch load posts for multiple users."""
        # Single query for all posts
        posts = await self.db.query(
            "SELECT * FROM posts WHERE user_id IN ($1)",
            user_ids
        )
        
        # Group posts by user_id
        posts_by_user = {}
        for post in posts:
            user_id = post['user_id']
            if user_id not in posts_by_user:
                posts_by_user[user_id] = []
            posts_by_user[user_id].append(post)
        
        # Return posts for each requested user
        return [posts_by_user.get(uid, []) for uid in user_ids]


class CommentLoader(DataLoader):
    """Batch load comments by post IDs."""
    
    def __init__(self, db):
        super().__init__()
        self.db = db
    
    async def batch_load_fn(self, post_ids):
        """Batch load comments for multiple posts."""
        comments = await self.db.query(
            "SELECT * FROM comments WHERE post_id IN ($1)",
            post_ids
        )
        
        comments_by_post = {}
        for comment in comments:
            post_id = comment['post_id']
            if post_id not in comments_by_post:
                comments_by_post[post_id] = []
            comments_by_post[post_id].append(comment)
        
        return [comments_by_post.get(pid, []) for pid in post_ids]


# Context setup with data loaders
class GraphQLContext:
    def __init__(self, db):
        self.db = db
        self.user_loader = UserLoader(db)
        self.post_loader = PostLoader(db)
        self.comment_loader = CommentLoader(db)


# Resolver using DataLoader
async def resolve_user(user, info, request_context):
    """Get user - uses cache within request."""
    return await request_context.user_loader.load(user.id)


async def resolve_user_posts(user, info, request_context):
    """Get user's posts - batched automatically."""
    return await request_context.post_loader.load(user.id)


async def resolve_post_comments(post, info, request_context):
    """Get post's comments - batched automatically."""
    return await request_context.comment_loader.load(post.id)


# Query: { users { name posts { title comments { text } } } }
# Results in: 1 user query + 1 posts query + 1 comments query = 3 queries
# vs. 1 + N + N*M queries without DataLoader!

class OptimizedUserLoader(DataLoader):
    """User loader with additional caching."""
    
    def __init__(self, db, cache={}):
        super().__init__()
        self.db = db
        self.request_cache = cache
    
    async def batch_load_fn(self, user_ids):
        # Check request cache first
        results = []
        uncached_ids = []
        
        for uid in user_ids:
            if uid in self.request_cache:
                results.append(self.request_cache[uid])
            else:
                uncached_ids.append(uid)
        
        # Load uncached users
        if uncached_ids:
            users = await self.db.query(
                "SELECT * FROM users WHERE id IN ($1)",
                uncached_ids
            )
            
            for user in users:
                self.request_cache[user['id']] = user
                results.append(user)
        
        return results

import asyncio
from aiohttp import web
from graphql import (
    GraphQLSchema,
    GraphQLObjectType,
    GraphQLString,
    GraphQLField,
    Subscription as GraphQLSubscription
)


# Define subscription type
QueryType = GraphQLObjectType(
    name='Query',
    fields={
        'hello': GraphQLField(GraphQLString)
    }
)

SubscriptionType = GraphQLObjectType(
    name='Subscription',
    fields={
        'userCreated': GraphQLField(
            GraphQLString,
            resolve=lambda root, info: root
        ),
        'postUpdated': GraphQLField(
            GraphQLString,
            args={'id': GraphQLString()},
            resolve=lambda root, info, id: f"Post {id} updated"
        )
    }
)

schema = GraphQLSchema(
    query=QueryType,
    subscription=SubscriptionType
)


# Pub/Sub implementation
class PubSub:
    """Simple pub/sub for subscriptions."""
    
    def __init__(self):
        self.subscriptions = {}
        self.subscription_id = 0
    
    def subscribe(self, event_name, callback):
        """Subscribe to an event."""
        if event_name not in self.subscriptions:
            self.subscriptions[event_name] = []
        
        sub_id = self.subscription_id
        self.subscriptions[event_name].append((sub_id, callback))
        self.subscription_id += 1
        
        return sub_id
    
    def unsubscribe(self, event_name, sub_id):
        """Unsubscribe from an event."""
        if event_name in self.subscriptions:
            self.subscriptions[event_name] = [
                (sid, cb) for sid, cb in self.subscriptions[event_name]
                if sid != sub_id
            ]
    
    async def publish(self, event_name, payload):
        """Publish an event."""
        if event_name in self.subscriptions:
            for sub_id, callback in self.subscriptions[event_name]:
                await callback(payload)


pubsub = PubSub()


# Async iterator for subscription
async def on_user_created(root, info):
    """Stream user created events."""
    queue = asyncio.Queue()
    
    async def handler(user):
        await queue.put(user)
    
    sub_id = pubsub.subscribe('user_created', handler)
    
    try:
        while True:
            user = await queue.get()
            yield user
    finally:
        pubsub.unsubscribe('user_created', sub_id)


# Trigger events
async def create_user(user_data):
    user = await save_user(user_data)
    await pubsub.publish('user_created', user)
    return user

from aiohttp import web
from aiohttp_wsgi import WSGIHandler
import graphql
from graphql import subscribe


async def websocket_handler(request):
    """Handle GraphQL WebSocket subscriptions."""
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    
    # Send connection ack
    await ws.send_json({'type': 'connection_ack'})
    
    async for msg in ws:
        if msg.type == web.WSMsgType.TEXT:
            data = msg.json()
            
            if data.get('type') == 'start':
                # Parse and execute subscription
                payload = data.get('payload', {})
                query = payload.get('query')
                variables = payload.get('variables', {})
                
                # Execute subscription
                result = await subscribe(
                    graphql.build_schema(schema),
                    query,
                    root_value=None,
                    variable_values=variables
                )
                
                if hasattr(result, '__anext__'):
                    # It's an async iterator
                    try:
                        while True:
                            chunk = await result.__anext__()
                            await ws.send_json({
                                'id': data.get('id'),
                                'type': 'data',
                                'payload': chunk.data
                            })
                    except StopAsyncIteration:
                        pass
                        
        elif msg.type == web.WSMsgType.ERROR:
            print(f'WebSocket error: {ws.exception()}')
    
    return ws


app = web.Application()
app.router.add_get('/subscriptions', websocket_handler)

import strawberry
from strawberry import federated


@strawberry.federation.type(keys=["id"])
class User:
    id: strawberry.ID
    username: str
    email: str
    
    @strawberry.field
    async def posts(self) -> list["Post"]:
        """Get user's posts from Posts service."""
        return await get_posts_by_user(self.id)


@strawberry.federation.type(keys=["id"])
class Post:
    id: strawberry.ID
    title: str
    content: str


@strawberry.type
class Query:
    @strawberry.federation.field(extensions=[strawberry.federation.field_extension_query])
    async def me(self, info: strawberry.Info) -> User:
        """Get current user."""
        user_id = info.context.get('user_id')
        return await get_user_by_id(user_id)
    
    @strawberry.federation.field
    async def user(self, id: strawberry.ID) -> User:
        """Get user by ID."""
        return await get_user_by_id(id)


schema = strawberry.federated.Schema(query=Query, types=[User, Post])

import strawberry
from strawberry import federated


@strawberry.federation.type(keys=["id"])
class Product:
    id: strawberry.ID
    name: str
    price: float
    in_stock: bool


@strawberry.type
class Query:
    @strawberry.federation.field
    async def product(self, id: strawberry.ID) -> Product:
        return await get_product_by_id(id)
    
    @strawberry.federation.field
    async def products(self) -> list[Product]:
        return await get_all_products()


schema = strawberry.federated.Schema(query=Query, types=[Product])

# gateway.config.js
const { ApolloGateway } = require('@apollo/gateway');
const gateway = new ApolloGateway({
  serviceList: [
    { name: 'users', url: 'http://users-service:4000/graphql' },
    { name: 'orders', url: 'http://orders-service:4000/graphql' },
    { name: 'products', url: 'http://products-service:4000/graphql' }
  ],
  // Experimental: Use improved service health checking
  experimental_useServiceHealthCheck: true
});

import hashlib
from typing import Any


class QueryComplexityAnalyzer:
    """Analyze and limit GraphQL query complexity."""
    
    def __init__(
        self,
        max_depth: int = 10,
        max_complexity: int = 1000,
        complexity_weights: dict = None
    ):
        self.max_depth = max_depth
        self.max_complexity = max_complexity
        self.complexity_weights = complexity_weights or {
            'object': 1,
            'list': 5,
            'connection': 10,
            'scalar': 1
        }
    
    def analyze(self, query: str) -> dict:
        """Analyze query complexity."""
        try:
            from graphql import parse, visit
            
            document = parse(query)
            
            depth = 0
            complexity = 0
            violations = []
            
            class ComplexityVisitor:
                def __init__(self):
                    self.depth = 0
                    self.max_depth = 0
                    self.complexity = 0
                
                def enter(self, node, *args):
                    if hasattr(node, 'selection_set'):
                        self.depth += 1
                        self.max_depth = max(self.max_depth, self.depth)
                        
                        # Count selections
                        if node.selection_set:
                            selections = node.selection_set.selections
                            for sel in selections:
                                if hasattr(sel, 'selection_set'):
                                    # It's a field with sub-selections
                                    self.complexity += 10
                                else:
                                    # Simple field
                                    self.complexity += 1
                
                def leave(self, node, *args):
                    if hasattr(node, 'selection_set'):
                        self.depth -= 1
            
            visitor = ComplexityVisitor()
            visit(document, {'enter': visitor.enter, 'leave': visitor.leave})
            
            return {
                'valid': visitor.complexity <= self.max_complexity and visitor.max_depth <= self.max_depth,
                'complexity': visitor.complexity,
                'depth': visitor.max_depth,
                'max_complexity': self.max_complexity,
                'max_depth': self.max_depth
            }
            
        except Exception as e:
            return {
                'valid': False,
                'error': str(e)
            }


# Usage as middleware
class ComplexityValidationMiddleware:
    """Middleware to validate query complexity."""
    
    def __init__(self, analyzer: QueryComplexityAnalyzer):
        self.analyzer = analyzer
    
    async def resolve(self, next, root, info, **args):
        query = info.field_nodes[0].loc.source.body if info.field_nodes else ''
        
        result = self.analyzer.analyze(query)
        
        if not result.get('valid', False):
            raise Exception(
                f"Query too complex: {result.get('complexity', 0)}/{result.get('max_complexity')}"
            )
        
        return await next(root, info, **args)

import hashlib
import json
import redis
from typing import Optional, Any


class GraphQLResponseCache:
    """Redis-based GraphQL response cache."""
    
    def __init__(
        self,
        redis_client: redis.Redis,
        default_ttl: int = 300,
        cache_key_prefix: str = 'graphql:cache:'
    ):
        self.redis = redis_client
        self.default_ttl = default_ttl
        self.prefix = cache_key_prefix
    
    def make_key(
        self,
        query: str,
        variables: dict = None,
        operation_name: str = None
    ) -> str:
        """Generate cache key from query and variables."""
        key_data = {
            'query': query,
            'variables': variables or {},
            'operation_name': operation_name
        }
        
        key_hash = hashlib.sha256(
            json.dumps(key_data, sort_keys=True).encode()
        ).hexdigest()
        
        return f"{self.prefix}{key_hash}"
    
    def get(
        self,
        query: str,
        variables: dict = None,
        operation_name: str = None
    ) -> Optional[dict]:
        """Get cached response."""
        key = self.make_key(query, variables, operation_name)
        
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)
        
        return None
    
    def set(
        self,
        query: str,
        data: dict,
        variables: dict = None,
        operation_name: str = None,
        ttl: int = None
    ) -> bool:
        """Cache response."""
        key = self.make_key(query, variables, operation_name)
        ttl = ttl or self.default_ttl
        
        return self.redis.setex(
            key,
            ttl,
            json.dumps(data)
        )
    
    def invalidate(self, pattern: str) -> int:
        """Invalidate cache entries matching pattern."""
        keys = self.redis.keys(f"{self.prefix}{pattern}")
        
        if keys:
            return self.redis.delete(*keys)
        
        return 0


# Cache key normalization
class NormalizingCache(GraphQLResponseCache):
    """Cache that normalizes queries before caching."""
    
    def make_key(self, query: str, variables: dict = None, operation_name: str = None) -> str:
        # Remove whitespace and normalize
        normalized = ' '.join(query.split())
        
        # Remove comments
        import re
        normalized = re.sub(r'#.*$', '', normalized, flags=re.MULTILINE)
        
        return super().make_key(normalized, variables, operation_name)

# Register persisted query
async def register_persisted_query(query_id: str, query: str):
    """Register a query for persisted queries."""
    await redis.set(
        f"pq:{query_id}",
        query,
        ex=None  # No expiration
    )

# Execute persisted query
async def execute_persisted_query(query_id: str, variables: dict = None):
    """Execute a persisted query."""
    # Get query from cache
    query = await redis.get(f"pq:{query_id}")
    
    if not query:
        raise Exception("Persisted query not found")
    
    # Execute
    return await execute_query(query.decode(), variables)


# Apollo persisted queries
# In Apollo Server:
const apolloServer = new ApolloServer({
    plugins: [
        ApolloServerPluginPersistedQueries({
            cache: new MyCustomMap(),
            generateHash: ({ query }) => hash(query)
        })
    ]
})

class DepthLimitRule:
    """Rule to limit query depth."""
    
    def __init__(self, max_depth: int = 10):
        self.max_depth = max_depth
    
    def __call__(self, document):
        errors = []
        
        def visit(node, depth=0):
            if depth > self.max_depth:
                errors.append(f"Query exceeds max depth of {self.max_depth}")
                return
            
            for field in node.get('selectionSet', {}).get('selections', []):
                visit(field, depth + 1)
        
        for definition in document.get('definitions', []):
            if definition.get('kind') == 'OperationDefinition':
                for selection in definition.get('selectionSet', {}).get('selections', []):
                    visit(selection, 1)
        
        return errors


# Use with validation
from graphql import validate
from graphql.validation.rules import DepthLimit

errors = validate(
    schema,
    query,
    rules=[DepthLimit(max_depth=10)]
)

# Disable introspection in production
def disable_introspection_middleware(next, root, info, **args):
    if info.field_name == '__schema' or info.field_name == '__type':
        raise Exception("Introspection disabled")
    
    return next(root, info, **args)