GraphQLの基本とPythonでの実装

今日のWebアプリケーション開発において、効率的で柔軟なAPIは不可欠です。GraphQLは、クライアントが必要とするデータを正確に要求できるAPIのクエリ言語であり、実行環境です。これにより、RESTful APIが抱えるいくつかの課題を解決し、より強力で進化しやすいAPI設計を可能にします。

GraphQLとRESTの比較

RESTful APIでは、通常、リソースごとに異なるエンドポイントが存在し、クライアントはしばしば必要以上のデータ（オーバーフェッチ）や、複数のエンドポイントへのリクエスト（アンダーフェッチや複数リクエスト）を強いられます。一方、GraphQLは以下の特徴により、これらの問題を改善します。

• 単一エンドポイント: 全てのデータリクエストは単一のGraphQLエンドポイントに対して行われます。
• 要求に応じたデータ取得: クライアントは必要なフィールドのみを指定してデータを取得できます。これにより、ネットワーク帯域の節約とアプリケーションのパフォーマンス向上が期待できます。
• 厳格な型システム: GraphQLはスキーマ定義言語（SDL）によってAPIの構造を厳密に定義します。これにより、クライアントとサーバー間の契約が明確になり、開発時のエラーを減らします。
• 階層的なクエリ: 関連するデータをネストされた構造で一度に取得できるため、N+1問題のようなオーバーヘッドを回避できます。
• 進化しやすいAPI: スキーマの変更が既存のクライアントに与える影響を最小限に抑えながら、APIを継続的に発展させることができます。

本記事では、PythonでGraphQLサーバーを構築するためのフレームワークであるGrapheneと、WebフレームワークのFlask、およびORMのSQLAlchemyを組み合わせて、実践的なGraphQL APIの実装方法を紹介します。

Grapheneを使用したGraphQL APIの試行

まずは、Grapheneを使った基本的なGraphQL APIの例を見てみましょう。ここでは、メモリ上のデータとデータベース（MySQL）からのデータを両方扱うシンプルなスキーマを定義します。

以下のコードでは、Teacher（教師）とStudent（生徒）という2つのエンティティを扱い、それぞれに対するクエリを定義しています。データベースにはteachersとstudentsというテーブルを使用します。

データベースとGrapheneのスキーマ定義

import json
import graphene
from graphene import relay, ObjectType, Schema
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField
from flask import Flask
from flask_graphql import GraphQLView

# データベース接続設定
DATABASE_URL = 'mysql+pymysql://root:123456@127.0.0.1:3306/test_db' # データベース名を'test'から'test_db'に変更
engine = create_engine(DATABASE_URL)
db_session = scoped_session(sessionmaker(autoflush=False, bind=engine))

Base = declarative_base()
Base.query = db_session.query_property()

# JSON出力ユーティリティ
def format_json_output(data):
    return json.dumps(data, sort_keys=True, indent=4, separators=(',', ':'), ensure_ascii=False)

# SQLAlchemyモデル定義
class TeacherModel(Base):
    __tablename__ = 'teachers' # テーブル名を'laoshi'から'teachers'に変更
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    age = Column(Integer)

class StudentModel(Base):
    __tablename__ = 'students' # テーブル名を'xuesheng'から'students'に変更
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    grade = Column(Integer) # 'age'から'grade'に変更

# Grapheneオブジェクトタイプ定義 (SQLAlchemyモデルベース)
class TeacherType(SQLAlchemyObjectType):
    class Meta:
        model = TeacherModel
        interfaces = (graphene.relay.Node,)

class StudentType(SQLAlchemyObjectType):
    class Meta:
        model = StudentModel
        interfaces = (graphene.relay.Node,)

# インメモリデータの準備（初期テスト用）
mock_teachers = [
    TeacherType(id=1, name="田中先生", age=45),
    TeacherType(id=2, name="鈴木先生", age=38),
    TeacherType(id=3, name="佐藤先生", age=52),
]

mock_students = [
    StudentType(id=1, name="山田太郎", grade=10),
    StudentType(id=2, name="小林花子", grade=11),
    StudentType(id=3, name="井上健太", grade=10),
    StudentType(id=4, name="渡辺美咲", grade=12),
]

