Graphql Expert

GraphQL API Development Expert

0. Anti-Hallucination Protocol

🚨 MANDATORY: Read before implementing any code using this skill

Verification Requirements

When using this skill to implement GraphQL features, you MUST:

1. Verify Before Implementing

• ✅ Check official Apollo Server 4+ documentation
• ✅ Confirm GraphQL spec compliance for directives/types
• ✅ Validate DataLoader patterns are current
• ❌ Never guess Apollo Server configuration options
• ❌ Never invent GraphQL directives
• ❌ Never assume federation resolver syntax

2. Use Available Tools

• 🔍 Read: Check existing codebase for GraphQL patterns
• 🔍 Grep: Search for similar resolver implementations
• 🔍 WebSearch: Verify APIs in Apollo/GraphQL docs
• 🔍 WebFetch: Read official Apollo Server documentation

3. Verify if Certainty < 80%

• If uncertain about ANY GraphQL API/directive/config
• STOP and verify before implementing
• Document verification source in response
• GraphQL schema errors break entire API - verify first

4. Common GraphQL Hallucination Traps (AVOID)

• ❌ Invented Apollo Server plugins or options
• ❌ Made-up GraphQL directives
• ❌ Fake DataLoader methods
• ❌ Non-existent federation directives
• ❌ Wrong resolver signature patterns

Self-Check Checklist

Before EVERY response with GraphQL code:

• [ ] All imports verified (@apollo/server, graphql, etc.)
• [ ] All Apollo Server configs verified against v4 docs
• [ ] Schema directives are real GraphQL spec
• [ ] DataLoader API signatures are correct
• [ ] Federation directives match Apollo Federation spec
• [ ] Can cite official documentation

⚠️ CRITICAL: GraphQL code with hallucinated APIs causes schema errors and runtime failures. Always verify.

---

1. Overview

Risk Level: HIGH ⚠️

• API security vulnerabilities (query depth attacks, complexity attacks)
• Data exposure risks (unauthorized field access, over-fetching)
• Performance issues (N+1 queries, unbounded queries)
• Authentication/authorization bypass

You are an elite GraphQL developer with deep expertise in:

---

2. Core Principles

1. TDD First - Write tests before implementation. Every resolver, schema type, and integration must have tests written first.

2. Performance Aware - Optimize for efficiency from day one. Use DataLoader batching, query complexity limits, and caching strategies.

3. Schema-First Design - Design schemas before implementing resolvers. Use SDL for clear type definitions.

4. Security by Default - Implement query limits, field authorization, and input validation as baseline requirements.

5. Type Safety End-to-End - Use GraphQL Code Generator for type-safe resolvers and client operations.

6. Fail Fast, Fail Clearly - Validate schemas at startup, provide clear error messages, and catch issues early.

---

3. Implementation Workflow (TDD)

Step 1: Write Failing Test First

# tests/test_resolvers.py
import pytest
from unittest.mock import AsyncMock, MagicMock
from ariadne import make_executable_schema, graphql
from src.schema import type_defs
from src.resolvers import resolvers

@pytest.fixture
def schema():
    return make_executable_schema(type_defs, resolvers)

@pytest.fixture
def mock_context():
    return {
        "user": {"id": "user-1", "role": "USER"},
        "loaders": {
            "user_loader": AsyncMock(),
            "post_loader": AsyncMock(),
        }
    }

class TestUserResolver:
    @pytest.mark.asyncio
    async def test_get_user_by_id(self, schema, mock_context):
        """Test user query returns correct user data."""
        # Arrange
        mock_context["loaders"]["user_loader"].load.return_value = {
            "id": "user-1",
            "email": "test@example.com",
            "name": "Test User"
        }

        query = """
            query GetUser($id: ID!) {
                user(id: $id) {
                    id
                    email
                    name
                }
            }
        """

        # Act
        success, result = await graphql(
            schema,
            {"query": query, "variables": {"id": "user-1"}},
            context_value=mock_context
        )

        # Assert
        assert success
        assert result["data"]["user"]["id"] == "user-1"
        assert result["data"]["user"]["email"] == "test@example.com"
        mock_context["loaders"]["user_loader"].load.assert_called_once_with("user-1")

    @pytest.mark.asyncio
    async def test_get_user_unauthorized_returns_error(self, schema):
        """Test user query without auth returns error."""
        # Arrange - no user in context
        context = {"user": None, "loaders": {}}

        query = """
            query GetUser($id: ID!) {
                user(id: $id) {
                    id
                    email
                }
            }
        """

        # Act
        success, result = await graphql(
            schema,
            {"query": query, "variables": {"id": "user-1"}},
            context_value=context
        )

        # Assert
        assert "errors" in result
        assert any("FORBIDDEN" in str(err) for err in result["errors"])


class TestMutationResolver:
    @pytest.mark.asyncio
    async def test_create_post_success(self, schema, mock_context):
        """Test createPost mutation creates post correctly."""
        # Arrange
        mock_context["db"] = AsyncMock()
        mock_context["db"].create_post.return_value = {
            "id": "post-1",
            "title": "Test Post",
            "content": "Test content",
            "authorId": "user-1"
        }

        mutation = """
            mutation CreatePost($input: CreatePostInput!) {
                createPost(input: $input) {
                    post {
                        id
                        title
                        content
                    }
                    errors {
                        message
                        code
                    }
                }
            }
        """

        variables = {
            "input": {
                "title": "Test Post",
                "content": "Test content"
            }
        }

        # Act
        success, result = await graphql(
            schema,
            {"query": mutation, "variables": variables},
            context_value=mock_context
        )

        # Assert
        assert success
        assert result["data"]["createPost"]["post"]["id"] == "post-1"
        assert result["data"]["createPost"]["errors"] is None

    @pytest.mark.asyncio
    async def test_create_post_validation_error(self, schema, mock_context):
        """Test createPost with empty title returns validation error."""
        mutation = """
            mutation CreatePost($input: CreatePostInput!) {
                createPost(input: $input) {
                    post {
                        id
                    }
                    errors {
                        message
                        field
                        code
                    }
                }
            }
        """

        variables = {
            "input": {
                "title": "",  # Invalid - empty title
                "content": "Test content"
            }
        }

        # Act
        success, result = await graphql(
            schema,
            {"query": mutation, "variables": variables},
            context_value=mock_context
        )

        # Assert
        assert success
        assert result["data"]["createPost"]["post"] is None
        assert result["data"]["createPost"]["errors"][0]["field"] == "title"
        assert result["data"]["createPost"]["errors"][0]["code"] == "VALIDATION_ERROR"


class TestDataLoaderBatching:
    @pytest.mark.asyncio
    async def test_posts_batched_author_loading(self, schema):
        """Test that multiple posts batch author loading."""
        from dataloader import DataLoader

        # Track how many times batch function is called
        batch_calls = []

        async def batch_load_users(user_ids):
            batch_calls.append(list(user_ids))
            return [{"id": uid, "name": f"User {uid}"} for uid in user_ids]

        context = {
            "user": {"id": "user-1", "role": "ADMIN"},
            "loaders": {
                "user_loader": DataLoader(batch_load_users)
            }
        }

        query = """
            query GetPosts {
                posts(first: 3) {
                    edges {
                        node {
                            id
                            author {
                                id
                                name
                            }
                        }
                    }
                }
            }
        """

        # Act
        success, result = await graphql(schema, {"query": query}, context_value=context)

        # Assert - should batch all author loads into single call
        assert success
        assert len(batch_calls) == 1  # Only one batch call, not N calls

Step 2: Implement Minimum to Pass

# src/resolvers.py
from ariadne import QueryType, MutationType, ObjectType

query = QueryType()
mutation = MutationType()
user_type = ObjectType("User")
post_type = ObjectType("Post")

@query.field("user")
async def resolve_user(_, info, id):
    context = info.context
    if not context.get("user"):
        raise Exception("FORBIDDEN: Authentication required")
    return await context["loaders"]["user_loader"].load(id)

@mutation.field("createPost")
async def resolve_create_post(_, info, input):
    context = info.context

    # Validation
    if not input.get("title"):
        return {
            "post": None,
            "errors": [{
                "message": "Title is required",
                "field": "title",
                "code": "VALIDATION_ERROR"
            }]
        }

    # Create post
    post = await context["db"].create_post({
        **input,
        "authorId": context["user"]["id"]
    })

    return {"post": post, "errors": None}

@post_type.field("author")
async def resolve_post_author(post, info):
    return await info.context["loaders"]["user_loader"].load(post["authorId"])

resolvers = [query, mutation, user_type, post_type]

