graphql

GraphQL 技能

概述

GraphQL 是一种用于 API 的查询语言和运行时，客户端可以借此精确请求所需数据。它提供强类型架构和单端点架构，并能消除 REST API 中常见的过度获取和获取不足问题。

适用场景

• 为多种客户端类型（Web、移动设备、IoT）构建灵活的 API
• 数据需求复杂，且包含嵌套关系
• 注重移动端体验、需要提高带宽利用效率的应用
• 降低 API 版本管理的复杂度
• 需要通过订阅获取实时数据
• 微服务聚合与联邦
• 通过强类型和自省提升开发者体验

快速入门

1. 定义架构（SDL）

# schema.graphql
type User {
  id: ID!
  name: String!
  email: String!
  posts: [Post!]!
}

type Post {
  id: ID!
  title: String!
  content: String!
  author: User!
  publishedAt: DateTime
}

type Query {
  user(id: ID!): User
  users: [User!]!
  post(id: ID!): Post
}

type Mutation {
  createPost(title: String!, content: String!, authorId: ID!): Post!
  updatePost(id: ID!, title: String, content: String): Post!
  deletePost(id: ID!): Boolean!
}

2. 编写解析器（TypeScript + Apollo Server）

import { ApolloServer } from '@apollo/server';
import { startStandaloneServer } from '@apollo/server/standalone';
import { readFileSync } from 'fs';

// Load schema
const typeDefs = readFileSync('./schema.graphql', 'utf-8');

// Mock data
const users = [
  { id: '1', name: 'Alice', email: '[email protected]' },
  { id: '2', name: 'Bob', email: '[email protected]' },
];

const posts = [
  { id: '1', title: 'GraphQL Intro', content: 'Learning GraphQL...', authorId: '1' },
  { id: '2', title: 'Apollo Server', content: 'Building APIs...', authorId: '1' },
];

// Resolvers
const resolvers = {
  Query: {
    user: (_, { id }) => users.find(u => u.id === id),
    users: () => users,
    post: (_, { id }) => posts.find(p => p.id === id),
  },

  Mutation: {
    createPost: (_, { title, content, authorId }) => {
      const post = {
        id: String(posts.length + 1),
        title,
        content,
        authorId,
      };
      posts.push(post);
      return post;
    },

    updatePost: (_, { id, title, content }) => {
      const post = posts.find(p => p.id === id);
      if (!post) throw new Error('Post not found');
      if (title) post.title = title;
      if (content) post.content = content;
      return post;
    },

    deletePost: (_, { id }) => {
      const index = posts.findIndex(p => p.id === id);
      if (index === -1) return false;
      posts.splice(index, 1);
      return true;
    },
  },

  User: {
    posts: (user) => posts.filter(p => p.authorId === user.id),
  },

  Post: {
    author: (post) => users.find(u => u.id === post.authorId),
  },
};

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

startStandaloneServer(server, {
  listen: { port: 4000 },
}).then(({ url }) => {
  console.log(`🚀 Server ready at ${url}`);
});

3. 查询数据（客户端）

// Using Apollo Client
import { ApolloClient, InMemoryCache, gql } from '@apollo/client';

const client = new ApolloClient({
  uri: 'http://localhost:4000',
  cache: new InMemoryCache(),
});

// Query
const GET_USER = gql`
  query GetUser($id: ID!) {
    user(id: $id) {
      id
      name
      email
      posts {
        id
        title
        publishedAt
      }
    }
  }
`;

const { data } = await client.query({
  query: GET_USER,
  variables: { id: '1' },
});

// Mutation
const CREATE_POST = gql`
  mutation CreatePost($title: String!, $content: String!, $authorId: ID!) {
    createPost(title: $title, content: $content, authorId: $authorId) {
      id
      title
      content
    }
  }
`;

const { data: postData } = await client.mutate({
  mutation: CREATE_POST,
  variables: {
    title: 'New Post',
    content: 'Hello GraphQL!',
    authorId: '1',
  },
});

---

核心概念

GraphQL 基础

• 架构优先设计：使用架构定义语言（SDL）定义 API 契约
• 类型安全：强类型架构在运行时和构建时受到强制约束
• 单一端点：所有查询和变更都通过同一个 URL（例如 /graphql）
• 客户端指定查询：客户端精确请求所需数据
• 层级数据：查询结构与返回数据的结构相对应
• 自省：架构自带文档说明，且可查询

操作

# Query - Read data (GET-like)
query GetUser {
  user(id: "1") {
    name
  }
}

# Mutation - Modify data (POST/PUT/DELETE-like)
mutation CreateUser {
  createUser(name: "Alice", email: "[email protected]") {
    id
    name
  }
}

# Subscription - Real-time updates (WebSocket)
subscription OnPostCreated {
  postCreated {
    id
    title
    author {
      name
    }
  }
}

字段和参数

type Query {
  # Field with arguments
  user(id: ID!): User
  users(limit: Int = 10, offset: Int = 0): [User!]!

  # Search with multiple arguments
  searchPosts(
    query: String!
    category: String
    limit: Int = 20
  ): [Post!]!
}

---

架构定义语言（SDL）

基本类型定义

type User {
  id: ID!              # Non-null ID scalar
  name: String!        # Non-null String
  email: String!
  age: Int             # Nullable Int
  isActive: Boolean!
  posts: [Post!]!      # Non-null list of non-null Posts
  profile: Profile     # Nullable object type
}

type Profile {
  bio: String
  avatarUrl: String
  website: String
}

type Post {
  id: ID!
  title: String!
  content: String!
  author: User!
  tags: [String!]      # Non-null list, nullable elements
  publishedAt: DateTime
}

输入类型（用于变更）

