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

bash

# 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

graphql

# 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

bash

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

Generate Resolvers

bash

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:

graphql

# 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

typescript

// 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

typescript

// 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

typescript

// 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

bash

# 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

graphql

# 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

typescript

// 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

typescript

// 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

graphql

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

typescript

// 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

typescript

// 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

graphql

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

typescript

// 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

typescript

// 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:

bash

# 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:

bash

# 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:

bash

# 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

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)

Search AI Tools

senior-graphql

Install this agent skill to your Project

Metadata

SKILL.md

Senior GraphQL Specialist

Overview

Core Capabilities

Schema Architecture

Resolver Development

Apollo Federation

Performance Optimization

Real-time Features

Quick Start

Key Workflows

1. Schema-First API Design

2. DataLoader Implementation for N+1 Prevention

3. Apollo Federation Setup

4. Real-time Subscriptions

Python Tools

schema_analyzer.py

resolver_generator.py

federation_scaffolder.py

Best Practices

Schema Design

Resolver Patterns

Federation

Performance

References

Reference Files

External Resources