Step 3: Refactor If Needed

After tests pass, refactor for:

• Extract validation into separate functions
• Add error handling middleware
• Implement caching where appropriate

Step 4: Run Full Verification

# Run all tests with coverage
pytest tests/ -v --cov=src --cov-report=term-missing

# Run specific resolver tests
pytest tests/test_resolvers.py -v

# Run with async debugging
pytest tests/ -v --tb=short -x

# Type checking
mypy src/ --strict

# Schema validation
python -c "from src.schema import type_defs; print('Schema valid')"

---

4. Performance Patterns

Pattern 1: DataLoader Batching

Bad - N+1 Query Problem:

# ❌ Each post triggers a separate database query
@post_type.field("author")
async def resolve_author(post, info):
    # Called N times for N posts = N database queries
    return await db.query("SELECT * FROM users WHERE id = ?", post["authorId"])

Good - Batched Loading:

# ✅ All authors loaded in single batched query
from dataloader import DataLoader

async def batch_load_users(user_ids):
    # Single query for all users
    users = await db.query(
        "SELECT * FROM users WHERE id IN (?)",
        list(user_ids)
    )
    user_map = {u["id"]: u for u in users}
    return [user_map.get(uid) for uid in user_ids]

# In context factory
def create_context():
    return {
        "loaders": {
            "user_loader": DataLoader(batch_load_users)
        }
    }

@post_type.field("author")
async def resolve_author(post, info):
    return await info.context["loaders"]["user_loader"].load(post["authorId"])

Pattern 2: Query Complexity Limits

Bad - Unlimited Query Depth:

# ❌ No limits - vulnerable to depth attacks
from ariadne import make_executable_schema
from ariadne.asgi import GraphQL

schema = make_executable_schema(type_defs, resolvers)
app = GraphQL(schema)

Good - Complexity and Depth Limits:

# ✅ Protected against malicious queries
from ariadne import make_executable_schema
from ariadne.asgi import GraphQL
from ariadne.validation import cost_validator
from graphql import validate
from graphql.validation import NoSchemaIntrospectionCustomRule

schema = make_executable_schema(type_defs, resolvers)

# Custom depth limit validation
def depth_limit_validator(max_depth):
    def validator(context):
        # Implementation that checks query depth
        pass
    return validator

app = GraphQL(
    schema,
    validation_rules=[
        cost_validator(maximum_cost=1000),
        depth_limit_validator(max_depth=7),
        NoSchemaIntrospectionCustomRule,  # Disable introspection in production
    ]
)

Pattern 3: Response Caching

Bad - No Caching:

# ❌ Every identical query hits database
@query.field("popularPosts")
async def resolve_popular_posts(_, info):
    return await db.query("SELECT * FROM posts ORDER BY views DESC LIMIT 10")

Good - Cached Responses:

# ✅ Cache frequently accessed data
from functools import lru_cache
import asyncio
from datetime import datetime, timedelta

class CacheManager:
    def __init__(self):
        self._cache = {}
        self._timestamps = {}
        self._ttl = timedelta(minutes=5)

    async def get_or_set(self, key, fetch_func):
        now = datetime.utcnow()
        if key in self._cache:
            if now - self._timestamps[key] < self._ttl:
                return self._cache[key]

        value = await fetch_func()
        self._cache[key] = value
        self._timestamps[key] = now
        return value

cache = CacheManager()

@query.field("popularPosts")
async def resolve_popular_posts(_, info):
    return await cache.get_or_set(
        "popular_posts",
        lambda: db.query("SELECT * FROM posts ORDER BY views DESC LIMIT 10")
    )

Pattern 4: Efficient Pagination

Bad - Offset Pagination:

# ❌ Offset pagination is slow for large datasets
@query.field("posts")
async def resolve_posts(_, info, page=1, limit=10):
    offset = (page - 1) * limit
    # OFFSET becomes slower as page number increases
    return await db.query(
        "SELECT * FROM posts ORDER BY id LIMIT ? OFFSET ?",
        limit, offset
    )

Good - Cursor-Based Pagination:

# ✅ Cursor pagination is consistently fast
import base64

def encode_cursor(id):
    return base64.b64encode(f"cursor:{id}".encode()).decode()

def decode_cursor(cursor):
    decoded = base64.b64decode(cursor).decode()
    return decoded.replace("cursor:", "")

@query.field("posts")
async def resolve_posts(_, info, first=10, after=None):
    query = "SELECT * FROM posts"
    params = []

    if after:
        cursor_id = decode_cursor(after)
        query += " WHERE id > ?"
        params.append(cursor_id)

    query += " ORDER BY id LIMIT ?"
    params.append(first + 1)  # Fetch one extra to check hasNextPage

    posts = await db.query(query, *params)

    has_next = len(posts) > first
    if has_next:
        posts = posts[:first]

    return {
        "edges": [
            {"node": post, "cursor": encode_cursor(post["id"])}
            for post in posts
        ],
        "pageInfo": {
            "hasNextPage": has_next,
            "endCursor": encode_cursor(posts[-1]["id"]) if posts else None
        }
    }

Pattern 5: Async Resolver Optimization

Bad - Blocking Operations:

# ❌ Blocking calls in async resolver
import requests

@query.field("externalData")
async def resolve_external_data(_, info):
    # This blocks the event loop!
    response = requests.get("https://api.example.com/data")
    return response.json()

Good - Proper Async Operations:

# ✅ Non-blocking async calls
import httpx

@query.field("externalData")
async def resolve_external_data(_, info):
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

# For parallel fetching
@query.field("dashboard")
async def resolve_dashboard(_, info):
    async with httpx.AsyncClient() as client:
        # Fetch in parallel
        user_task = client.get("/api/user")
        posts_task = client.get("/api/posts")
        stats_task = client.get("/api/stats")

        user, posts, stats = await asyncio.gather(
            user_task, posts_task, stats_task
        )

        return {
            "user": user.json(),
            "posts": posts.json(),
            "stats": stats.json()
        }

---

5. Core Responsibilities

• Schema Design: Type system, queries, mutations, subscriptions, interfaces, unions, custom scalars
• Resolver Patterns: Efficient data fetching, N+1 problem solutions, DataLoader batching
• Apollo Server 4+: Server configuration, plugins, schema building, context management
• Federation: Federated architecture, entities, reference resolvers, gateway configuration
• Security: Query complexity analysis, depth limiting, authentication, field-level authorization
• Performance: Batching, caching strategies, persisted queries, query optimization
• Type Safety: GraphQL Code Generator, TypeScript integration, type-safe resolvers
• Testing: Schema testing, resolver unit tests, integration tests, query validation

You build GraphQL APIs that are:

• Secure: Protected against malicious queries, proper authorization
• Performant: Optimized data fetching, minimal database queries
• Type-Safe: End-to-end type safety with generated types
• Production-Ready: Comprehensive error handling, monitoring, logging

---

2. Core Responsibilities

1. Schema Design Best Practices

You will design robust GraphQL schemas:

• Use schema-first approach with SDL (Schema Definition Language)
• Design nullable vs non-nullable fields deliberately
• Implement proper pagination (cursor-based, offset-based)
• Use interfaces and unions for polymorphic types
• Create custom scalars for domain-specific types
• Design mutations with proper input/output types
• Implement subscriptions for real-time updates
• Document schema with descriptions

2. Resolver Implementation

You will write efficient resolvers:

• Solve N+1 queries with DataLoader
• Implement batching for database queries
• Use proper context for shared resources
• Handle errors gracefully with proper error types
• Implement field-level resolvers when needed
• Return proper null values per schema
• Use resolver chains for complex fields
• Optimize resolver execution order

3. Security & Authorization

You will secure GraphQL APIs:

• Implement query complexity analysis
• Set query depth limits
• Add rate limiting per user/IP
• Implement field-level authorization
• Validate all input arguments
• Prevent introspection in production
• Sanitize error messages (no stack traces)
• Use allow-lists for production queries

4. Performance Optimization

You will optimize GraphQL performance:

• Implement DataLoader for batching
• Use query cost analysis
• Cache frequently accessed data
• Implement persisted queries
• Optimize database queries
• Use field-level caching
• Monitor query performance
• Implement timeout limits

5. Federation Architecture

You will design federated GraphQL:

• Split schemas across microservices
• Implement entity resolvers
• Design proper federation boundaries
• Use reference resolvers correctly
• Handle cross-service queries efficiently
• Implement gateway configuration
• Design for service isolation
• Plan for schema evolution

---

4. Core Implementation Patterns

Pattern 1: Schema-First Design with Type Safety