input CreateUserInput {
  name: String!
  email: String!
  age: Int
}

input UpdateUserInput {
  name: String
  email: String
  age: Int
}

type Mutation {
  createUser(input: CreateUserInput!): User!
  updateUser(id: ID!, input: UpdateUserInput!): User!
}

接口

interface Node {
  id: ID!
  createdAt: DateTime!
  updatedAt: DateTime!
}

type User implements Node {
  id: ID!
  createdAt: DateTime!
  updatedAt: DateTime!
  name: String!
  email: String!
}

type Post implements Node {
  id: ID!
  createdAt: DateTime!
  updatedAt: DateTime!
  title: String!
  content: String!
}

type Query {
  node(id: ID!): Node  # Can return User or Post
}

联合类型

union SearchResult = User | Post | Comment

type Query {
  search(query: String!): [SearchResult!]!
}

# Client query with fragments
query Search {
  search(query: "graphql") {
    ... on User {
      name
      email
    }
    ... on Post {
      title
      content
    }
    ... on Comment {
      text
      author { name }
    }
  }
}

枚举

enum Role {
  ADMIN
  MODERATOR
  USER
  GUEST
}

enum PostStatus {
  DRAFT
  PUBLISHED
  ARCHIVED
}

type User {
  id: ID!
  name: String!
  role: Role!
}

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

---

类型系统

标量类型

# Built-in scalars
scalar Int       # Signed 32-bit integer
scalar Float     # Signed double-precision floating-point
scalar String    # UTF-8 character sequence
scalar Boolean   # true or false
scalar ID        # Unique identifier (serialized as String)

# Custom scalars
scalar DateTime  # ISO 8601 timestamp
scalar Email     # Email address
scalar URL       # Valid URL
scalar JSON      # Arbitrary JSON
scalar Upload    # File upload

自定义标量实现

// DateTime scalar (TypeScript)
import { GraphQLScalarType, Kind } from 'graphql';

const DateTimeScalar = new GraphQLScalarType({
  name: 'DateTime',
  description: 'ISO 8601 DateTime',

  // Serialize to client (output)
  serialize(value: Date) {
    return value.toISOString();
  },

  // Parse from client (input)
  parseValue(value: string) {
    return new Date(value);
  },

  // Parse from query literal
  parseLiteral(ast) {
    if (ast.kind === Kind.STRING) {
      return new Date(ast.value);
    }
    return null;
  },
});

// Add to resolvers
const resolvers = {
  DateTime: DateTimeScalar,
  // ... other resolvers
};

非空和列表

type User {
  name: String!           # Non-null String
  email: String           # Nullable String

  tags: [String!]!        # Non-null list of non-null Strings
  friends: [User!]        # Nullable list of non-null Users
  posts: [Post]!          # Non-null list of nullable Posts
  comments: [Comment]     # Nullable list of nullable Comments
}

---

查询和变更

查询变量

// Define query with variables
const GET_USER = gql`
  query GetUser($id: ID!, $includePosts: Boolean = false) {
    user(id: $id) {
      id
      name
      email
      posts @include(if: $includePosts) {
        id
        title
      }
    }
  }
`;

// Execute with variables
const { data } = await client.query({
  query: GET_USER,
  variables: {
    id: '1',
    includePosts: true,
  },
});

别名

query {
  # Fetch same field with different arguments
  user1: user(id: "1") {
    name
  }
  user2: user(id: "2") {
    name
  }

  # Alias for clarity
  currentUser: me {
    id
    name
  }
}

片段

# Define reusable fragment
fragment UserFields on User {
  id
  name
  email
  createdAt
}

fragment PostSummary on Post {
  id
  title
  publishedAt
  author {
    ...UserFields
  }
}

# Use fragments in query
query {
  user(id: "1") {
    ...UserFields
    posts {
      ...PostSummary
    }
  }
}

指令

# Built-in directives
query GetUser($id: ID!, $withPosts: Boolean!, $skipEmail: Boolean!) {
  user(id: $id) {
    name
    email @skip(if: $skipEmail)
    posts @include(if: $withPosts) {
      title
    }
  }
}

# Custom directive definition
directive @auth(requires: Role = USER) on FIELD_DEFINITION

type Query {
  users: [User!]! @auth(requires: ADMIN)
  me: User! @auth
}

变更最佳实践

# Single mutation with input type
type Mutation {
  createPost(input: CreatePostInput!): CreatePostPayload!
  updatePost(id: ID!, input: UpdatePostInput!): UpdatePostPayload!
  deletePost(id: ID!): DeletePostPayload!
}

input CreatePostInput {
  title: String!
  content: String!
  categoryId: ID!
  tags: [String!]
}

# Payload pattern for mutations
type CreatePostPayload {
  post: Post          # Created resource
  userErrors: [UserError!]!  # Client errors
  success: Boolean!
}

type UserError {
  field: String!      # Which field caused error
  message: String!    # Human-readable message
}

---

解析器和 DataLoader

解析器签名

type Resolver<TParent, TArgs, TContext, TResult> = (
  parent: TParent,      // Parent object
  args: TArgs,          // Field arguments
  context: TContext,    // Shared context (auth, db, etc.)
  info: GraphQLResolveInfo  // Query metadata
) => TResult | Promise<TResult>;

基本解析器

const resolvers = {
  Query: {
    user: async (_, { id }, { db }) => {
      return db.users.findById(id);
    },

    users: async (_, { limit = 10, offset = 0 }, { db }) => {
      return db.users.findMany({ limit, offset });
    },
  },

  Mutation: {
    createUser: async (_, { input }, { db, userId }) => {
      if (!userId) {
        throw new Error('Authentication required');
      }

      const user = await db.users.create(input);
      return { user, userErrors: [], success: true };
    },
  },

  User: {
    // Field resolver - only called if client requests 'posts'
    posts: async (user, _, { db }) => {
      return db.posts.findByAuthorId(user.id);
    },

    // Computed field
    fullName: (user) => {
      return `${user.firstName} ${user.lastName}`;
    },
  },
};