# ルートクエリ定義
class Query(graphene.ObjectType):
    node = graphene.relay.Node.Field()

    # 単一教師をIDで取得
    teacher_by_id = graphene.Field(TeacherType, teacher_id=graphene.ID(required=True))

    # 生徒を名前で取得
    student_by_name = graphene.Field(StudentType, student_name=graphene.String())

    # 特定の学年の生徒を取得
    students_by_grade = graphene.List(StudentType, student_grade=graphene.Int())

    # 複数の学年の生徒を取得
    students_in_grades = graphene.List(StudentType, grades_str=graphene.String())

    # データベースからの全教師を取得 (Relay Connection)
    all_teachers_from_db = SQLAlchemyConnectionField(TeacherType.connection)

    # データベースから検索文字列を含む教師を取得
    find_teachers_from_db = graphene.List(TeacherType, search_term=graphene.String())

    # リゾルバの実装
    def resolve_teacher_by_id(self, info, teacher_id):
        # インメモリデータから検索
        for teacher in mock_teachers:
            if teacher.id == int(teacher_id):
                return teacher
        return None

    def resolve_student_by_name(self, info, student_name):
        # インメモリデータから検索
        for student in mock_students:
            if student.name == student_name:
                return student
        return None

    def resolve_students_by_grade(self, info, student_grade):
        # インメモリデータから検索
        result = []
        for student in mock_students:
            if student.grade == student_grade:
                result.append(student)
        return result

    def resolve_students_in_grades(self, info, grades_str):
        # インメモリデータから検索
        target_grades = [int(g) for g in grades_str.split(',')]
        result = []
        for student in mock_students:
            if student.grade in target_grades:
                result.append(student)
        return result

    def resolve_find_teachers_from_db(self, info, search_term=None):
        query = TeacherType.get_query(info)
        if search_term:
            query = query.filter(TeacherModel.name.ilike(f'%{search_term}%')) # 大文字小文字を区別しない検索
        return query.all()

# スキーマの生成
schema = graphene.Schema(query=Query)

# データベースの初期化 (初回実行時のみコメントを外す)
# Base.metadata.create_all(engine)
# db_session.add_all([
#     TeacherModel(name="教員A", age=30),
#     TeacherModel(name="教員B", age=35),
#     TeacherModel(name="教員C", age=40),
#     StudentModel(name="生徒X", grade=10),
#     StudentModel(name="生徒Y", grade=11),
#     StudentModel(name="生徒Z", grade=10),
# ])
# db_session.commit()

if __name__ == '__main__':
    print("--- IDによる教師情報の取得 ---")
    query_teacher_by_id = '''
    query {
      teacherById(teacherId: 1) {
        id, name, age
      }
    }
    '''
    result = schema.execute(query_teacher_by_id)
    print(format_json_output(result.data))

    print("\n--- 名前による生徒情報の取得 ---")
    query_student_by_name = '''
    query {
      studentByName(studentName: "山田太郎") {
        id, name, grade
      }
    }
    '''
    result = schema.execute(query_student_by_name)
    print(format_json_output(result.data))

    print("\n--- 特定の学年の生徒情報の取得 ---")
    query_students_by_grade = '''
    query {
      studentsByGrade(studentGrade: 10) {
        id, name
      }
    }
    '''
    result = schema.execute(query_students_by_grade)
    print(format_json_output(result.data))

    print("\n--- 複数の学年の生徒情報の取得 ---")
    query_students_in_grades = '''
    query {
      studentsInGrades(gradesStr: "11,12") {
        id, name, grade
      }
    }
    '''
    result = schema.execute(query_students_in_grades)
    print(format_json_output(result.data))

    print("\n--- データベースからの全教師情報取得 (Relay形式) ---")
    query_all_teachers_db = '''
    query {
      allTeachersFromDb {
        edges {
          node {
            id, name, age
          }
        }
      }
    }
    '''
    result = schema.execute(query_all_teachers_db)
    print(format_json_output(result.data))

    print("\n--- データベースから検索文字列を含む教師情報取得 (全件) ---")
    query_find_teachers_all = '''
        query {
          findTeachersFromDb(searchTerm: "") {
            id, name, age
          }
        }
    '''
    result = schema.execute(query_find_teachers_all)
    print(format_json_output(result.data))

    print("\n--- データベースから検索文字列を含む教師情報取得 ('B'を含む) ---")
    query_find_teachers_b = '''
        query {
          findTeachersFromDb(searchTerm: "B") {
            id, name, age
          }
        }
    '''
    result = schema.execute(query_find_teachers_b)
    print(format_json_output(result.data))

コード実行結果の例

上記のコードを実行すると、定義されたGraphQLクエリに基づいて以下のようなJSON形式のデータが出力されます。これは、Grapheneがスキーマ定義とリゾルバに基づいてデータを取得・整形していることを示しています。