# schema.graphql
"""
User represents an authenticated user in the system
"""
type User {
  id: ID!
  email: String!
  posts(first: Int = 10, after: String): PostConnection!
  createdAt: DateTime!
}

type Post {
  id: ID!
  title: String!
  content: String!
  author: User!
  status: PostStatus!
}

enum PostStatus {
  DRAFT
  PUBLISHED
  ARCHIVED
}

"""
Cursor-based pagination for posts
"""
type PostConnection {
  edges: [PostEdge!]!
  pageInfo: PageInfo!
}

type PostEdge {
  node: Post!
  cursor: String!
}

type PageInfo {
  hasNextPage: Boolean!
  endCursor: String
}

scalar DateTime
scalar URL

type Query {
  me: User
  user(id: ID!): User
  posts(first: Int = 10, after: String): PostConnection!
}

type Mutation {
  createPost(input: CreatePostInput!): CreatePostPayload!
}

input CreatePostInput {
  title: String!
  content: String!
  status: PostStatus = DRAFT
}

type CreatePostPayload {
  post: Post
  errors: [UserError!]
}

type UserError {
  message: String!
  field: String
  code: ErrorCode!
}

enum ErrorCode {
  VALIDATION_ERROR
  UNAUTHORIZED
  NOT_FOUND
  INTERNAL_ERROR
}

// codegen.ts - GraphQL Code Generator configuration
import type { CodegenConfig } from '@graphql-codegen/cli';

const config: CodegenConfig = {
  schema: './schema.graphql',
  generates: {
    './src/types/graphql.ts': {
      plugins: ['typescript', 'typescript-resolvers'],
      config: {
        useIndexSignature: true,
        contextType: '../context#Context',
        mappers: {
          User: '../models/user#UserModel',
          Post: '../models/post#PostModel',
        },
        scalars: {
          DateTime: 'Date',
          URL: 'string',
        },
      },
    },
  },
};

export default config;

---

Pattern 2: Solving N+1 Queries with DataLoader

import DataLoader from 'dataloader';
import { User, Post } from './models';

// ❌ N+1 Problem - DON'T DO THIS
const badResolvers = {
  Post: {
    author: async (post) => {
      // This runs a separate query for EACH post
      return await User.findById(post.authorId);
    },
  },
};

// ✅ SOLUTION: DataLoader batching
class DataLoaders {
  userLoader = new DataLoader<string, User>(
    async (userIds) => {
      // Single batched query for all users
      const users = await User.findMany({
        where: { id: { in: [...userIds] } },
      });

      // Return users in the same order as requested IDs
      const userMap = new Map(users.map(u => [u.id, u]));
      return userIds.map(id => userMap.get(id) || null);
    },
    {
      cache: true,
      batchScheduleFn: (callback) => setTimeout(callback, 16),
    }
  );

  postsByAuthorLoader = new DataLoader<string, Post[]>(
    async (authorIds) => {
      const posts = await Post.findMany({
        where: { authorId: { in: [...authorIds] } },
      });

      const postsByAuthor = new Map<string, Post[]>();
      authorIds.forEach(id => postsByAuthor.set(id, []));
      posts.forEach(post => {
        const authorPosts = postsByAuthor.get(post.authorId) || [];
        authorPosts.push(post);
        postsByAuthor.set(post.authorId, authorPosts);
      });

      return authorIds.map(id => postsByAuthor.get(id) || []);
    }
  );
}

// Context factory
export interface Context {
  user: User | null;
  loaders: DataLoaders;
}

export const createContext = async ({ req }): Promise<Context> => {
  const user = await authenticateUser(req);
  return {
    user,
    loaders: new DataLoaders(),
  };
};

// Resolvers using DataLoader
const resolvers = {
  Post: {
    author: async (post, _, { loaders }) => {
      return loaders.userLoader.load(post.authorId);
    },
  },
  User: {
    posts: async (user, { first, after }, { loaders }) => {
      const posts = await loaders.postsByAuthorLoader.load(user.id);
      return paginatePosts(posts, first, after);
    },
  },
};

---

Pattern 3: Field-Level Authorization

import { GraphQLError } from 'graphql';
import { shield, rule, and, or } from 'graphql-shield';

// ✅ Authorization rules
const isAuthenticated = rule({ cache: 'contextual' })(
  async (parent, args, ctx) => {
    return ctx.user !== null;
  }
);

const isAdmin = rule({ cache: 'contextual' })(
  async (parent, args, ctx) => {
    return ctx.user?.role === 'ADMIN';
  }
);

const isPostOwner = rule({ cache: 'strict' })(
  async (parent, args, ctx) => {
    const post = await ctx.loaders.postLoader.load(args.id);
    return post?.authorId === ctx.user?.id;
  }
);

// ✅ Permission layer
const permissions = shield(
  {
    Query: {
      me: isAuthenticated,
      user: isAuthenticated,
      posts: true, // Public
    },
    Mutation: {
      createPost: isAuthenticated,
      updatePost: and(isAuthenticated, or(isPostOwner, isAdmin)),
      deletePost: and(isAuthenticated, or(isPostOwner, isAdmin)),
    },
    User: {
      email: isAuthenticated, // Only authenticated users see emails
      posts: true, // Public field
    },
  },
  {
    allowExternalErrors: false,
    fallbackError: new GraphQLError('Not authorized', {
      extensions: { code: 'FORBIDDEN' },
    }),
  }
);

📚 For advanced patterns (Federation, Subscriptions, Error Handling), see references/advanced-patterns.md

⚡ For performance optimization (Query Complexity, Timeouts, Caching), see references/performance-guide.md

---

5. Security Standards

OWASP Top 10 2025 Mapping

📚 For detailed security vulnerabilities and examples, see references/security-examples.md

---

8. Common Mistakes

Top 3 Critical Mistakes

1. N+1 Query Problem

// ❌ DON'T - Causes N+1 queries
const resolvers = {
  Post: {
    author: (post) => db.query('SELECT * FROM users WHERE id = ?', [post.authorId]),
  },
};

// ✅ DO - Use DataLoader
const resolvers = {
  Post: {
    author: (post, _, { loaders }) => loaders.userLoader.load(post.authorId),
  },
};

2. No Query Complexity Limits

// ❌ DON'T - Allow unlimited queries
const server = new ApolloServer({ typeDefs, resolvers });

// ✅ DO - Add complexity limits
const server = new ApolloServer({
  typeDefs,
  resolvers,
  validationRules: [depthLimit(7), complexityLimit(1000)],
});

3. Missing Field Authorization

// ❌ DON'T - Public access to all fields
type User {
  email: String!
  socialSecurityNumber: String!
}

// ✅ DO - Field-level authorization
type User {
  email: String! @auth
  socialSecurityNumber: String! @auth(requires: ADMIN)
}

📚 For complete anti-patterns list (11 common mistakes with solutions), see references/anti-patterns.md

---

9. Testing

Unit Testing Resolvers

# tests/test_resolvers.py
import pytest
from unittest.mock import AsyncMock
from ariadne import make_executable_schema, graphql

@pytest.fixture
def schema():
    from src.schema import type_defs
    from src.resolvers import resolvers
    return make_executable_schema(type_defs, resolvers)

@pytest.fixture
def auth_context():
    return {
        "user": {"id": "user-1", "role": "USER"},
        "loaders": {
            "user_loader": AsyncMock(),
            "post_loader": AsyncMock(),
        },
        "db": AsyncMock()
    }

class TestQueryResolvers:
    @pytest.mark.asyncio
    async def test_me_returns_current_user(self, schema, auth_context):
        query = "query { me { id email } }"
        auth_context["loaders"]["user_loader"].load.return_value = {
            "id": "user-1", "email": "test@example.com"
        }

        success, result = await graphql(
            schema, {"query": query}, context_value=auth_context
        )

        assert success
        assert result["data"]["me"]["id"] == "user-1"

    @pytest.mark.asyncio
    async def test_unauthorized_query_returns_error(self, schema):
        query = "query { me { id } }"
        context = {"user": None, "loaders": {}}

        success, result = await graphql(
            schema, {"query": query}, context_value=context
        )

        assert "errors" in result

class TestMutationResolvers:
    @pytest.mark.asyncio
    async def test_create_post_validates_input(self, schema, auth_context):
        mutation = """
            mutation {
                createPost(input: {title: "", content: "test"}) {
                    errors { field code }
                }
            }
        """

        success, result = await graphql(
            schema, {"query": mutation}, context_value=auth_context
        )

        assert result["data"]["createPost"]["errors"][0]["field"] == "title"

Integration Testing

# tests/test_integration.py
import pytest
from httpx import AsyncClient
from src.main import app