N+1 问题

// ❌ BAD - N+1 queries
const resolvers = {
  Query: {
    users: () => db.users.findMany(), // 1 query
  },
  User: {
    // Called for EACH user - N queries!
    posts: (user) => db.posts.findByAuthorId(user.id),
  },
};

// Querying 100 users = 1 + 100 = 101 database queries!

DataLoader 解决方案

import DataLoader from 'dataloader';

// Batch function - receives array of keys
async function batchLoadPosts(authorIds: string[]) {
  const posts = await db.posts.findByAuthorIds(authorIds);

  // Group by author ID
  const postsByAuthor = authorIds.map(authorId =>
    posts.filter(post => post.authorId === authorId)
  );

  return postsByAuthor;
}

// Create context with loaders
function createContext({ req }) {
  return {
    db,
    userId: req.userId,
    loaders: {
      posts: new DataLoader(batchLoadPosts),
    },
  };
}

// ✅ GOOD - Batched queries
const resolvers = {
  Query: {
    users: () => db.users.findMany(), // 1 query
  },
  User: {
    // Uses DataLoader - batches all requests into 1 query!
    posts: (user, _, { loaders }) => {
      return loaders.posts.load(user.id);
    },
  },
};

// Querying 100 users = 1 + 1 = 2 database queries!

高级 DataLoader 模式

// DataLoader with caching
const userLoader = new DataLoader(
  async (ids) => {
    const users = await db.users.findByIds(ids);
    return ids.map(id => users.find(u => u.id === id));
  },
  {
    cache: true,           // Enable caching (default)
    maxBatchSize: 100,     // Limit batch size
    batchScheduleFn: (cb) => setTimeout(cb, 10), // Debounce batching
  }
);

// Cache manipulation
userLoader.clear(id);              // Clear single key
userLoader.clearAll();             // Clear entire cache
userLoader.prime(id, user);        // Prime cache with value

---

订阅

WebSocket 设置（Apollo Server）

import { ApolloServer } from '@apollo/server';
import { expressMiddleware } from '@apollo/server/express4';
import { ApolloServerPluginDrainHttpServer } from '@apollo/server/plugin/drainHttpServer';
import { createServer } from 'http';
import { WebSocketServer } from 'ws';
import { useServer } from 'graphql-ws/lib/use/ws';
import { makeExecutableSchema } from '@graphql-tools/schema';
import express from 'express';

const app = express();
const httpServer = createServer(app);

const schema = makeExecutableSchema({ typeDefs, resolvers });

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

const serverCleanup = useServer({ schema }, wsServer);

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

await server.start();
app.use('/graphql', express.json(), expressMiddleware(server));

httpServer.listen(4000);

订阅架构

type Subscription {
  postCreated: Post!
  postUpdated(id: ID!): Post!
  messageAdded(channelId: ID!): Message!
  userStatusChanged(userId: ID!): UserStatus!
}

type Message {
  id: ID!
  text: String!
  author: User!
  channelId: ID!
  createdAt: DateTime!
}

enum UserStatus {
  ONLINE
  OFFLINE
  AWAY
}

订阅解析器（PubSub）

import { PubSub } from 'graphql-subscriptions';

const pubsub = new PubSub();

const resolvers = {
  Mutation: {
    createPost: async (_, { input }, { db }) => {
      const post = await db.posts.create(input);

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

      return { post, success: true, userErrors: [] };
    },

    sendMessage: async (_, { channelId, text }, { db, userId }) => {
      const message = await db.messages.create({
        channelId,
        text,
        authorId: userId,
      });

      pubsub.publish(`MESSAGE_${channelId}`, {
        messageAdded: message,
      });

      return message;
    },
  },

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

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

    messageAdded: {
      subscribe: (_, { channelId }) => {
        return pubsub.asyncIterator([`MESSAGE_${channelId}`]);
      },
    },
  },
};

客户端订阅（Apollo Client）

import { ApolloClient, InMemoryCache, split, HttpLink } from '@apollo/client';
import { GraphQLWsLink } from '@apollo/client/link/subscriptions';
import { getMainDefinition } from '@apollo/client/utilities';
import { createClient } from 'graphql-ws';

// HTTP link for queries and mutations
const httpLink = new HttpLink({
  uri: 'http://localhost:4000/graphql',
});

// WebSocket link for subscriptions
const wsLink = new GraphQLWsLink(
  createClient({
    url: 'ws://localhost:4000/graphql',
  })
);

// Split based on operation type
const splitLink = split(
  ({ query }) => {
    const definition = getMainDefinition(query);
    return (
      definition.kind === 'OperationDefinition' &&
      definition.operation === 'subscription'
    );
  },
  wsLink,
  httpLink
);

const client = new ApolloClient({
  link: splitLink,
  cache: new InMemoryCache(),
});

// Use subscription
const MESSAGES_SUBSCRIPTION = gql`
  subscription OnMessageAdded($channelId: ID!) {
    messageAdded(channelId: $channelId) {
      id
      text
      author {
        name
      }
      createdAt
    }
  }
`;

function ChatComponent({ channelId }) {
  const { data, loading } = useSubscription(MESSAGES_SUBSCRIPTION, {
    variables: { channelId },
  });

  if (loading) return <p>Loading...</p>;

  return <div>New message: {data.messageAdded.text}</div>;
}

Redis PubSub（生产环境）

import { RedisPubSub } from 'graphql-redis-subscriptions';
import Redis from 'ioredis';

