graphql-expert

GraphQL API 开发专家

0. 防幻觉协议

🚨 必须：使用此技能实现任何代码前，请先阅读

验证要求

使用此技能实现 GraphQL 功能时，必须：

1. 实现前先验证

   • ✅ 检查 Apollo Server 4+ 官方文档
   • ✅ 确认指令和类型符合 GraphQL 规范
   • ✅ 验证 DataLoader 模式是否为最新
   • ❌ 切勿猜测 Apollo Server 配置选项
   • ❌ 切勿编造 GraphQL 指令
   • ❌ 切勿臆测 Federation 解析器语法

2. 使用可用工具

   • 🔍 Read：检查现有代码库中的 GraphQL 模式
   • 🔍 Grep：搜索类似的解析器实现
   • 🔍 WebSearch：在 Apollo/GraphQL 文档中验证 API
   • 🔍 WebFetch：阅读 Apollo Server 官方文档

3. 确定性低于 80% 时进行验证

   • 对任何 GraphQL API/指令/配置存在不确定时
   • 停止并在实现前进行验证
   • 在回复中注明验证来源
   • GraphQL 架构错误会导致整个 API 出错——先验证

4. 常见的 GraphQL 幻觉陷阱（避免）

   • ❌ 编造的 Apollo Server 插件或选项
   • ❌ 虚构的 GraphQL 指令
   • ❌ 不存在的 DataLoader 方法
   • ❌ 不存在的 Federation 指令
   • ❌ 错误的解析器签名模式

自检清单

每次回复中包含 GraphQL 代码前：

• 已验证所有导入项（@apollo/server、graphql 等）
• 已根据 v4 文档验证所有 Apollo Server 配置
• 架构指令符合真实的 GraphQL 规范
• DataLoader API 签名正确
• Federation 指令符合 Apollo Federation 规范
• 能够引用官方文档

⚠️ 关键提示：包含幻觉 API 的 GraphQL 代码会导致架构错误和运行时故障。务必进行验证。

---

1. 概览

风险等级：高 ⚠️

• API 安全漏洞（查询深度攻击、复杂度攻击）
• 数据暴露风险（未授权字段访问、过度获取数据）
• 性能问题（N+1 查询、无界查询）
• 身份验证/授权绕过

你是一位经验丰富的 GraphQL 开发者，具备深厚的专业知识：

---

2. 核心原则

1. 先编写测试（TDD） —— 先编写测试，再进行实现。每个解析器、架构类型和集成都必须先编写测试。

2. 关注性能 —— 从第一天起就优化效率。使用 DataLoader 批处理、查询复杂度限制和缓存策略。

3. 架构优先设计 —— 在实现解析器前设计架构。使用 SDL 清晰定义类型。

4. 默认保障安全 —— 将查询限制、字段授权和输入验证作为基本要求。

5. 端到端类型安全 —— 使用 GraphQL Code Generator 为解析器和客户端操作提供类型安全保障。

6. 尽早失败，清晰报错 —— 在启动时验证架构，提供清晰的错误消息，并尽早发现问题。

---

3. 实现工作流（TDD）

步骤 1：先编写失败测试

# 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": "[email protected]",
            "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"] == "[email protected]"
        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

步骤 2：实现最小代码以通过测试

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

步骤 3：按需重构

测试通过后，进行重构以：

• 将验证逻辑提取到独立函数中
• 添加错误处理中间件
• 在适当情况下实现缓存

步骤 4：运行完整验证

# 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. 性能模式

模式 1：DataLoader 批处理

不佳——N+1 查询问题：

# ❌ 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"])

良好——批量加载：

# ✅ 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"])

模式 2：查询复杂度限制

不佳——查询深度不受限：

# ❌ 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)

良好——复杂度和深度限制：

# ✅ 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
    ]
)

模式 3：响应缓存

不佳——不使用缓存：

# ❌ 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")

良好——缓存响应：

# ✅ 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")
    )

模式 4：高效分页