@pytest.fixture
async def client():
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

class TestGraphQLEndpoint:
    @pytest.mark.asyncio
    async def test_query_execution(self, client):
        response = await client.post(
            "/graphql",
            json={
                "query": "query { posts(first: 5) { edges { node { id } } } }"
            }
        )

        assert response.status_code == 200
        data = response.json()
        assert "data" in data
        assert "posts" in data["data"]

    @pytest.mark.asyncio
    async def test_query_depth_limit(self, client):
        # Query that exceeds depth limit
        deep_query = """
            query {
                user(id: "1") {
                    posts {
                        edges {
                            node {
                                author {
                                    posts {
                                        edges {
                                            node {
                                                author {
                                                    posts { edges { node { id } } }
                                                }
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        """

        response = await client.post("/graphql", json={"query": deep_query})
        data = response.json()

        assert "errors" in data
        assert any("depth" in str(err).lower() for err in data["errors"])

    @pytest.mark.asyncio
    async def test_introspection_disabled_in_production(self, client):
        introspection_query = """
            query { __schema { types { name } } }
        """

        response = await client.post(
            "/graphql",
            json={"query": introspection_query}
        )
        data = response.json()

        # Should be blocked in production
        assert "errors" in data

DataLoader Testing

# tests/test_dataloaders.py
import pytest
from src.loaders import DataLoaders

class TestDataLoaders:
    @pytest.mark.asyncio
    async def test_user_loader_batches_requests(self):
        batch_calls = []

        async def mock_batch(ids):
            batch_calls.append(list(ids))
            return [{"id": id, "name": f"User {id}"} for id in ids]

        loader = DataLoader(mock_batch)

        # Load multiple users
        results = await asyncio.gather(
            loader.load("1"),
            loader.load("2"),
            loader.load("3")
        )

        # Should batch into single call
        assert len(batch_calls) == 1
        assert set(batch_calls[0]) == {"1", "2", "3"}
        assert len(results) == 3

    @pytest.mark.asyncio
    async def test_user_loader_caches_results(self):
        call_count = 0

        async def mock_batch(ids):
            nonlocal call_count
            call_count += 1
            return [{"id": id} for id in ids]

        loader = DataLoader(mock_batch)

        # Load same user twice
        await loader.load("1")
        await loader.load("1")

        # Should only call batch once due to caching
        assert call_count == 1

Schema Validation Testing

# tests/test_schema.py
import pytest
from graphql import build_schema, validate_schema

def test_schema_is_valid():
    from src.schema import type_defs

    schema = build_schema(type_defs)
    errors = validate_schema(schema)

    assert len(errors) == 0, f"Schema errors: {errors}"

def test_required_types_exist():
    from src.schema import type_defs

    schema = build_schema(type_defs)
    type_map = schema.type_map

    required_types = ["User", "Post", "Query", "Mutation"]
    for type_name in required_types:
        assert type_name in type_map, f"Missing type: {type_name}"

def test_pagination_types_exist():
    from src.schema import type_defs

    schema = build_schema(type_defs)
    type_map = schema.type_map

    # Verify pagination types
    assert "PageInfo" in type_map
    assert "PostConnection" in type_map
    assert "PostEdge" in type_map

Running Tests

# Run all tests
pytest tests/ -v

# Run with coverage
pytest tests/ -v --cov=src --cov-report=term-missing

# Run specific test file
pytest tests/test_resolvers.py -v

# Run tests matching pattern
pytest tests/ -k "test_user" -v

# Run with async debugging
pytest tests/ -v --tb=short -x --asyncio-mode=auto

---

13. Critical Reminders

NEVER

• ❌ Allow unbounded queries without limits
• ❌ Skip field-level authorization
• ❌ Expose introspection in production
• ❌ Ignore N+1 query problems
• ❌ Trust user input without validation
• ❌ Return stack traces in errors
• ❌ Use blocking operations in resolvers

ALWAYS

• ✅ Use DataLoader for batching
• ✅ Implement query depth limits (≤7)
• ✅ Add query complexity analysis
• ✅ Validate all input arguments
• ✅ Implement field-level authorization
• ✅ Use pagination for lists
• ✅ Disable introspection in production
• ✅ Log query performance

Pre-Implementation Checklist

Phase 1: Before Writing Code

• [ ] Schema design reviewed and documented
• [ ] DataLoader strategy planned for relationships
• [ ] Authorization requirements identified per field
• [ ] Query complexity costs estimated
• [ ] Test cases written (TDD)
• [ ] Existing patterns in codebase reviewed

Phase 2: During Implementation

• [ ] Tests passing for each resolver
• [ ] DataLoader implemented for all relationships
• [ ] Field-level authorization in place
• [ ] Input validation on all mutations
• [ ] Error types properly defined
• [ ] No N+1 queries (verified with query logging)
• [ ] Pagination using cursor-based approach

Phase 3: Before Committing

• [ ] All tests pass: pytest tests/ -v
• [ ] Type checking passes: mypy src/ --strict
• [ ] Schema validates successfully
• [ ] Query depth limit configured (≤7)
• [ ] Query complexity limit configured
• [ ] Introspection disabled in production
• [ ] Error formatting hides stack traces
• [ ] Rate limiting configured
• [ ] Query timeout limits set
• [ ] Monitoring/logging configured
• [ ] Code review checklist completed

---

14. Summary

You are a GraphQL expert focused on: 1. Schema design - Type-safe, well-documented schemas 2. Performance - DataLoader batching, query optimization 3. Security - Complexity limits, field authorization, input validation 4. Type safety - Generated types, end-to-end type safety 5. Production readiness - Error handling, monitoring, testing

Key principles:

• Solve N+1 queries with DataLoader
• Protect against malicious queries with complexity/depth limits
• Implement field-level authorization
• Validate all inputs
• Design schemas for evolution
• Optimize for performance from day one
• Never expose sensitive data or errors

Technology stack:

• GraphQL 16+
• Apollo Server 4+
• DataLoader for batching
• GraphQL Code Generator for types
• Apollo Federation for microservices

📚 Reference Documentation:

• Advanced Patterns - Federation, Subscriptions, Error Handling
• Performance Guide - Query Optimization, Complexity Analysis, Caching
• Security Examples - Vulnerabilities, Attack Scenarios, Mitigations
• Anti-Patterns - Common Mistakes and How to Avoid Them

When building GraphQL APIs, prioritize security and performance equally. A fast API that's insecure is useless. A secure API that's slow is unusable. Design for both from the start.

Advanced GraphQL Patterns

Apollo Federation

User Service

// ---- User Service ----
import { buildSubgraphSchema } from '@apollo/subgraph';

const typeDefs = gql`
  extend schema
    @link(url: "https://specs.apollo.dev/federation/v2.3",
          import: ["@key", "@shareable"])

  type User @key(fields: "id") {
    id: ID!
    email: String!
    profile: Profile
  }

  type Profile {
    displayName: String!
    bio: String
  }

  type Query {
    me: User
    user(id: ID!): User
  }
`;

const resolvers = {
  Query: {
    me: (_, __, { user }) => user,
    user: (_, { id }) => UserService.findById(id),
  },

  User: {
    // Reference resolver for federation
    __resolveReference: async (reference) => {
      return UserService.findById(reference.id);
    },

    profile: (user) => ProfileService.findByUserId(user.id),
  },
};

const userSubgraph = buildSubgraphSchema({ typeDefs, resolvers });

Posts Service

// ---- Posts Service ----
const postTypeDefs = gql`
  extend schema
    @link(url: "https://specs.apollo.dev/federation/v2.3",
          import: ["@key", "@external"])

  # Extend User from user service
  type User @key(fields: "id") {
    id: ID! @external
    posts(first: Int = 10): [Post!]!
  }

  type Post @key(fields: "id") {
    id: ID!
    title: String!
    content: String!
    author: User!
    publishedAt: DateTime
  }

  type Query {
    post(id: ID!): Post
    posts(first: Int = 10): [Post!]!
  }
`;

const postResolvers = {
  Query: {
    post: (_, { id }) => PostService.findById(id),
    posts: (_, { first }) => PostService.findMany({ first }),
  },

  Post: {
    __resolveReference: (reference) => PostService.findById(reference.id),
    author: (post) => ({ __typename: 'User', id: post.authorId }),
  },

  User: {
    // Extend User type with posts field
    posts: (user, { first }) => {
      return PostService.findByAuthorId(user.id, first);
    },
  },
};

const postSubgraph = buildSubgraphSchema({
  typeDefs: postTypeDefs,
  resolvers: postResolvers,
});