const options = {
  host: 'localhost',
  port: 6379,
  retryStrategy: (times) => Math.min(times * 50, 2000),
};

const pubsub = new RedisPubSub({
  publisher: new Redis(options),
  subscriber: new Redis(options),
});

// Use same as in-memory PubSub
pubsub.publish('POST_CREATED', { postCreated: post });
pubsub.asyncIterator(['POST_CREATED']);

---

错误处理

错误类型

import { GraphQLError } from 'graphql';

// Custom error classes
class AuthenticationError extends GraphQLError {
  constructor(message: string) {
    super(message, {
      extensions: {
        code: 'UNAUTHENTICATED',
        http: { status: 401 },
      },
    });
  }
}

class ForbiddenError extends GraphQLError {
  constructor(message: string) {
    super(message, {
      extensions: {
        code: 'FORBIDDEN',
        http: { status: 403 },
      },
    });
  }
}

class ValidationError extends GraphQLError {
  constructor(message: string, invalidFields: Record<string, string>) {
    super(message, {
      extensions: {
        code: 'BAD_USER_INPUT',
        invalidFields,
      },
    });
  }
}

在解析器中抛出错误

const resolvers = {
  Query: {
    user: async (_, { id }, { db, userId }) => {
      if (!userId) {
        throw new AuthenticationError('Must be logged in');
      }

      const user = await db.users.findById(id);
      if (!user) {
        throw new GraphQLError('User not found', {
          extensions: { code: 'NOT_FOUND' },
        });
      }

      return user;
    },
  },

  Mutation: {
    createPost: async (_, { input }, { db, userId }) => {
      const errors: Record<string, string> = {};

      if (!input.title || input.title.length < 3) {
        errors.title = 'Title must be at least 3 characters';
      }

      if (!input.content) {
        errors.content = 'Content is required';
      }

      if (Object.keys(errors).length > 0) {
        throw new ValidationError('Invalid input', errors);
      }

      const post = await db.posts.create({ ...input, authorId: userId });
      return { post, success: true, userErrors: [] };
    },
  },
};

错误响应格式

{
  "errors": [
    {
      "message": "User not found",
      "locations": [{ "line": 2, "column": 3 }],
      "path": ["user"],
      "extensions": {
        "code": "NOT_FOUND"
      }
    }
  ],
  "data": {
    "user": null
  }
}

字段级错误处理

// Nullable fields allow partial results
type Query {
  user(id: ID!): User        # null on error
  users: [User!]!            # throws on error
  post(id: ID!): Post        # null on error
}

// Resolver can return null instead of throwing
const resolvers = {
  Query: {
    user: async (_, { id }, { db }) => {
      try {
        return await db.users.findById(id);
      } catch (error) {
        console.error('Failed to fetch user:', error);
        return null;  // Returns null instead of error
      }
    },
  },
};

---

架构设计模式

Relay 游标连接

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

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

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

type Query {
  posts(
    first: Int
    after: String
    last: Int
    before: String
  ): PostConnection!
}

Relay 连接解析器

import { fromGlobalId, toGlobalId } from 'graphql-relay';

function encodeCursor(id: string): string {
  return Buffer.from(`cursor:${id}`).toString('base64');
}

function decodeCursor(cursor: string): string {
  return Buffer.from(cursor, 'base64').toString('utf-8').replace('cursor:', '');
}

const resolvers = {
  Query: {
    posts: async (_, { first = 10, after }, { db }) => {
      const startId = after ? decodeCursor(after) : null;

      // Fetch first + 1 to determine hasNextPage
      const posts = await db.posts.findMany({
        where: startId ? { id: { gt: startId } } : {},
        take: first + 1,
        orderBy: { createdAt: 'desc' },
      });

      const hasNextPage = posts.length > first;
      const nodes = hasNextPage ? posts.slice(0, -1) : posts;

      const edges = nodes.map(node => ({
        node,
        cursor: encodeCursor(node.id),
      }));

      return {
        edges,
        pageInfo: {
          hasNextPage,
          hasPreviousPage: !!after,
          startCursor: edges[0]?.cursor,
          endCursor: edges[edges.length - 1]?.cursor,
        },
        totalCount: await db.posts.count(),
      };
    },
  },
};

偏移分页（更简单）

type PostsResponse {
  posts: [Post!]!
  total: Int!
  hasMore: Boolean!
}

type Query {
  posts(limit: Int = 10, offset: Int = 0): PostsResponse!
}

全局对象标识（Relay）

interface Node {
  id: ID!  # Global unique ID
}

type User implements Node {
  id: ID!
  name: String!
}

type Post implements Node {
  id: ID!
  title: String!
}

type Query {
  node(id: ID!): Node
}

const resolvers = {
  Query: {
    node: async (_, { id }, { db }) => {
      const { type, id: rawId } = fromGlobalId(id);

      switch (type) {
        case 'User':
          return db.users.findById(rawId);
        case 'Post':
          return db.posts.findById(rawId);
        default:
          return null;
      }
    },
  },

  User: {
    id: (user) => toGlobalId('User', user.id),
  },

  Post: {
    id: (post) => toGlobalId('Post', post.id),
  },
};

---

服务端实现

Apollo Server（TypeScript）

import { ApolloServer } from '@apollo/server';
import { startStandaloneServer } from '@apollo/server/standalone';

const server = new ApolloServer({
  typeDefs,
  resolvers,
  introspection: process.env.NODE_ENV !== 'production',
  plugins: [
    // Custom plugin
    {
      async requestDidStart() {
        return {
          async willSendResponse({ response }) {
            console.log('Response:', response);
          },
        };
      },
    },
  ],
});