--- IDによる教師情報の取得 ---
{
    "teacherById": {
        "age": 45,
        "id": "1",
        "name": "田中先生"
    }
}

--- 名前による生徒情報の取得 ---
{
    "studentByName": {
        "grade": 10,
        "id": "1",
        "name": "山田太郎"
    }
}

--- 特定の学年の生徒情報の取得 ---
{
    "studentsByGrade": [
        {
            "id": "1",
            "name": "山田太郎"
        },
        {
            "id": "3",
            "name": "井上健太"
        }
    ]
}

--- 複数の学年の生徒情報の取得 ---
{
    "studentsInGrades": [
        {
            "grade": 11,
            "id": "2",
            "name": "小林花子"
        },
        {
            "grade": 12,
            "id": "4",
            "name": "渡辺美咲"
        }
    ]
}

--- データベースからの全教師情報取得 (Relay形式) ---
{
    "allTeachersFromDb": {
        "edges": [
            {
                "node": {
                    "age": 30,
                    "id": "VGVhY2hlclR5cGU6MQ==",
                    "name": "教員A"
                }
            },
            {
                "node": {
                    "age": 35,
                    "id": "VGVhY2hlclR5cGU6Mg==",
                    "name": "教員B"
                }
            },
            {
                "node": {
                    "age": 40,
                    "id": "VGVhY2hlclR5cGU6Mw==",
                    "name": "教員C"
                }
            }
        ]
    }
}

--- データベースから検索文字列を含む教師情報取得 (全件) ---
{
    "findTeachersFromDb": [
        {
            "age": 30,
            "id": "VGVhY2hlclR5cGU6MQ==",
            "name": "教員A"
        },
        {
            "age": 35,
            "id": "VGVhY2hlclR5cGU6Mg==",
            "name": "教員B"
        },
        {
            "age": 40,
            "id": "VGVhY2hlclR5cGU6Mw==",
            "name": "教員C"
        }
    ]
}

--- データベースから検索文字列を含む教師情報取得 ('B'を含む) ---
{
    "findTeachersFromDb": [
        {
            "age": 35,
            "id": "VGVhY2hlclR5cGU6Mg==",
            "name": "教員B"
        }
    ]
}

FlaskでGraphQLサーバーを実装する

次に、上記で定義したGraphQLスキーマをWeb APIとして公開するため、軽量なWebフレームワークであるFlaskと統合します。flask-graphqlライブラリを使用すると、簡単にGraphQLエンドポイントをセットアップできます。

サーバーサイドの実装 (server.py)

このサーバーは、/graphqlパスでGraphQLクエリを受け付け、データベースから教師情報を検索して返します。

import json
import graphene
from graphene import relay, ObjectType, Schema
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField
from flask import Flask, request
from flask_graphql import GraphQLView

# データベース接続設定
DATABASE_URL = 'mysql+pymysql://root:123456@127.0.0.1:3306/test_db'
engine = create_engine(DATABASE_URL)
db_session = scoped_session(sessionmaker(autoflush=False, bind=engine))

Base = declarative_base()
Base.query = db_session.query_property()

# JSON出力ユーティリティ
def format_json_output(data):
    return json.dumps(data, sort_keys=True, indent=4, separators=(',', ':'), ensure_ascii=False)

# SQLAlchemyモデル定義
class TeacherModel(Base):
    __tablename__ = 'teachers'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    age = Column(Integer)

# Grapheneオブジェクトタイプ定義
class TeacherType(SQLAlchemyObjectType):
    class Meta:
        model = TeacherModel
        interfaces = (graphene.relay.Node,)

# ルートクエリ定義
class Query(graphene.ObjectType):
    node = graphene.relay.Node.Field()
    # 検索文字列に基づいて教師のリストを取得
    teachers = graphene.List(TeacherType, search_name=graphene.String())

    # リゾルバの実装
    def resolve_teachers(self, info, search_name=None):
        query = TeacherType.get_query(info)
        if search_name:
            query = query.filter(TeacherModel.name.ilike(f'%{search_name}%'))
        return query.all()

# GraphQLスキーマの生成
schema = graphene.Schema(query=Query)

app = Flask(__name__)

# GraphQLエンドポイントの追加
app.add_url_rule(
    '/graphql',
    view_func=GraphQLView.as_view(
        'graphql',
        schema=schema,
        graphiql=True # ブラウザからインタラクティブなIDE (GraphiQL) を利用可能にする
    )
)