Gateway Configuration

// ---- Gateway ----
import { ApolloGateway, IntrospectAndCompose } from '@apollo/gateway';

const gateway = new ApolloGateway({
  supergraphSdl: new IntrospectAndCompose({
    subgraphs: [
      { name: 'users', url: 'http://localhost:4001' },
      { name: 'posts', url: 'http://localhost:4002' },
    ],
    pollIntervalInMs: 10000, // Poll for schema changes
  }),

  buildService({ url }) {
    return new RemoteGraphQLDataSource({
      url,
      willSendRequest({ request, context }) {
        // Forward auth headers to subgraphs
        request.http.headers.set('authorization', context.token);
      },
    });
  },
});

const gatewayServer = new ApolloServer({
  gateway,
  subscriptions: false,
});

---

GraphQL Subscriptions

Real-Time Updates with PubSub

import { WebSocketServer } from 'ws';
import { useServer } from 'graphql-ws/lib/use/ws';
import { PubSub } from 'graphql-subscriptions';

const pubsub = new PubSub();

const typeDefs = gql`
  type Subscription {
    postCreated: Post!
    postUpdated(id: ID!): Post!
    commentAdded(postId: ID!): Comment!
  }
`;

const resolvers = {
  Mutation: {
    createPost: async (_, { input }, { user }) => {
      const post = await PostService.create({
        ...input,
        authorId: user.id,
      });

      // Publish to subscribers
      await pubsub.publish('POST_CREATED', { postCreated: post });

      return { post, errors: [] };
    },
  },

  Subscription: {
    postCreated: {
      subscribe: () => pubsub.asyncIterator(['POST_CREATED']),
    },

    postUpdated: {
      subscribe: (_, { id }) => {
        return pubsub.asyncIterator([`POST_UPDATED_${id}`]);
      },
    },

    commentAdded: {
      // Filter by postId
      subscribe: withFilter(
        () => pubsub.asyncIterator(['COMMENT_ADDED']),
        (payload, variables) => {
          return payload.commentAdded.postId === variables.postId;
        }
      ),
    },
  },
};

WebSocket Server Setup

// WebSocket server setup
const wsServer = new WebSocketServer({
  server: httpServer,
  path: '/graphql',
});

useServer(
  {
    schema,
    context: async (ctx) => {
      // Authenticate WebSocket connections
      const token = ctx.connectionParams?.authorization;
      const user = await authenticateToken(token);
      return { user };
    },
    onConnect: async (ctx) => {
      console.log('Client connected');
    },
    onDisconnect: () => {
      console.log('Client disconnected');
    },
  },
  wsServer
);

---

Advanced Error Handling

Custom Error Types

import { GraphQLError } from 'graphql';
import { ApolloServerErrorCode } from '@apollo/server/errors';

// ✅ Custom error types
class ValidationError extends GraphQLError {
  constructor(message: string, field?: string) {
    super(message, {
      extensions: {
        code: 'VALIDATION_ERROR',
        field,
      },
    });
  }
}

class NotFoundError extends GraphQLError {
  constructor(resource: string, id: string) {
    super(`${resource} not found`, {
      extensions: {
        code: 'NOT_FOUND',
        resource,
        id,
      },
    });
  }
}

Input Validation with Zod

import { z } from 'zod';

const CreatePostInputSchema = z.object({
  title: z.string().min(1).max(200),
  content: z.string().min(1).max(10000),
  status: z.enum(['DRAFT', 'PUBLISHED']).default('DRAFT'),
});

const resolvers = {
  Mutation: {
    createPost: async (_, { input }, { user, loaders }) => {
      // Authentication check
      if (!user) {
        throw new GraphQLError('Authentication required', {
          extensions: { code: 'UNAUTHENTICATED' },
        });
      }

      // Input validation
      const validation = CreatePostInputSchema.safeParse(input);
      if (!validation.success) {
        return {
          post: null,
          errors: validation.error.errors.map(err => ({
            message: err.message,
            field: err.path.join('.'),
            code: 'VALIDATION_ERROR',
          })),
        };
      }

      try {
        const post = await PostService.create({
          ...validation.data,
          authorId: user.id,
        });

        return { post, errors: [] };
      } catch (error) {
        // Don't expose internal errors
        console.error('Failed to create post:', error);

        return {
          post: null,
          errors: [{
            message: 'Failed to create post',
            code: 'INTERNAL_ERROR',
          }],
        };
      }
    },
  },
};

Error Formatting Plugin

// ✅ Error formatting plugin
const errorFormattingPlugin = {
  async requestDidStart() {
    return {
      async didEncounterErrors({ errors }) {
        errors.forEach(error => {
          // Log internal errors
          console.error('GraphQL Error:', {
            message: error.message,
            path: error.path,
            extensions: error.extensions,
          });

          // Remove stack traces in production
          if (process.env.NODE_ENV === 'production') {
            delete error.extensions?.stacktrace;
          }
        });
      },
    };
  },
};

---

Custom Authorization Directives

Directive Transformer

import { mapSchema, getDirective, MapperKind } from '@graphql-tools/utils';

function authDirective(directiveName: string) {
  return (schema) => {
    return mapSchema(schema, {
      [MapperKind.OBJECT_FIELD]: (fieldConfig) => {
        const authDirective = getDirective(schema, fieldConfig, directiveName)?.[0];

        if (authDirective) {
          const { resolve = defaultFieldResolver } = fieldConfig;
          const { requires } = authDirective;

          fieldConfig.resolve = async function (source, args, context, info) {
            if (requires === 'ADMIN' && context.user?.role !== 'ADMIN') {
              throw new GraphQLError('Admin access required');
            }

            return resolve(source, args, context, info);
          };
        }

        return fieldConfig;
      },
    });
  };
}

Schema with Custom Directives

// Schema with directive
const typeDefs = `
  directive @auth(requires: Role = USER) on FIELD_DEFINITION

  enum Role {
    USER
    ADMIN
  }

  type User {
    id: ID!
    email: String! @auth
    role: Role! @auth(requires: ADMIN)
  }
`;

GraphQL Anti-Patterns & Common Mistakes

Performance Anti-Patterns

Mistake 1: N+1 Query Problem

Problem:

// ❌ DON'T - Causes N+1 queries
const resolvers = {
  Query: {
    posts: async () => {
      return await Post.findMany(); // 1 query
    },
  },

  Post: {
    author: async (post) => {
      // This runs a separate query for EACH post
      // If you have 100 posts, this runs 100 queries!
      return await User.findById(post.authorId); // N queries
    },

    comments: async (post) => {
      // Another N queries!
      return await Comment.findByPostId(post.id);
    },
  },
};

// Query that triggers N+1:
// query {
//   posts {        # 1 query
//     author {     # 100 queries (if 100 posts)
//       name
//     }
//     comments {   # 100 more queries
//       text
//     }
//   }
// }
// Total: 201 database queries!

Solution:

// ✅ DO - Use DataLoader for batching
import DataLoader from 'dataloader';

class DataLoaders {
  userLoader = new DataLoader<string, User>(
    async (userIds) => {
      // Single batched query
      const users = await User.findMany({
        where: { id: { in: [...userIds] } },
      });

      const userMap = new Map(users.map(u => [u.id, u]));
      return userIds.map(id => userMap.get(id) || null);
    }
  );

  commentsByPostLoader = new DataLoader<string, Comment[]>(
    async (postIds) => {
      // Single batched query
      const comments = await Comment.findMany({
        where: { postId: { in: [...postIds] } },
      });

      const commentsByPost = new Map<string, Comment[]>();
      postIds.forEach(id => commentsByPost.set(id, []));
      comments.forEach(c => {
        const postComments = commentsByPost.get(c.postId) || [];
        postComments.push(c);
        commentsByPost.set(c.postId, postComments);
      });

      return postIds.map(id => commentsByPost.get(id) || []);
    }
  );
}

const resolvers = {
  Post: {
    author: (post, _, { loaders }) => loaders.userLoader.load(post.authorId),
    comments: (post, _, { loaders }) => loaders.commentsByPostLoader.load(post.id),
  },
};

// Same query now runs only 3 queries total!
// 1. Fetch posts
// 2. Batch fetch all authors
// 3. Batch fetch all comments

---

Mistake 2: No Query Complexity Limits

Problem:

// ❌ DON'T - Allow unlimited query complexity
const server = new ApolloServer({
  typeDefs,
  resolvers,
  // No protection against expensive queries!
});

// Attacker can send this:
// query {
//   posts(first: 10000) {
//     author {
//       posts(first: 10000) {
//         author {
//           posts(first: 10000) {
//             # This could fetch 1 trillion records!
//           }
//         }
//       }
//     }
//   }
// }

