graphql-expert

GraphQL API 開発エキスパート

0. ハルシネーション防止プロトコル

🚨 必須：このスキルを使ってコードを実装する前にお読みください

検証要件

このスキルを使用して GraphQL 機能を実装する際は、以下を必ず行ってください。

1. 実装前に検証する

   • ✅ Apollo Server 4 以降の公式ドキュメントを確認する
   • ✅ ディレクティブ／型が GraphQL 仕様に準拠していることを確認する
   • ✅ DataLoader のパターンが最新であることを検証する
   • ❌ Apollo Server の設定オプションを推測しない
   • ❌ GraphQL ディレクティブをでっち上げない
   • ❌ フェデレーションのリゾルバー構文を決めつけない

2. 利用可能なツールを使う

   • 🔍 読む：既存のコードベースで GraphQL のパターンを確認する
   • 🔍 Grep：類似するリゾルバーの実装を検索する
   • 🔍 WebSearch：Apollo／GraphQL のドキュメントで API を検証する
   • 🔍 WebFetch：Apollo Server の公式ドキュメントを読む

3. 確信度が 80％ 未満なら検証する

   • GraphQL API／ディレクティブ／設定について少しでも不確かな場合
   • 実装を中断し、先に検証する
   • 回答に検証元を記載する
   • GraphQL スキーマのエラーは API 全体を停止させます。必ず先に検証してください

4. GraphQL でよくあるハルシネーションの落とし穴（回避すること）

   • ❌ 実在しない Apollo Server のプラグインやオプション
   • ❌ 架空の GraphQL ディレクティブ
   • ❌ 実在しない DataLoader のメソッド
   • ❌ 実在しないフェデレーションディレクティブ
   • ❌ 誤ったリゾルバーのシグネチャパターン

セルフチェックリスト

GraphQL コードを含む回答をする前に、毎回確認してください。

• すべてのインポートを検証済み（@apollo/server、graphql など）
• すべての Apollo Server 設定を v4 のドキュメントと照合済み
• スキーマのディレクティブが GraphQL 仕様に実在する
• DataLoader API のシグネチャが正しい
• フェデレーションディレクティブが 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（スキーマ定義言語）を使用したスキーマファーストのアプローチを採用する
• null 許容フィールドと null 非許容フィールドを慎重に設計する
• 適切なページネーション（カーソルベース、オフセットベース）を実装する
• 多態的な型にはインターフェースとユニオンを使用する
• ドメイン固有の型にはカスタムスカラーを作成する
• 適切な入力型と出力型でミューテーションを設計する
• リアルタイム更新にはサブスクリプションを実装する
• 説明を使ってスキーマを文書化する

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

スキーマ検証テスト

# 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：コードを書く前

• スキーマ設計をレビューし、文書化した
• リレーションシップに対する DataLoader 戦略を計画した
• フィールドごとの認可要件を特定した
• クエリの複雑度コストを見積もった
• テストケースを作成した（TDD）
• コードベース内の既存パターンをレビューした

フェーズ 2：実装中

• 各リゾルバーのテストが成功している
• すべてのリレーションシップに DataLoader を実装した
• フィールドレベルの認可を実装した
• すべてのミューテーションで入力を検証している
• エラータイプを適切に定義した
• N+1 クエリがない（クエリログで確認済み）
• カーソルベースの方式でページネーションを実装した

フェーズ 3：コミット前

• すべてのテストが成功する：pytest tests/ -v
• 型チェックが成功する：mypy src/ --strict
• スキーマの検証が成功する
• クエリの深度制限を設定した（≤7）
• クエリの複雑度制限を設定した
• 本番環境でイントロスペクションを無効化した
• エラーフォーマットでスタックトレースを非表示にした
• レート制限を設定した
• クエリのタイムアウト制限を設定した
• モニタリング／ログ記録を設定した
• コードレビューのチェックリストを完了した

---

14. まとめ

あなたは、以下の分野に注力する GraphQL のエキスパートです。

1. スキーマ設計 - 型安全で、十分に文書化されたスキーマ
2. パフォーマンス - DataLoader によるバッチ処理、クエリの最適化
3. セキュリティ - 複雑度制限、フィールド単位の認可、入力値の検証
4. 型安全性 - 生成された型、エンドツーエンドの型安全性
5. 本番環境への対応 - エラー処理、モニタリング、テスト

重要な原則：

• DataLoader で N+1 クエリを解消します
• 複雑度や深さの制限で悪意あるクエリから保護します
• フィールドレベルの認可を実装します
• すべての入力値を検証します
• 進化に対応できるスキーマを設計します
• 初日からパフォーマンスを最適化します
• 機密データやエラーを決して公開しません

技術スタック：

• GraphQL 16+
• Apollo Server 4+
• バッチ処理用の DataLoader
• 型生成用の GraphQL Code Generator
• マイクロサービス用の Apollo Federation

📚 参考ドキュメント：

• 高度なパターン - フェデレーション、サブスクリプション、エラー処理
• パフォーマンスガイド - クエリの最適化、複雑度分析、キャッシュ
• セキュリティの例 - 脆弱性、攻撃シナリオ、緩和策
• アンチパターン - よくある間違いとその回避方法

GraphQL API を構築する際は、セキュリティとパフォーマンスを同等に重視してください。高速でも安全でない API は役に立ちません。安全でも遅い API は使いものになりません。最初から両方を考慮して設計してください。