const { url } = await startStandaloneServer(server, {
  listen: { port: 4000 },
  context: async ({ req }) => ({
    token: req.headers.authorization,
    db: database,
  }),
});

GraphQL Yoga（现代替代方案）

import { createYoga, createSchema } from 'graphql-yoga';
import { createServer } from 'node:http';

const yoga = createYoga({
  schema: createSchema({
    typeDefs,
    resolvers,
  }),
  graphiql: true,
  context: ({ request }) => ({
    userId: request.headers.get('x-user-id'),
  }),
});

const server = createServer(yoga);
server.listen(4000, () => {
  console.log('Server on http://localhost:4000/graphql');
});

Graphene（Python/Django）

import graphene
from graphene_django import DjangoObjectType
from .models import User, Post

class UserType(DjangoObjectType):
    class Meta:
        model = User
        fields = '__all__'

class PostType(DjangoObjectType):
    class Meta:
        model = Post
        fields = '__all__'

class Query(graphene.ObjectType):
    users = graphene.List(UserType)
    user = graphene.Field(UserType, id=graphene.ID(required=True))
    posts = graphene.List(PostType)

    def resolve_users(self, info):
        return User.objects.all()

    def resolve_user(self, info, id):
        return User.objects.get(pk=id)

    def resolve_posts(self, info):
        return Post.objects.all()

class CreateUser(graphene.Mutation):
    class Arguments:
        name = graphene.String(required=True)
        email = graphene.String(required=True)

    user = graphene.Field(UserType)
    success = graphene.Boolean()

    def mutate(self, info, name, email):
        user = User.objects.create(name=name, email=email)
        return CreateUser(user=user, success=True)

class Mutation(graphene.ObjectType):
    create_user = CreateUser.Field()

schema = graphene.Schema(query=Query, mutation=Mutation)

Strawberry（Python，现代方案）

import strawberry
from typing import List, Optional
from datetime import datetime

@strawberry.type
class User:
    id: strawberry.ID
    name: str
    email: str
    created_at: datetime

@strawberry.type
class Post:
    id: strawberry.ID
    title: str
    content: str
    author_id: strawberry.ID

@strawberry.input
class CreateUserInput:
    name: str
    email: str

@strawberry.type
class Query:
    @strawberry.field
    def users(self) -> List[User]:
        return User.objects.all()

    @strawberry.field
    def user(self, id: strawberry.ID) -> Optional[User]:
        return User.objects.get(pk=id)

@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_user(self, input: CreateUserInput) -> User:
        return User.objects.create(**input.__dict__)

schema = strawberry.Schema(query=Query, mutation=Mutation)

# FastAPI integration
from strawberry.fastapi import GraphQLRouter

app = FastAPI()
app.include_router(GraphQLRouter(schema), prefix="/graphql")

---

客户端集成

Apollo Client（React）

import { ApolloClient, InMemoryCache, ApolloProvider, useQuery, useMutation } from '@apollo/client';

const client = new ApolloClient({
  uri: 'http://localhost:4000/graphql',
  cache: new InMemoryCache(),
});

function App() {
  return (
    <ApolloProvider client={client}>
      <UserList />
    </ApolloProvider>
  );
}

function UserList() {
  const { loading, error, data, refetch } = useQuery(GET_USERS);
  const [createUser] = useMutation(CREATE_USER, {
    refetchQueries: [{ query: GET_USERS }],
  });

  if (loading) return <p>Loading...</p>;
  if (error) return <p>Error: {error.message}</p>;

  return (
    <div>
      {data.users.map(user => (
        <div key={user.id}>{user.name}</div>
      ))}
      <button onClick={() => createUser({ variables: { name: 'New User' } })}>
        Add User
      </button>
    </div>
  );
}

urql（轻量级替代方案）

import { createClient, Provider, useQuery, useMutation } from 'urql';

const client = createClient({
  url: 'http://localhost:4000/graphql',
});

function App() {
  return (
    <Provider value={client}>
      <UserList />
    </Provider>
  );
}

function UserList() {
  const [result, reexecuteQuery] = useQuery({ query: GET_USERS });
  const [, createUser] = useMutation(CREATE_USER);

  const { data, fetching, error } = result;

  if (fetching) return <p>Loading...</p>;
  if (error) return <p>Error: {error.message}</p>;

  return (
    <div>
      {data.users.map(user => (
        <div key={user.id}>{user.name}</div>
      ))}
    </div>
  );
}

graphql-request（最精简）

import { GraphQLClient, gql } from 'graphql-request';

const client = new GraphQLClient('http://localhost:4000/graphql');

async function fetchUsers() {
  const query = gql`
    query {
      users {
        id
        name
      }
    }
  `;

  const data = await client.request(query);
  return data.users;
}

async function createUser(name: string) {
  const mutation = gql`
    mutation CreateUser($name: String!) {
      createUser(name: $name) {
        id
        name
      }
    }
  `;

  const data = await client.request(mutation, { name });
  return data.createUser;
}

TanStack Query + GraphQL

import { useQuery, useMutation, QueryClient, QueryClientProvider } from '@tanstack/react-query';
import { GraphQLClient } from 'graphql-request';

const graphQLClient = new GraphQLClient('http://localhost:4000/graphql');

function useUsers() {
  return useQuery({
    queryKey: ['users'],
    queryFn: async () => {
      const { users } = await graphQLClient.request(GET_USERS);
      return users;
    },
  });
}

function useCreateUser() {
  const queryClient = useQueryClient();

  return useMutation({
    mutationFn: async (name: string) => {
      const { createUser } = await graphQLClient.request(CREATE_USER, { name });
      return createUser;
    },
    onSuccess: () => {
      queryClient.invalidateQueries({ queryKey: ['users'] });
    },
  });
}

---