Solution:

// ✅ DO - Add complexity and depth limits
import depthLimit from 'graphql-depth-limit';
import { createComplexityLimitRule } from 'graphql-validation-complexity';

const server = new ApolloServer({
  typeDefs,
  resolvers,
  validationRules: [
    depthLimit(7), // Max nesting depth
    createComplexityLimitRule(1000), // Max complexity score
  ],
});

// Also limit pagination
const resolvers = {
  Query: {
    posts: async (_, { first }) => {
      const limit = Math.min(first || 10, 100); // Max 100
      return Post.findMany({ first: limit });
    },
  },
};

---

Mistake 3: Not Using Pagination

Problem:

// ❌ DON'T - Return unbounded lists
type Query {
  posts: [Post!]!  # Could return millions of records
  users: [User!]!
}

const resolvers = {
  Query: {
    posts: () => Post.findAll(), // Returns ALL posts!
  },
};

Solution:

// ✅ DO - Implement cursor-based pagination
type Query {
  posts(first: Int = 10, after: String): PostConnection!
}

type PostConnection {
  edges: [PostEdge!]!
  pageInfo: PageInfo!
}

type PostEdge {
  node: Post!
  cursor: String!
}

type PageInfo {
  hasNextPage: Boolean!
  endCursor: String
}

const resolvers = {
  Query: {
    posts: async (_, { first = 10, after }) => {
      const limit = Math.min(first, 100);
      const posts = await Post.findMany({
        take: limit + 1, // Fetch one extra to check hasNextPage
        cursor: after ? { id: after } : undefined,
      });

      const hasNextPage = posts.length > limit;
      const edges = posts.slice(0, limit).map(post => ({
        node: post,
        cursor: post.id,
      }));

      return {
        edges,
        pageInfo: {
          hasNextPage,
          endCursor: edges[edges.length - 1]?.cursor,
        },
      };
    },
  },
};

---

Security Anti-Patterns

Mistake 4: Missing Field-Level Authorization

Problem:

// ❌ DON'T - No authorization on sensitive fields
type User {
  id: ID!
  email: String!              # Anyone can see any user's email!
  socialSecurityNumber: String!  # Exposed to everyone!
  salary: Float!              # Private data exposed!
}

const resolvers = {
  Query: {
    user: (_, { id }) => User.findById(id), // No auth check!
  },
};

Solution:

// ✅ DO - Implement field-level authorization
import { shield, rule, and, or } from 'graphql-shield';

const isAuthenticated = rule()(
  (parent, args, ctx) => ctx.user !== null
);

const isOwner = rule()(
  (parent, args, ctx) => parent.id === ctx.user?.id
);

const isAdmin = rule()(
  (parent, args, ctx) => ctx.user?.role === 'ADMIN'
);

const permissions = shield({
  Query: {
    user: isAuthenticated,
  },
  User: {
    email: or(isOwner, isAdmin),
    socialSecurityNumber: and(isOwner, isAdmin),
    salary: or(isOwner, isAdmin),
  },
});

// Or use directives
type User {
  id: ID!
  email: String! @auth
  socialSecurityNumber: String! @auth(requires: ADMIN)
  salary: Float! @auth
}

---

Mistake 5: Not Validating Input

Problem:

// ❌ DON'T - Trust user input
const resolvers = {
  Mutation: {
    createPost: async (_, { input }) => {
      // No validation!
      return db.insert('posts', input);
    },

    updateUser: async (_, { id, email }) => {
      // No email validation!
      return User.update(id, { email });
    },
  },
};

// Attacker can send:
// mutation {
//   createPost(input: {
//     title: "A".repeat(1000000)  # 1 million characters
//     content: "<script>alert('xss')</script>"
//   })
// }

Solution:

// ✅ DO - Validate all inputs
import { z } from 'zod';

const CreatePostInputSchema = z.object({
  title: z.string().min(1).max(200),
  content: z.string().min(1).max(10000),
  tags: z.array(z.string()).max(10).optional(),
});

const EmailSchema = z.string().email();

const resolvers = {
  Mutation: {
    createPost: async (_, { input }, { user }) => {
      // Validate input
      const validation = CreatePostInputSchema.safeParse(input);
      if (!validation.success) {
        return {
          post: null,
          errors: validation.error.errors.map(err => ({
            message: err.message,
            field: err.path.join('.'),
            code: 'VALIDATION_ERROR',
          })),
        };
      }

      return PostService.create({
        ...validation.data,
        authorId: user.id,
      });
    },

    updateUser: async (_, { id, email }, { user }) => {
      // Validate email
      if (!EmailSchema.safeParse(email).success) {
        throw new GraphQLError('Invalid email format');
      }

      return User.update(id, { email });
    },
  },
};

---

Mistake 6: Exposing Internal Errors

Problem:

// ❌ DON'T - Expose internal error details
const resolvers = {
  Query: {
    user: async (_, { id }) => {
      try {
        return await db.query(
          'SELECT * FROM users WHERE id = $1',
          [id]
        );
      } catch (error) {
        // Exposes database structure and internals!
        throw new Error(`Database error: ${error.message}\n${error.stack}`);
      }
    },
  },
};

// Error response exposes:
// {
//   "errors": [{
//     "message": "Database error: relation 'users_internal_v2' does not exist",
//     "extensions": {
//       "stacktrace": [
//         "at Object.query (/app/db/index.js:42:15)",
//         "at /app/src/database/postgres.ts:123:45"
//       ]
//     }
//   }]
// }

Solution:

// ✅ DO - Sanitize error messages
const formatError = (error) => {
  // Log full error internally
  console.error('GraphQL Error:', {
    message: error.message,
    stack: error.stack,
    path: error.path,
  });

  // Production: sanitize errors
  if (process.env.NODE_ENV === 'production') {
    return {
      message: 'An error occurred',
      extensions: {
        code: error.extensions?.code || 'INTERNAL_ERROR',
      },
    };
  }

  // Development: show details (but still remove stack)
  return {
    message: error.message,
    extensions: {
      code: error.extensions?.code,
    },
  };
};

const server = new ApolloServer({
  typeDefs,
  resolvers,
  formatError,
});

---

Mistake 7: Introspection Enabled in Production

Problem:

// ❌ DON'T - Leave introspection enabled
const server = new ApolloServer({
  typeDefs,
  resolvers,
  introspection: true, // Exposes entire schema!
});

// Attacker can discover your entire API:
// query {
//   __schema {
//     types {
//       name
//       fields {
//         name
//       }
//     }
//   }
// }

Solution:

// ✅ DO - Disable introspection in production
const server = new ApolloServer({
  typeDefs,
  resolvers,
  introspection: process.env.NODE_ENV !== 'production',

  // Or conditional introspection for admins only
  plugins: [{
    async requestDidStart({ request, contextValue }) {
      if (request.operationName === 'IntrospectionQuery') {
        if (!contextValue.user?.isAdmin) {
          throw new GraphQLError('Forbidden');
        }
      }
    },
  }],
});

---

Schema Design Anti-Patterns

Mistake 8: Poor Nullability Design

Problem:

// ❌ DON'T - Everything nullable or everything non-null
type User {
  id: ID            # Could be null?
  email: String     # Should this ever be null?
  name: String!     # What if user hasn't set a name?
  posts: [Post]!    # Array can't be null but items can?
}

Solution:

// ✅ DO - Thoughtful nullability
type User {
  id: ID!              # ID is always required
  email: String!       # Email is always present
  displayName: String  # Optional: user may not have set it
  bio: String          # Optional: user may not have bio
  posts: [Post!]!      # Array never null, items never null
  avatar: URL          # Optional: user may not have avatar
}

---

Mistake 9: Not Using Proper Error Types

Problem:

// ❌ DON'T - Throw errors for expected conditions
type Mutation {
  createPost(input: CreatePostInput!): Post!
}

const resolvers = {
  Mutation: {
    createPost: async (_, { input }) => {
      if (!input.title) {
        throw new Error('Title is required'); // Forces client to parse errors
      }

      return Post.create(input);
    },
  },
};

Solution:

// ✅ DO - Use union types or error fields
type Mutation {
  createPost(input: CreatePostInput!): CreatePostPayload!
}

type CreatePostPayload {
  post: Post
  errors: [UserError!]
}

type UserError {
  message: String!
  field: String
  code: ErrorCode!
}

enum ErrorCode {
  VALIDATION_ERROR
  UNAUTHORIZED
  NOT_FOUND
}

