Senior GraphQL Specialist

Expert GraphQL API design and architecture skill for building scalable, type-safe APIs with Apollo Server, Federation, and modern GraphQL patterns.

Overview

This skill provides comprehensive GraphQL development capabilities including schema design, resolver implementation, federation architecture, real-time subscriptions, and performance optimization through DataLoader patterns.

Time Savings: 50%+ reduction in GraphQL API development time through schema generation, resolver scaffolding, and automated federation setup.

Quality Improvement: 40%+ improvement in API consistency through schema-first development, type safety enforcement, and automated best practices.

Core Capabilities

Schema Architecture

• Schema-first design methodology

• Type system design (scalars, enums, interfaces, unions)

• Input type and argument patterns

• Custom directive implementation

• Schema stitching and composition

Resolver Development

• Resolver pattern implementation

• Context and middleware integration

• Authentication/authorization in resolvers

• Error handling and formatting

• N+1 query prevention with DataLoader

Apollo Federation

• Supergraph architecture design

• Subgraph creation and entity definitions

• @key , @external , @requires directive usage

• Gateway configuration

• Schema composition validation

Performance Optimization

• Query complexity analysis and limiting

• Depth limiting implementation

• Caching strategies (Apollo Cache, Redis)

• Batching with DataLoader

• Persisted queries

Real-time Features

• Subscription implementation

• WebSocket configuration

• PubSub patterns

• Filtered subscriptions

Quick Start

Analyze existing GraphQL schema

python scripts/schema_analyzer.py schema.graphql --output json

Generate resolvers from schema

python scripts/resolver_generator.py schema.graphql --output src/resolvers

Scaffold Apollo Federation subgraph

python scripts/federation_scaffolder.py users-service --entities User,Profile

Key Workflows

1. Schema-First API Design

Goal: Design a type-safe GraphQL schema following best practices.

Steps:

Analyze Requirements

• Identify domain entities and relationships

• Map CRUD operations to queries/mutations

• Define subscription needs for real-time features

Design Schema

Types with clear naming conventions

type User { id: ID! email: String! profile: Profile posts(first: Int, after: String): PostConnection! createdAt: DateTime! }

Relay-style pagination

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

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

type PageInfo { hasNextPage: Boolean! hasPreviousPage: Boolean! startCursor: String endCursor: String }

Input types for mutations

input CreateUserInput { email: String! name: String! password: String! }

Clear query/mutation organization

type Query { user(id: ID!): User users(first: Int, after: String): UserConnection! me: User }

type Mutation { createUser(input: CreateUserInput!): CreateUserPayload! updateUser(id: ID!, input: UpdateUserInput!): UpdateUserPayload! deleteUser(id: ID!): DeleteUserPayload! }

Subscription for real-time

type Subscription { userCreated: User! postPublished(authorId: ID): Post! }

Validate Schema

python scripts/schema_analyzer.py schema.graphql --validate

Generate Resolvers

python scripts/resolver_generator.py schema.graphql --output src/resolvers --typescript

Success Criteria:

• Schema passes validation

• All types have descriptions

• Relay pagination implemented for lists

• Input types for all mutations

• Clear naming conventions followed

2. DataLoader Implementation for N+1 Prevention

Goal: Eliminate N+1 queries using DataLoader batching.

Problem Example:

This query would cause N+1 without DataLoader