TypeScript 代码生成

GraphQL Code Generator 设置

npm install -D @graphql-codegen/cli @graphql-codegen/typescript @graphql-codegen/typescript-operations @graphql-codegen/typescript-react-apollo

codegen.yml

schema: http://localhost:4000/graphql
documents: 'src/**/*.graphql'
generates:
  src/generated/graphql.ts:
    plugins:
      - typescript
      - typescript-operations
      - typescript-react-apollo
    config:
      withHooks: true
      withComponent: false
      skipTypename: false
      enumsAsTypes: true

生成的类型

// src/queries/users.graphql
// query GetUsers {
//   users {
//     id
//     name
//     email
//   }
// }

// Generated types
export type GetUsersQuery = {
  __typename?: 'Query';
  users: Array<{
    __typename?: 'User';
    id: string;
    name: string;
    email: string;
  }>;
};

export function useGetUsersQuery(
  baseOptions?: Apollo.QueryHookOptions<GetUsersQuery, GetUsersQueryVariables>
) {
  return Apollo.useQuery<GetUsersQuery, GetUsersQueryVariables>(
    GetUsersDocument,
    baseOptions
  );
}

使用生成的类型

import { useGetUsersQuery, useCreateUserMutation } from './generated/graphql';

function UserList() {
  const { data, loading, error } = useGetUsersQuery();
  const [createUser] = useCreateUserMutation();

  // Fully typed!
  const users = data?.users; // Type: User[] | undefined
}

---

身份验证和授权

基于上下文的身份验证

import jwt from 'jsonwebtoken';

async function createContext({ req }) {
  const token = req.headers.authorization?.replace('Bearer ', '');

  if (!token) {
    return { userId: null, db };
  }

  try {
    const { userId } = jwt.verify(token, process.env.JWT_SECRET);
    return { userId, db };
  } catch (error) {
    return { userId: null, db };
  }
}

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

await startStandaloneServer(server, {
  context: createContext,
});

解析器级身份验证

const resolvers = {
  Query: {
    me: (_, __, { userId }) => {
      if (!userId) {
        throw new AuthenticationError('Not authenticated');
      }
      return db.users.findById(userId);
    },

    users: (_, __, { userId, db }) => {
      if (!userId) {
        throw new AuthenticationError('Not authenticated');
      }

      const user = db.users.findById(userId);
      if (user.role !== 'ADMIN') {
        throw new ForbiddenError('Admin access required');
      }

      return db.users.findMany();
    },
  },
};

基于指令的身份验证

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

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

  enum Role {
    ADMIN
    USER
  }

  type Query {
    users: [User!]! @auth(requires: ADMIN)
    me: User! @auth
  }
`;

// Directive transformer
function authDirectiveTransformer(schema, directiveName = 'auth') {
  return mapSchema(schema, {
    [MapperKind.OBJECT_FIELD]: (fieldConfig) => {
      const authDirective = getDirective(schema, fieldConfig, directiveName)?.[0];

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

        fieldConfig.resolve = async (source, args, context, info) => {
          if (!context.userId) {
            throw new AuthenticationError('Not authenticated');
          }

          if (requires) {
            const user = await context.db.users.findById(context.userId);
            if (user.role !== requires) {
              throw new ForbiddenError(`${requires} role required`);
            }
          }

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

      return fieldConfig;
    },
  });
}

let schema = makeExecutableSchema({ typeDefs, resolvers });
schema = authDirectiveTransformer(schema);

---

性能优化

查询复杂度分析

import { createComplexityLimitRule } from 'graphql-validation-complexity';

const server = new ApolloServer({
  typeDefs,
  resolvers,
  validationRules: [
    createComplexityLimitRule(1000, {
      scalarCost: 1,
      objectCost: 2,
      listFactor: 10,
    }),
  ],
});

持久化查询（APQ）

import { ApolloServer } from '@apollo/server';
import { ApolloServerPluginInlineTraceDisabled } from '@apollo/server/plugin/disabled';

const server = new ApolloServer({
  typeDefs,
  resolvers,
  persistedQueries: {
    cache: new Map(), // Or Redis
  },
});

// Client sends hash instead of full query
// Reduces payload size by ~80%

响应缓存

import responseCachePlugin from '@apollo/server-plugin-response-cache';

const server = new ApolloServer({
  typeDefs,
  resolvers,
  plugins: [
    responseCachePlugin({
      sessionId: (context) => context.userId || null,
      shouldReadFromCache: (context) => !context.userId, // Cache only public queries
    }),
  ],
});

// Schema directive for cache control
const typeDefs = gql`
  type Query {
    posts: [Post!]! @cacheControl(maxAge: 60)
    user(id: ID!): User @cacheControl(maxAge: 30)
  }
`;

字段级缓存

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: {
    user: async (_, { id }, { cache, db }) => {
      const cacheKey = `user:${id}`;
      const cached = await cache.get(cacheKey);

      if (cached) {
        return JSON.parse(cached);
      }

      const user = await db.users.findById(id);
      await cache.set(cacheKey, JSON.stringify(user), { ttl: 60 });

      return user;
    },
  },
};

---

文件上传

架构

scalar Upload

type Mutation {
  uploadFile(file: Upload!): File!
  uploadMultiple(files: [Upload!]!): [File!]!
}

type File {
  filename: String!
  mimetype: String!
  encoding: String!
  url: String!
}

服务端（graphql-upload）

import graphqlUploadExpress from 'graphql-upload/graphqlUploadExpress.mjs';
import { GraphQLUpload } from 'graphql-upload/GraphQLUpload.mjs';
import fs from 'fs';
import path from 'path';

app.use('/graphql', graphqlUploadExpress({ maxFileSize: 10000000, maxFiles: 10 }));

const resolvers = {
  Upload: GraphQLUpload,

  Mutation: {
    uploadFile: async (_, { file }) => {
      const { createReadStream, filename, mimetype, encoding } = await file;

      const stream = createReadStream();
      const uploadPath = path.join(__dirname, 'uploads', filename);

      await new Promise((resolve, reject) => {
        stream
          .pipe(fs.createWriteStream(uploadPath))
          .on('finish', resolve)
          .on('error', reject);
      });

      return {
        filename,
        mimetype,
        encoding,
        url: `/uploads/${filename}`,
      };
    },
  },
};

客户端（Apollo Client）

import { useMutation } from '@apollo/client';

const UPLOAD_FILE = gql`
  mutation UploadFile($file: Upload!) {
    uploadFile(file: $file) {
      filename
      url
    }
  }