const resolvers = {
  Mutation: {
    createPost: async (_, { input }) => {
      if (!input.title) {
        return {
          post: null,
          errors: [{
            message: 'Title is required',
            field: 'title',
            code: 'VALIDATION_ERROR',
          }],
        };
      }

      const post = await Post.create(input);
      return { post, errors: [] };
    },
  },
};

---

Mistake 10: Blocking Operations in Resolvers

Problem:

// ❌ DON'T - Use synchronous/blocking operations
const resolvers = {
  Query: {
    user: (_, { id }) => {
      // Blocks the event loop!
      const data = fs.readFileSync(`/users/${id}.json`);
      return JSON.parse(data);
    },

    processData: () => {
      // CPU-intensive blocking operation
      let result = 0;
      for (let i = 0; i < 1000000000; i++) {
        result += Math.sqrt(i);
      }
      return result;
    },
  },
};

Solution:

// ✅ DO - Use async operations
const resolvers = {
  Query: {
    user: async (_, { id }) => {
      // Non-blocking I/O
      const data = await fs.promises.readFile(`/users/${id}.json`);
      return JSON.parse(data);
    },

    processData: async () => {
      // Offload to worker thread
      return new Promise((resolve) => {
        const worker = new Worker('./worker.js');
        worker.on('message', resolve);
        worker.postMessage({ type: 'process' });
      });
    },
  },
};

---

Testing Anti-Patterns

Mistake 11: Not Testing Resolvers

Problem:

// ❌ DON'T - Skip resolver tests
// No tests for business logic

Solution:

// ✅ DO - Test resolvers thoroughly
import { expect, test } from 'vitest';

test('createPost resolver validates input', async () => {
  const result = await resolvers.Mutation.createPost(
    null,
    { input: { title: '', content: 'test' } },
    { user: { id: '1' } }
  );

  expect(result.errors).toHaveLength(1);
  expect(result.errors[0].code).toBe('VALIDATION_ERROR');
});

test('createPost requires authentication', async () => {
  await expect(
    resolvers.Mutation.createPost(
      null,
      { input: { title: 'test', content: 'test' } },
      { user: null }
    )
  ).rejects.toThrow('Authentication required');
});

GraphQL Performance Optimization Guide

Query Complexity & Depth Limiting

Complexity Analysis Configuration

import { ApolloServer } from '@apollo/server';
import { createComplexityLimitRule } from 'graphql-validation-complexity';
import depthLimit from 'graphql-depth-limit';

// ✅ Complexity analysis
const complexityLimit = createComplexityLimitRule(1000, {
  // Custom complexity per field
  scalarCost: 1,
  objectCost: 2,
  listFactor: 10,

  // Field-specific costs
  formatCosts: (cost, args, ctx, field) => {
    // Lists with arguments (pagination) cost more
    if (field.type.toString().includes('[') && args.first) {
      return cost * Math.min(args.first, 100);
    }
    return cost;
  },

  onCost: (cost) => {
    console.log('Query cost:', cost);
  },
});

Apollo Server Configuration with Limits

// ✅ Apollo Server configuration
const server = new ApolloServer({
  typeDefs,
  resolvers,

  validationRules: [
    depthLimit(7), // Max query depth
    complexityLimit,
  ],

  plugins: [
    {
      async requestDidStart() {
        return {
          async didResolveOperation({ request, document }) {
            // Log query complexity
            const complexity = calculateComplexity({
              schema,
              query: document,
              variables: request.variables,
            });

            if (complexity > 1000) {
              throw new GraphQLError('Query is too complex', {
                extensions: {
                  code: 'COMPLEXITY_LIMIT_EXCEEDED',
                  complexity,
                },
              });
            }
          },

          async executionDidStart() {
            const startTime = Date.now();

            return {
              async executionDidEnd() {
                const duration = Date.now() - startTime;

                // Timeout protection
                if (duration > 10000) {
                  console.warn('Slow query detected:', duration);
                }
              },
            };
          },
        };
      },
    },
  ],
});

Query Timeout Implementation

// ✅ Query timeout
const resolvers = {
  Query: {
    posts: async (_, args, ctx) => {
      // Limit expensive queries
      if (args.first > 100) {
        throw new GraphQLError('Maximum limit is 100', {
          extensions: { code: 'BAD_USER_INPUT' },
        });
      }

      // Set query timeout
      return Promise.race([
        Post.findMany(args),
        new Promise((_, reject) =>
          setTimeout(() => reject(new Error('Query timeout')), 5000)
        ),
      ]);
    },
  },
};

---

Advanced DataLoader Patterns

Multi-Key DataLoader

import DataLoader from 'dataloader';

class DataLoaders {
  // Composite key loader
  postsByAuthorAndStatusLoader = new DataLoader<
    { authorId: string; status: string },
    Post[]
  >(
    async (keys) => {
      // Extract unique author IDs and statuses
      const authorIds = [...new Set(keys.map(k => k.authorId))];
      const statuses = [...new Set(keys.map(k => k.status))];

      // Single query with all combinations
      const posts = await Post.findMany({
        where: {
          authorId: { in: authorIds },
          status: { in: statuses },
        },
      });

      // Group by composite key
      const postsByKey = new Map<string, Post[]>();
      keys.forEach(key => {
        const mapKey = `${key.authorId}:${key.status}`;
        postsByKey.set(mapKey, []);
      });

      posts.forEach(post => {
        const mapKey = `${post.authorId}:${post.status}`;
        const existingPosts = postsByKey.get(mapKey) || [];
        existingPosts.push(post);
        postsByKey.set(mapKey, existingPosts);
      });

      return keys.map(key => {
        const mapKey = `${key.authorId}:${key.status}`;
        return postsByKey.get(mapKey) || [];
      });
    },
    {
      // Custom cache key function
      cacheKeyFn: (key) => `${key.authorId}:${key.status}`,
    }
  );
}

Primed DataLoader Cache

// Pre-populate DataLoader cache
const resolvers = {
  Query: {
    posts: async (_, { first, after }, { loaders }) => {
      const posts = await Post.findMany({ first, after });

      // Prime the cache with fetched posts
      posts.forEach(post => {
        loaders.postLoader.prime(post.id, post);
        // Also prime related data
        if (post.author) {
          loaders.userLoader.prime(post.authorId, post.author);
        }
      });

      return posts;
    },
  },
};

---

Caching Strategies

Field-Level Caching

import { InMemoryLRUCache } from '@apollo/utils.keyvaluecache';

const cache = new InMemoryLRUCache({
  maxSize: Math.pow(2, 20) * 100, // 100 MB
  ttl: 300, // 5 minutes
});

const resolvers = {
  Query: {
    popularPosts: async (_, __, { cache }) => {
      const cacheKey = 'popular_posts';

      // Check cache first
      const cached = await cache.get(cacheKey);
      if (cached) {
        return JSON.parse(cached);
      }

      // Compute expensive query
      const posts = await Post.findPopular();

      // Store in cache
      await cache.set(cacheKey, JSON.stringify(posts), { ttl: 300 });

      return posts;
    },
  },
};

Persisted Queries

import { ApolloServer } from '@apollo/server';
import { createPersistedQueryLink } from '@apollo/client/link/persisted-queries';

const server = new ApolloServer({
  typeDefs,
  resolvers,
  persistedQueries: {
    cache: new InMemoryLRUCache(),
    ttl: 900, // 15 minutes
  },
});

// Client-side configuration
const link = createPersistedQueryLink({
  sha256,
  useGETForHashedQueries: true,
});

---

Database Query Optimization

Projection/Selection Optimization

import { parseResolveInfo, simplifyParsedResolveInfoFragmentWithType } from 'graphql-parse-resolve-info';

const resolvers = {
  Query: {
    user: async (_, { id }, ctx, info) => {
      // Parse requested fields from GraphQL query
      const parsedInfo = parseResolveInfo(info);
      const { fields } = simplifyParsedResolveInfoFragmentWithType(parsedInfo, info.returnType);

      // Only select requested fields from database
      const select = {};
      if (fields.email) select.email = true;
      if (fields.profile) select.profile = true;

      return db.user.findUnique({
        where: { id },
        select,
      });
    },
  },
};

Batched Database Writes

const resolvers = {
  Mutation: {
    createPosts: async (_, { inputs }, { user }) => {
      // Batch insert instead of individual inserts
      const posts = await db.$transaction(
        inputs.map(input =>
          db.post.create({
            data: {
              ...input,
              authorId: user.id,
            },
          })
        )
      );

      return { posts, errors: [] };
    },
  },
};

---

Performance Monitoring

Query Performance Plugin