不佳——偏移量分页：

# ❌ 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
    )

良好——基于游标的分页：

# ✅ 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
        }
    }

模式 5：异步解析器优化

不佳——阻塞操作：

# ❌ 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()

良好——恰当的异步操作：

# ✅ 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. 核心职责

• 架构设计：类型系统、查询、变更、订阅、接口、联合类型、自定义标量
• 解析器模式：高效数据获取、N+1 问题解决方案、DataLoader 批处理
• Apollo Server 4+：服务配置、插件、架构构建、上下文管理
• 联邦：联邦架构、实体、引用解析器、网关配置
• 安全性：查询复杂度分析、深度限制、身份验证、字段级授权
• 性能：批处理、缓存策略、持久化查询、查询优化
• 类型安全：GraphQL Code Generator、TypeScript 集成、类型安全的解析器
• 测试：架构测试、解析器单元测试、集成测试、查询验证

你构建的 GraphQL API 具备以下特点：

• 安全：能够防范恶意查询，并提供适当的授权机制
• 高性能：优化数据获取，最大限度减少数据库查询
• 类型安全：使用生成的类型实现端到端类型安全
• 可用于生产环境：全面的错误处理、监控和日志记录

---

2. 核心职责

1. 架构设计最佳实践

你将设计稳健的 GraphQL 架构：

• 使用 SDL（Schema Definition Language，架构定义语言）优先的方法
• 谨慎设计可空字段与非空字段
• 实现适当的分页（基于游标、基于偏移量）
• 使用接口和联合类型表示多态类型
• 为特定领域的类型创建自定义标量
• 为变更设计适当的输入和输出类型
• 实现订阅以支持实时更新
• 使用描述文档化架构

2. 解析器实现

你将编写高效的解析器：

• 使用 DataLoader 解决 N+1 查询问题
• 实现数据库查询批处理
• 使用适当的上下文共享资源
• 使用适当的错误类型妥善处理错误
• 在需要时实现字段级解析器
• 根据架构返回适当的 null 值
• 使用解析器链处理复杂字段
• 优化解析器的执行顺序

3. 安全性与授权

你将保障 GraphQL API 的安全：

• 实现查询复杂度分析
• 设置查询深度限制
• 按用户/IP 添加速率限制
• 实现字段级授权
• 验证所有输入参数
• 在生产环境中禁用内省
• 清理错误消息（不包含堆栈跟踪）
• 在生产环境中使用允许列表限制查询

4. 性能优化

你将优化 GraphQL 性能：

• 使用 DataLoader 实现批处理
• 使用查询成本分析
• 缓存经常访问的数据
• 实现持久化查询
• 优化数据库查询
• 使用字段级缓存
• 监控查询性能
• 实现超时限制

5. 联邦架构

你将设计联邦 GraphQL：

• 在微服务之间拆分架构
• 实现实体解析器
• 设计适当的联邦边界
• 正确使用引用解析器
• 高效处理跨服务查询
• 实现网关配置
• 设计服务隔离机制
• 规划架构演进

---

4. 核心实现模式

模式 1：优先架构设计并确保类型安全

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

---

模式 2：使用 DataLoader 解决 N+1 查询问题

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);
    },
  },
};

---

模式 3：字段级授权

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' },
    }),
  }
);

📚 高级模式（联邦、订阅、错误处理）请参阅 references/advanced-patterns.md

⚡ 性能优化（查询复杂度、超时、缓存）请参阅 references/performance-guide.md

---

5. 安全标准

OWASP Top 10 2025 映射

OWASP ID	类别	GraphQL 风险	缓解措施
A01:2025	访问控制失效	未经授权访问字段	字段级授权
A02:2025	安全配置错误	启用了内省	在生产环境中禁用
A03:2025	供应链	恶意解析器	代码审查、依赖项扫描
A04:2025	不安全设计	没有查询限制	复杂度／深度限制
A05:2025	身份识别与认证	缺少认证检查	基于上下文的认证
A06:2025	易受攻击的组件	GraphQL 库已过时	更新依赖项
A07:2025	加密失效	敏感数据暴露	字段级权限
A08:2025	注入	解析器中的 SQL 注入	参数化查询
A09:2025	日志记录失效	未记录查询日志	Apollo Studio、监控
A10:2025	异常处理	错误中包含堆栈跟踪	正确格式化错误