`;

function FileUpload() {
  const [uploadFile] = useMutation(UPLOAD_FILE);

  const handleFileChange = (e: React.ChangeEvent<HTMLInputElement>) => {
    const file = e.target.files?.[0];
    if (!file) return;

    uploadFile({ variables: { file } });
  };

  return <input type="file" onChange={handleFileChange} />;
}

---

测试

解析器单元测试

import { describe, it, expect, vi } from 'vitest';

describe('User Resolvers', () => {
  it('should fetch user by ID', async () => {
    const mockDb = {
      users: {
        findById: vi.fn().mockResolvedValue({
          id: '1',
          name: 'Alice',
          email: '[email protected]',
        }),
      },
    };

    const result = await resolvers.Query.user(
      null,
      { id: '1' },
      { db: mockDb, userId: '1' },
      {} as any
    );

    expect(result.name).toBe('Alice');
    expect(mockDb.users.findById).toHaveBeenCalledWith('1');
  });

  it('should throw error when not authenticated', async () => {
    await expect(
      resolvers.Query.me(null, {}, { userId: null }, {} as any)
    ).rejects.toThrow('Not authenticated');
  });
});

集成测试（Apollo Server）

import { ApolloServer } from '@apollo/server';
import assert from 'assert';

it('fetches users', async () => {
  const server = new ApolloServer({ typeDefs, resolvers });

  const response = await server.executeOperation({
    query: 'query { users { id name } }',
  });

  assert(response.body.kind === 'single');
  expect(response.body.singleResult.errors).toBeUndefined();
  expect(response.body.singleResult.data?.users).toHaveLength(2);
});

it('creates user', async () => {
  const response = await server.executeOperation({
    query: `
      mutation CreateUser($input: CreateUserInput!) {
        createUser(input: $input) {
          user { id name }
          success
        }
      }
    `,
    variables: {
      input: { name: 'Charlie', email: '[email protected]' },
    },
  });

  assert(response.body.kind === 'single');
  expect(response.body.singleResult.data?.createUser.success).toBe(true);
});

端到端测试（Supertest）

import request from 'supertest';
import { app } from './server';

describe('GraphQL API', () => {
  it('should query users', async () => {
    const response = await request(app)
      .post('/graphql')
      .send({
        query: '{ users { id name } }',
      })
      .expect(200);

    expect(response.body.data.users).toBeDefined();
  });

  it('should require authentication', async () => {
    const response = await request(app)
      .post('/graphql')
      .send({
        query: '{ me { id } }',
      })
      .expect(200);

    expect(response.body.errors[0].extensions.code).toBe('UNAUTHENTICATED');
  });
});

---

生产环境模式

架构拼接

import { stitchSchemas } from '@graphql-tools/stitch';

const userSchema = makeExecutableSchema({ typeDefs: userTypeDefs, resolvers: userResolvers });
const postSchema = makeExecutableSchema({ typeDefs: postTypeDefs, resolvers: postResolvers });

const schema = stitchSchemas({
  subschemas: [
    { schema: userSchema },
    { schema: postSchema },
  ],
});

Apollo 联邦

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

const typeDefs = gql`
  type User @key(fields: "id") {
    id: ID!
    name: String!
    email: String!
  }
`;

const resolvers = {
  User: {
    __resolveReference(user, { db }) {
      return db.users.findById(user.id);
    },
  },
};

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

// Post service
const typeDefs = gql`
  extend type User @key(fields: "id") {
    id: ID! @external
    posts: [Post!]!
  }

  type Post @key(fields: "id") {
    id: ID!
    title: String!
    authorId: ID!
  }
`;

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

const gateway = new ApolloGateway({
  supergraphSdl: readFileSync('./supergraph.graphql', 'utf-8'),
});

const server = new ApolloServer({ gateway });

速率限制

import { GraphQLRateLimitDirective } from 'graphql-rate-limit-directive';

const typeDefs = gql`
  directive @rateLimit(
    limit: Int = 10
    duration: Int = 60
  ) on FIELD_DEFINITION

  type Query {
    users: [User!]! @rateLimit(limit: 100, duration: 60)
    search(query: String!): [Result!]! @rateLimit(limit: 10, duration: 60)
  }