const performanceMonitoringPlugin = {
  async requestDidStart() {
    const start = Date.now();

    return {
      async willSendResponse({ response, queryHash }) {
        const duration = Date.now() - start;

        // Log slow queries
        if (duration > 1000) {
          console.warn('Slow query detected:', {
            queryHash,
            duration,
            response: response.body,
          });
        }

        // Send to monitoring service
        metrics.recordQueryDuration(queryHash, duration);
      },
    };
  },
};

Resolver-Level Tracing

import { wrapResolver } from '@apollo/server';

const tracingPlugin = {
  async requestDidStart() {
    return {
      async executionDidStart() {
        return {
          willResolveField({ info }) {
            const start = Date.now();

            return () => {
              const duration = Date.now() - start;
              console.log(`${info.parentType.name}.${info.fieldName}: ${duration}ms`);
            };
          },
        };
      },
    };
  },
};

GraphQL Security Examples & Vulnerabilities

Critical Security Vulnerabilities

1. Unbounded Query Attacks

Attack Example:

# ❌ Malicious query - can crash server
query {
  posts(first: 999999) {
    author {
      posts(first: 999999) {
        author {
          posts(first: 999999) {
            author {
              posts(first: 999999) {
                # Exponential data explosion
                # Can request millions of records
              }
            }
          }
        }
      }
    }
  }
}

Impact:

• Server memory exhaustion
• Database overload
• Service denial for legitimate users
• Potential server crash

Mitigation:

import depthLimit from 'graphql-depth-limit';
import { createComplexityLimitRule } from 'graphql-validation-complexity';

const server = new ApolloServer({
  typeDefs,
  resolvers,
  validationRules: [
    depthLimit(7), // Max query depth
    createComplexityLimitRule(1000), // Max complexity score
  ],
});

// Also limit pagination arguments
const resolvers = {
  Query: {
    posts: async (_, { first }) => {
      if (first > 100) {
        throw new GraphQLError('Maximum limit is 100');
      }
      return Post.findMany({ first });
    },
  },
};

---

2. Introspection Exposure

Attack:

# Attacker discovers your entire schema
query IntrospectionQuery {
  __schema {
    types {
      name
      fields {
        name
        type {
          name
        }
      }
    }
  }
}

Risk:

• Exposes internal API structure
• Reveals hidden fields and types
• Helps attackers find vulnerabilities
• Leaks business logic

Mitigation:

const server = new ApolloServer({
  typeDefs,
  resolvers,
  // Disable in production
  introspection: process.env.NODE_ENV !== 'production',

  // Or use conditional introspection
  plugins: [
    {
      async requestDidStart({ request, contextValue }) {
        // Only allow introspection for admins
        if (request.operationName === 'IntrospectionQuery') {
          if (!contextValue.user?.isAdmin) {
            throw new GraphQLError('Introspection disabled');
          }
        }
      },
    },
  ],
});

---

3. Field-Level Authorization Bypass

Attack:

# Attacker requests sensitive fields
query {
  user(id: "123") {
    email              # Should be private
    socialSecurityNumber  # Admin-only field
    password           # Should never be exposed
    creditCard         # Sensitive data
  }
}

Risk:

• Unauthorized data access
• Privacy violations
• Compliance issues (GDPR, HIPAA)
• Data leaks

Mitigation with graphql-shield:

import { shield, rule, and, or } from 'graphql-shield';

const isAuthenticated = rule()(
  async (parent, args, ctx) => ctx.user !== null
);

const isAdmin = rule()(
  async (parent, args, ctx) => ctx.user?.role === 'ADMIN'
);

const isOwner = rule()(
  async (parent, args, ctx) => parent.id === ctx.user?.id
);

const permissions = shield({
  Query: {
    user: isAuthenticated,
  },
  User: {
    email: or(isOwner, isAdmin),
    socialSecurityNumber: isAdmin,
    // Never expose password field
    password: () => false,
  },
});

---

4. Batch Query Attacks

Attack:

# Send 1000+ queries in a single request
query {
  q1: user(id: "1") { email }
  q2: user(id: "2") { email }
  q3: user(id: "3") { email }
  # ... repeated 1000+ times
}

Mitigation:

import { ApolloServerPluginInlineTrace } from '@apollo/server/plugin/inlineTrace';

const queryLimitPlugin = {
  async requestDidStart({ request }) {
    // Count operation aliases
    const operationCount = Object.keys(request.query.match(/\w+:/g) || []).length;

    if (operationCount > 10) {
      throw new GraphQLError('Too many operations in single request');
    }
  },
};

---

5. SQL Injection via Resolvers

Vulnerable Code:

// ❌ NEVER DO THIS
const resolvers = {
  Query: {
    user: async (_, { id }) => {
      // Direct string interpolation - VULNERABLE!
      const query = `SELECT * FROM users WHERE id = '${id}'`;
      return db.raw(query);
    },
  },
};

Attack:

query {
  user(id: "1' OR '1'='1") {
    email
  }
}

Mitigation:

// ✅ Use parameterized queries
const resolvers = {
  Query: {
    user: async (_, { id }) => {
      // Parameterized query - SAFE
      return db.user.findUnique({
        where: { id }, // ORM handles sanitization
      });
    },
  },
};

---

6. Information Disclosure via Error Messages

Vulnerable:

// ❌ Exposes internal details
const resolvers = {
  Query: {
    user: async (_, { id }) => {
      try {
        return await db.query('SELECT * FROM users WHERE id = $1', [id]);
      } catch (error) {
        // Exposes database structure!
        throw new Error(`Database error: ${error.message}`);
      }
    },
  },
};

Mitigation:

// ✅ Sanitize error messages
const formatError = (error) => {
  // Log full error internally
  console.error('Internal error:', error);

  // Return sanitized error to client
  if (process.env.NODE_ENV === 'production') {
    return {
      message: 'An error occurred',
      extensions: {
        code: error.extensions?.code || 'INTERNAL_SERVER_ERROR',
      },
    };
  }

  // Development: show details
  return error;
};

const server = new ApolloServer({
  typeDefs,
  resolvers,
  formatError,
});

---

7. Denial of Service via Circular Queries

Attack:

query {
  user(id: "1") {
    friends {
      friends {
        friends {
          friends {
            # Circular reference explosion
          }
        }
      }
    }
  }
}

Mitigation:

import depthLimit from 'graphql-depth-limit';

const server = new ApolloServer({
  typeDefs,
  resolvers,
  validationRules: [
    depthLimit(5), // Prevent deep nesting
  ],
});

---

OWASP Top 10 2025 - GraphQL Specific Mitigations

A01:2025 - Broken Access Control

Implement Field-Level Authorization:

import { shield, rule } from 'graphql-shield';

const canAccessField = rule()(
  async (parent, args, ctx, info) => {
    const fieldName = info.fieldName;
    const userId = ctx.user?.id;

    // Check permissions in database
    const hasAccess = await PermissionService.canAccess(
      userId,
      parent.__typename,
      fieldName
    );

    return hasAccess;
  }
);

const permissions = shield({
  User: {
    '*': canAccessField, // Apply to all fields
  },
});

A04:2025 - Insecure Design

Implement Rate Limiting:

import rateLimit from 'graphql-rate-limit';

const rateLimitDirective = rateLimit({
  identifyContext: (ctx) => ctx.user?.id || ctx.ip,
  formatError: () => new GraphQLError('Rate limit exceeded'),
});

const typeDefs = gql`
  directive @rateLimit(
    max: Int = 10
    window: String = "1m"
  ) on FIELD_DEFINITION

  type Query {
    expensiveQuery: [Result!]! @rateLimit(max: 5, window: "1m")
  }
`;

A05:2025 - Security Misconfiguration

Secure Production Configuration:

const server = new ApolloServer({
  typeDefs,
  resolvers,

  // Production security settings
  introspection: false,
  playground: false,
  debug: false,

  validationRules: [
    depthLimit(7),
    complexityLimit(1000),
  ],

  plugins: [
    ApolloServerPluginLandingPageDisabled(),
  ],

  formatError: sanitizeError,
});

---

Security Checklist

• [ ] Query depth limiting (≤7 levels)
• [ ] Query complexity analysis
• [ ] Introspection disabled in production
• [ ] Field-level authorization
• [ ] Input validation on all mutations
• [ ] Rate limiting per user/IP
• [ ] Parameterized database queries
• [ ] Error message sanitization
• [ ] No sensitive data in schema
• [ ] Authentication on all protected fields
• [ ] HTTPS only in production
• [ ] CORS properly configured
• [ ] No password fields in types
• [ ] Audit logging for sensitive operations
• [ ] Regular security audits