query { posts { # 1 query for posts author { # N queries for authors (one per post!) name } } }

Solution:

Create DataLoader Factory

// src/dataloaders/index.ts import DataLoader from 'dataloader'; import { prisma } from '../lib/prisma';

export const createLoaders = () => ({ userLoader: new DataLoader<string, User>(async (userIds) => { const users = await prisma.user.findMany({ where: { id: { in: [...userIds] } } }); // Return in same order as requested IDs const userMap = new Map(users.map(u => [u.id, u])); return userIds.map(id => userMap.get(id) || null); }),

postsByAuthorLoader: new DataLoader<string, Post[]>(async (authorIds) => { const posts = await prisma.post.findMany({ where: { authorId: { in: [...authorIds] } } }); // Group posts by authorId const postMap = new Map<string, Post[]>(); posts.forEach(post => { const existing = postMap.get(post.authorId) || []; existing.push(post); postMap.set(post.authorId, existing); }); return authorIds.map(id => postMap.get(id) || []); }), });

export type Loaders = ReturnType<typeof createLoaders>;

Add Loaders to Context

// src/server.ts import { createLoaders } from './dataloaders';

const server = new ApolloServer({ typeDefs, resolvers, context: ({ req }) => ({ user: authenticateToken(req), loaders: createLoaders(), // Fresh loaders per request }), });

Use in Resolvers

// src/resolvers/post.resolver.ts export const PostResolvers = { Post: { author: (parent, _, { loaders }) => { return loaders.userLoader.load(parent.authorId); }, }, };

Verify Batching

• Enable query logging

• Run test query

• Confirm single batch query instead of N queries

Success Criteria:

• Batch queries visible in logs

• Query count reduced from N+1 to 2

• Response time improved significantly

• DataLoader cache cleared per request

3. Apollo Federation Setup

Goal: Build a federated supergraph from multiple subgraphs.

Architecture:

┌─────────────────────────────────────────────────┐ │ Apollo Gateway │ │ (Schema Composition) │ └─────────────────────────────────────────────────┘ │ │ │ ▼ ▼ ▼ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Users │ │ Posts │ │ Comments │ │ Subgraph │ │ Subgraph │ │ Subgraph │ └─────────────┘ └─────────────┘ └─────────────┘

Steps:

Scaffold Subgraphs

Create users subgraph

python scripts/federation_scaffolder.py users-service
--entities User,Profile
--port 4001

Create posts subgraph

python scripts/federation_scaffolder.py posts-service
--entities Post
--references User
--port 4002

Create comments subgraph

python scripts/federation_scaffolder.py comments-service
--entities Comment
--references User,Post
--port 4003

Define Entity References

users-service/schema.graphql

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

posts-service/schema.graphql

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

Extend User to add posts field

extend type User @key(fields: "id") { id: ID! @external posts: [Post!]! }

comments-service/schema.graphql

type Comment @key(fields: "id") { id: ID! content: String! author: User! post: Post! }

extend type Post @key(fields: "id") { id: ID! @external comments: [Comment!]! }

Implement Reference Resolvers

// posts-service/resolvers.ts export const resolvers = { User: { __resolveReference: async (user, { dataSources }) => { // Return only the fields this subgraph owns return { id: user.id }; }, posts: async (user, _, { dataSources }) => { return dataSources.postsAPI.getPostsByAuthor(user.id); }, }, Post: { __resolveReference: async (post, { dataSources }) => { return dataSources.postsAPI.getPost(post.id); }, author: (post) => { // Return reference for gateway to resolve return { __typename: 'User', id: post.authorId }; }, }, };

Configure Gateway

// gateway/index.ts import { ApolloGateway, IntrospectAndCompose } from '@apollo/gateway'; import { ApolloServer } from '@apollo/server';

const gateway = new ApolloGateway({ supergraphSdl: new IntrospectAndCompose({ subgraphs: [ { name: 'users', url: 'http://localhost:4001/graphql' }, { name: 'posts', url: 'http://localhost:4002/graphql' }, { name: 'comments', url: 'http://localhost:4003/graphql' }, ], }), });

const server = new ApolloServer({ gateway });

Test Federated Query

query FederatedQuery { user(id: "123") { id name posts { id title comments { content author { name # Resolves back to users subgraph } } } } }

Success Criteria:

• All subgraphs start without errors

• Schema composition succeeds

• Cross-subgraph queries resolve correctly

• Entity references work bidirectionally

4. Real-time Subscriptions

Goal: Implement GraphQL subscriptions for real-time updates.

Steps:

Configure WebSocket Server

// src/server.ts import { createServer } from 'http'; import { WebSocketServer } from 'ws'; import { useServer } from 'graphql-ws/lib/use/ws'; import { ApolloServer } from '@apollo/server'; import { ApolloServerPluginDrainHttpServer } from '@apollo/server/plugin/drainHttpServer';

const httpServer = createServer(app);

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

const serverCleanup = useServer( { schema, context: (ctx) => ({ user: authenticateWebSocket(ctx.connectionParams), }), }, wsServer );

const server = new ApolloServer({ schema, plugins: [ ApolloServerPluginDrainHttpServer({ httpServer }), { async serverWillStart() { return { async drainServer() { await serverCleanup.dispose(); }, }; }, }, ], });

Implement PubSub

// src/pubsub.ts import { PubSub } from 'graphql-subscriptions'; import { RedisPubSub } from 'graphql-redis-subscriptions';

// For production, use Redis PubSub export const pubsub = new RedisPubSub({ connection: process.env.REDIS_URL, });

// Event types export const EVENTS = { POST_CREATED: 'POST_CREATED', POST_UPDATED: 'POST_UPDATED', COMMENT_ADDED: 'COMMENT_ADDED', USER_ONLINE: 'USER_ONLINE', };

Define Subscription Schema

type Subscription { postCreated: Post! postUpdated(id: ID!): Post! commentAdded(postId: ID!): Comment! userPresence(roomId: ID!): UserPresenceEvent! }

type UserPresenceEvent { user: User! status: PresenceStatus! }

enum PresenceStatus { ONLINE OFFLINE AWAY }

Implement Subscription Resolvers

// src/resolvers/subscription.resolver.ts import { withFilter } from 'graphql-subscriptions'; import { pubsub, EVENTS } from '../pubsub';

export const SubscriptionResolvers = { Subscription: { postCreated: { subscribe: () => pubsub.asyncIterator([EVENTS.POST_CREATED]), },

postUpdated: {
  subscribe: withFilter(
    () => pubsub.asyncIterator([EVENTS.POST_UPDATED]),
    (payload, variables) => {
      return payload.postUpdated.id === variables.id;
    }
  ),
},

commentAdded: {
  subscribe: withFilter(
    () => pubsub.asyncIterator([EVENTS.COMMENT_ADDED]),
    (payload, variables, context) => {
      // Only notify if user has access to the post
      return payload.commentAdded.postId === variables.postId;
    }
  ),
},

}, };

Publish Events

// src/resolvers/mutation.resolver.ts export const MutationResolvers = { Mutation: { createPost: async (_, { input }, { user, prisma }) => { const post = await prisma.post.create({ data: { ...input, authorId: user.id }, });

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

  return { post };
},

addComment: async (_, { input }, { user, prisma }) => {
  const comment = await prisma.comment.create({
    data: { ...input, authorId: user.id },
  });

  // Publish to subscribers watching this post
  await pubsub.publish(EVENTS.COMMENT_ADDED, { commentAdded: comment });

  return { comment };
},

}, };

Success Criteria:

• WebSocket connection established

• Subscriptions receive real-time updates

• Filtering works correctly

• Connection cleanup on disconnect

• Production-ready with Redis PubSub

Python Tools

schema_analyzer.py

Purpose: Analyze GraphQL schemas for quality, complexity, and best practices.

Usage:

Basic analysis

python scripts/schema_analyzer.py schema.graphql

JSON output for tooling

python scripts/schema_analyzer.py schema.graphql --output json

Validate against best practices

python scripts/schema_analyzer.py schema.graphql --validate

Analyze complexity and depth

python scripts/schema_analyzer.py schema.graphql --complexity

Features:

• Type system analysis (types, interfaces, unions, enums)

• Query/mutation/subscription inventory

• Complexity scoring per operation

• Deprecation tracking

• Naming convention validation

• Description coverage checking

• Circular reference detection

resolver_generator.py

Purpose: Generate TypeScript resolvers from GraphQL schema.

Usage:

Generate resolvers

python scripts/resolver_generator.py schema.graphql --output src/resolvers

With DataLoader integration

python scripts/resolver_generator.py schema.graphql --output src/resolvers --dataloader

For specific types only

python scripts/resolver_generator.py schema.graphql --output src/resolvers --types User,Post

Generate with tests

python scripts/resolver_generator.py schema.graphql --output src/resolvers --tests

Generated Output:

• Resolver files per type

• Type definitions

• DataLoader factories

• Context type definitions

• Jest test stubs

federation_scaffolder.py

Purpose: Scaffold Apollo Federation subgraphs with proper entity definitions.

Usage:

Create new subgraph

python scripts/federation_scaffolder.py users-service --entities User,Profile

With entity references

python scripts/federation_scaffolder.py posts-service --entities Post --references User

Full service with Docker

python scripts/federation_scaffolder.py comments-service --entities Comment --docker --port 4003

Scaffold gateway

python scripts/federation_scaffolder.py gateway --subgraphs users:4001,posts:4002,comments:4003

Generated Structure:

service-name/ ├── src/ │ ├── schema.graphql # Federation schema │ ├── resolvers/ # Type resolvers │ ├── dataloaders/ # DataLoader factories │ ├── datasources/ # Data access layer │ └── index.ts # Apollo Server setup ├── tests/ # Jest tests ├── Dockerfile # Container definition ├── docker-compose.yml # Local development └── package.json

Best Practices

Schema Design

• Use descriptive names (avoid abbreviations)

• Document all types and fields

• Implement Relay-style pagination for lists

• Use input types for mutations

• Return payload types from mutations (not raw types)

• Version breaking changes with new fields (not removal)

Resolver Patterns

• Keep resolvers thin (delegate to services)

• Use DataLoader for all batch-able relations

• Implement proper error handling

• Add authentication at resolver level

• Log slow resolvers for optimization

Federation

• Define clear subgraph boundaries

• Minimize cross-subgraph queries

• Use @requires sparingly

• Implement proper health checks

• Version subgraph schemas independently

Performance

• Implement query complexity limits

• Use persisted queries in production

• Cache with appropriate TTLs

• Monitor resolver execution time

• Implement query depth limiting

References

Reference Files

• references/schema-patterns.md

• Schema design patterns and conventions

• references/federation-guide.md

• Apollo Federation architecture guide

• references/performance-optimization.md

• GraphQL performance best practices

External Resources

• GraphQL Specification

• Apollo Server Documentation

• Apollo Federation

• DataLoader

Version: 1.0.0 Last Updated: 2025-12-16 Skill Type: Engineering specialist Python Tools: 3 (schema_analyzer.py, resolver_generator.py, federation_scaffolder.py)