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

1. 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

1. **Verify if Certainty ?"

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)

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

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

### Pattern 5: Async Resolver Optimization

**Bad - Blocking Operations:**
```python
# ❌ 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(
    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(
    async (authorIds) => {
      const posts = await Post.findMany({
        where: { authorId: { in: [...authorIds] } },
      });

      const postsByAuthor = new Map();
      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 => {
  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](references/advanced-patterns.md)

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

---

5. Security Standards

OWASP Top 10 2025 Mapping

| OWASP ID | Category | GraphQL Risk | Mitigation | |----------|----------|--------------|------------| | A01:2025 | Broken Access Control | Unauthorized field access | Field-level authorization | | A02:2025 | Security Misconfiguration | Introspection enabled | Disable in production | | A03:2025 | Supply Chain | Malicious resolvers | Code review, dependency scanning | | A04:2025 | Insecure Design | No query limits | Complexity/depth limits | | A05:2025 | Identification & Auth | Missing auth checks | Context-based auth | | A06:2025 | Vulnerable Components | Outdated GraphQL libs | Update dependencies | | A07:2025 | Cryptographic Failures | Exposed sensitive data | Field-level permissions | | A08:2025 | Injection | SQL injection in resolvers | Parameterized queries | | A09:2025 | Logging Failures | No query logging | Apollo Studio, monitoring | | A10:2025 | Exception Handling | Stack traces in errors | Format errors properly |

📚 For detailed security vulnerabilities and examples, see [references/security-examples.md](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](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"]

    @pyte

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [martinholovsky](https://github.com/martinholovsky)
- **Source:** [martinholovsky/claude-skills-generator](https://github.com/martinholovsky/claude-skills-generator)
- **License:** Unlicense

Install and usage instructions live in the source repository linked above.