`;

let schema = makeExecutableSchema({ typeDefs, resolvers });
schema = GraphQLRateLimitDirective()(schema);

监控（Apollo Studio）

import { ApolloServerPluginUsageReporting } from '@apollo/server/plugin/usageReporting';

const server = new ApolloServer({
  typeDefs,
  resolvers,
  plugins: [
    ApolloServerPluginUsageReporting({
      sendVariableValues: { all: true },
      sendHeaders: { all: true },
    }),
  ],
});

---

框架集成

Next.js App Router

// app/api/graphql/route.ts
import { ApolloServer } from '@apollo/server';
import { startServerAndCreateNextHandler } from '@as-integrations/next';

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

const handler = startServerAndCreateNextHandler(server);

export { handler as GET, handler as POST };

使用 Apollo Client 的 Next.js（SSR）

// lib/apollo-client.ts
import { ApolloClient, InMemoryCache, HttpLink } from '@apollo/client';
import { registerApolloClient } from '@apollo/experimental-nextjs-app-support/rsc';

export const { getClient } = registerApolloClient(() => {
  return new ApolloClient({
    cache: new InMemoryCache(),
    link: new HttpLink({
      uri: 'http://localhost:4000/graphql',
    }),
  });
});

// app/users/page.tsx
import { getClient } from '@/lib/apollo-client';
import { gql } from '@apollo/client';

export default async function UsersPage() {
  const { data } = await getClient().query({
    query: gql`
      query GetUsers {
        users {
          id
          name
        }
      }
    `,
  });

  return (
    <div>
      {data.users.map(user => (
        <div key={user.id}>{user.name}</div>
      ))}
    </div>
  );
}

FastAPI + Strawberry

from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter

app = FastAPI()

graphql_app = GraphQLRouter(schema)
app.include_router(graphql_app, prefix="/graphql")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

---

与 REST 和 tRPC 的比较

GraphQL 与 REST 的比较

特性	GraphQL	REST
端点	单个端点	多个端点
数据获取	客户端指定字段	服务端决定响应内容
过度获取	否	是（额外字段）
获取不足	否	是（多个请求）
版本管理	无需	需要（v1、v2）
缓存	复杂	简单（HTTP 缓存）
类型系统	内置	外部（OpenAPI）
实时性	订阅	SSE/WebSocket

GraphQL 与 tRPC 的比较

特性	GraphQL	tRPC
类型安全	需要代码生成	原生 TypeScript
语言支持	任意语言	仅 TypeScript
客户端与服务端的耦合	松散	紧密
Schema	需要 SDL	从代码推断
学习曲线	陡峭	平缓
工具支持	丰富	不断发展
适用场景	公共 API、移动端	全栈 TypeScript

---

迁移策略

从 REST 逐步迁移到 GraphQL

// 1. Wrap existing REST endpoints
const resolvers = {
  Query: {
    user: async (_, { id }) => {
      const response = await fetch(`https://api.example.com/users/${id}`);
      return response.json();
    },
  },
};

// 2. Add GraphQL layer alongside REST
app.use('/api/rest', restRouter);
app.use('/graphql', graphqlMiddleware);

// 3. Migrate clients incrementally
// 4. Deprecate REST endpoints when ready

为现有应用添加 GraphQL

// Express + GraphQL
import express from 'express';
import { expressMiddleware } from '@apollo/server/express4';

const app = express();

// Existing routes
app.use('/api', existingApiRouter);

// Add GraphQL
app.use('/graphql', express.json(), expressMiddleware(server));

---

最佳实践

Schema 设计

• 使用语义明确的字段名称（created_at 应改为 createdAt）
• 优先使用具体类型，而非通用 JSON
• 使用枚举表示固定值集合
• 根据客户端用例设计，而非数据库结构
• 使用输入类型处理复杂的变更操作
• 为列表实现分页
• 遵循 Relay 的连接规范

Resolver 模式

• 保持 resolver 精简，将处理逻辑委托给服务层
• 使用 DataLoader 处理所有数据库查询
• 在 resolver 中验证输入，而非在数据库层验证
• 在负载中返回错误，而不只是抛出异常
• 使用上下文共享依赖项（db、身份验证、加载器）

错误处理

• 使用带错误代码的自定义错误类型
• 为变更操作返回字段级错误
• 在服务端记录错误，并为客户端清理错误信息
• 使用可空字段支持部分结果
• 不要暴露内部实现细节

性能

• 始终使用 DataLoader 防止 N+1 问题
• 实施查询复杂度限制
• 缓存频繁访问的数据
• 在生产环境中使用持久化查询
• 监控慢查询并进行优化
• 尽可能批量处理变更操作

安全

• 实施身份验证和授权
• 验证所有输入
• 限制查询深度
• 按用户实施速率限制
• 在生产环境中禁用自省（可选）
• 清理错误消息

测试

• 单独测试 resolver
• 模拟外部依赖项
• 测试错误情况
• 对关键流程进行集成测试
• 使用真实客户端进行端到端测试

文档

• 编写清晰的字段描述
• 使用 @deprecated 标注文档中的弃用项
• 在 schema 注释中提供使用示例
• 及时更新 schema 文档

概述

GraphQL 提供了功能强大且灵活的 API 层，具备强类型、高效的数据获取能力和出色的开发者体验。主要优势包括：

• 客户端控制：按需获取数据
• 类型安全：采用优先定义 Schema 的设计，并支持内省
• 单一端点：简化 API 接口
• 实时性：内置订阅支持
• 工具链：生态系统完善（Apollo、Relay、codegen）

需要权衡的因素：

• 对于简单的 CRUD，比 REST 更复杂
• 缓存需要比 HTTP 缓存更多考量
• 对刚接触 GraphQL 的团队来说有一定的学习曲线
• 查询复杂度可能影响性能

适用场景：

• 需要提升带宽利用率的移动应用
• 数据需求多样的复杂前端
• 微服务聚合
• 实时应用
• 多平台客户端（Web、移动端、IoT）

从简单开始：

1. 为核心实体定义 Schema
2. 使用 DataLoader 编写解析器
3. 添加身份验证/授权
4. 实现错误处理
5. 通过缓存进行优化
6. 按需添加订阅
7. 监控并持续迭代

当 API 灵活性和开发者体验是优先事项时，GraphQL 能发挥出色的效果。结合 TypeScript 代码生成，它可以实现从数据库到 UI 的端到端类型安全。