if __name__ == '__main__':
    app.run(debug=True, port=5000)

クライアントサイドの実装 (client.py)

Pythonのrequestsライブラリを使用して、上記のFlask GraphQLサーバーに対してクエリを送信します。

import requests
import json
import copy

# ベースURL設定
BASE_URL = 'http://127.0.0.1:5000/graphql'

# JSON出力ユーティリティ
def format_json_output(data):
    return json.dumps(data, sort_keys=True, indent=4, separators=(',', ':'), ensure_ascii=False)

# GraphQLリクエスト送信関数
def send_graphql_query(query_string, operation_name="GraphQL Request"):
    headers = {'Content-Type': 'application/json'}
    payload = {'query': query_string}

    print(f"--- {operation_name} ---")
    print(f"リクエストURL: {BASE_URL}")
    print(f"送信データ: {format_json_output(payload)}")

    try:
        response = requests.post(BASE_URL, headers=headers, json=payload)
        response.raise_for_status() # HTTPエラーをチェック
        json_response = response.json()
        print(f"レスポンス: {format_json_output(json_response)}")
    except requests.exceptions.RequestException as e:
        print(f"リクエストエラー: {e}")
        json_response = {"error": str(e)}
    print("\n")
    return json_response

if __name__ == '__main__':
    # 1. パラメータなしで全教師情報を取得
    query_all_teachers = '''
        query {
          teachers {
            id, name, age
          }
        }
    '''
    send_graphql_query(query_all_teachers, '全教師情報を取得')

    # 2. 検索文字列「教員B」で教師情報を取得
    query_search_teacher_b = '''
        query {
          teachers(searchName: "教員B") {
            id, name, age
          }
        }
    '''
    send_graphql_query(query_search_teacher_b, '名前で教師情報を検索（「教員B」）')

    # 3. 検索文字列「教員」で教師情報を取得し、IDと名前のみを要求
    query_search_teacher_partial_name_fields = '''
        query {
          teachers(searchName: "教員") {
            id, name
          }
        }
    '''
    send_graphql_query(query_search_teacher_partial_name_fields, '名前で教師情報を検索（「教員」、IDと名前のみ）')

サーバーとクライアントの実行結果

サーバー（server.py）を起動し、次にクライアント（client.py）を実行すると、以下のような結果が得られます。クライアントはGraphQLクエリをサーバーに送信し、サーバーはそれに応じたデータを返します。

クライアントの実行出力例

--- 全教師情報を取得 ---
リクエストURL: http://127.0.0.1:5000/graphql
送信データ: {
    "query": "\n        query {\n          teachers {\n            id, name, age\n          }\n        }\n    "
}
レスポンス: {
    "data": {
        "teachers": [
            {
                "age": 30,
                "id": "VGVhY2hlclR5cGU6MQ==",
                "name": "教員A"
            },
            {
                "age": 35,
                "id": "VGVhY2hlclR5cGU6Mg==",
                "name": "教員B"
            },
            {
                "age": 40,
                "id": "VGVhY2hlclR5cGU6Mw==",
                "name": "教員C"
            }
        ]
    }
}


--- 名前で教師情報を検索（「教員B」） ---
リクエストURL: http://127.0.0.1:5000/graphql
送信データ: {
    "query": "\n        query {\n          teachers(searchName: \"教員B\") {\n            id, name, age\n          }\n        }\n    "
}
レスポンス: {
    "data": {
        "teachers": [
            {
                "age": 35,
                "id": "VGVhY2hlclR5cGU6Mg==",
                "name": "教員B"
            }
        ]
    }
}


--- 名前で教師情報を検索（「教員」、IDと名前のみ） ---
リクエストURL: http://127.0.0.1:5000/graphql
送信データ: {
    "query": "\n        query {\n          teachers(searchName: \"教員\") {\n            id, name\n          }\n        }\n    "
}
レスポンス: {
    "data": {
        "teachers": [
            {
                "id": "VGVhY2hlclR5cGU6MQ==",
                "name": "教員A"
            },
            {
                "id": "VGVhY2hlclR5cGU6Mg==",
                "name": "教員B"
            },
            {
                "id": "VGVhY2hlclR5cGU6Mw==",
                "name": "教員C"
            }
        ]
    }
}

この例から、GraphQLがクライアントの要求に応じて柔軟にデータを整形して返す能力が明確に示されています。特に最後のクエリでは、ageフィールドが要求されなかったため、レスポンスに含まれていないことが確認できます。