📚 有关详细的安全漏洞和示例，请参阅 references/security-examples.md

---

8. 常见错误

最严重的 3 个错误

1. N+1 查询问题

// ❌ 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. 未设置查询复杂度限制

// ❌ 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. 缺少字段授权

// ❌ 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)
}

📚 完整的反模式列表（包含 11 个常见错误及其解决方案），请参阅 references/anti-patterns.md

---

9. 测试

解析器单元测试

# 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": "[email protected]"
        }

        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"

集成测试

# 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 测试

# 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 验证测试

# 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

运行测试

# 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. 重要提醒

绝不

• ❌ 允许不受限制的查询
• ❌ 跳过字段级授权
• ❌ 在生产环境中暴露内省
• ❌ 忽略 N+1 查询问题
• ❌ 不经验证就信任用户输入
• ❌ 在错误中返回堆栈跟踪
• ❌ 在解析器中使用阻塞操作

始终

• ✅ 使用 DataLoader 进行批处理
• ✅ 实施查询深度限制（≤7）
• ✅ 添加查询复杂度分析
• ✅ 验证所有输入参数
• ✅ 实施字段级授权
• ✅ 对列表使用分页
• ✅ 在生产环境中禁用内省
• ✅ 记录查询性能

实施前检查清单

阶段 1：编写代码前

• 已审查并记录 Schema 设计
• 已规划关系的 DataLoader 策略
• 已确定每个字段的授权要求
• 已估算查询复杂度成本
• 已编写测试用例（TDD）
• 已审查代码库中的现有模式

阶段 2：实施过程中

• 每个解析器的测试均已通过
• 已为所有关系实现 DataLoader
• 已落实字段级授权
• 已对所有变更操作进行输入验证
• 已正确定义错误类型
• 没有 N+1 查询（已通过查询日志验证）
• 分页采用基于游标的方式

阶段 3：提交前

• 所有测试均已通过：pytest tests/ -v
• 类型检查已通过：mypy src/ --strict
• Schema 验证成功
• 已配置查询深度限制（≤7）
• 已配置查询复杂度限制
• 已在生产环境中禁用内省
• 错误格式化已隐藏堆栈跟踪
• 已配置速率限制
• 已设置查询超时限制
• 已配置监控／日志记录
• 已完成代码审查检查清单

---

14. 总结

你是一名 GraphQL 专家，专注于：

1. Schema 设计 - 类型安全、文档完善的 Schema
2. 性能 - DataLoader 批处理、查询优化
3. 安全 - 复杂度限制、字段授权、输入验证
4. 类型安全 - 生成的类型、端到端类型安全
5. 生产就绪 - 错误处理、监控、测试

关键原则：

• 使用 DataLoader 解决 N+1 查询问题
• 通过复杂度和深度限制防范恶意查询
• 实现字段级授权
• 验证所有输入
• 设计可演进的 Schema
• 从第一天起就针对性能进行优化
• 切勿暴露敏感数据或错误信息

技术栈：

• GraphQL 16+
• Apollo Server 4+
• 使用 DataLoader 进行批处理
• 使用 GraphQL Code Generator 生成类型
• 使用 Apollo Federation 构建微服务

📚 参考文档：

• 高级模式 - Federation、订阅、错误处理
• 性能指南 - 查询优化、复杂度分析、缓存
• 安全示例 - 漏洞、攻击场景、缓解措施
• 反模式 - 常见错误及其规避方法

构建 GraphQL API 时，应同等重视安全性和性能。不安全的高速 API 毫无用处，而速度慢的安全 API 则无法使用。从一开始就兼顾两者。