Python GraphQLクライアント「gql」完全ガイド 🚀

はじめに：GraphQLとgqlライブラリ

近年、API開発の世界ではRESTに代わる選択肢としてGraphQLが注目を集めています。GraphQLは、クライアントが必要なデータを正確に指定して取得できるクエリ言語であり、APIのためのランタイムです。これにより、データの過不足（オーバーフェッチやアンダーフェッチ）を防ぎ、効率的なデータ通信を実現します。

PythonでGraphQL APIを利用する際に非常に役立つのがgqlライブラリです。`gql`はPython製のGraphQLクライアントで、クエリ、ミューテーション（データの作成・更新・削除）、サブスクリプション（リアルタイム更新）といったGraphQLの主要な操作をサポートしています。同期・非同期の両方の使い方に対応しており、様々な通信プロトコル（HTTP, WebSocketなど）を利用できる柔軟性も持ち合わせています。

このブログ記事では、`gql`ライブラリの基本的な使い方から、非同期処理、サブスクリプションといった応用的な機能まで、詳細に解説していきます。さあ、`gql`を使ってPythonアプリケーションとGraphQL APIを連携させましょう！✨

インストール方法 💻

`gql`ライブラリのインストールはpipを使って簡単に行えます。基本的な機能だけを使いたい場合や、特定の通信方式（Transport）だけを使いたい場合、あるいは全ての機能を使いたい場合に応じて、インストール方法を選択できます。

基本的なインストール

最小限の依存関係で`gql`コアライブラリをインストールします。

pip install gql

特定のTransportと共にインストール

`gql`は様々な通信プロトコルをサポートしており、それぞれに対応するライブラリが必要です。例えば、非同期HTTP通信にaiohttpを使用する場合は以下のようにインストールします。

pip install gql[aiohttp]

同期HTTP通信にrequestsを使用する場合：

pip install gql[requests]

WebSocket（サブスクリプション用）を使用する場合：

pip install gql[websockets]

全ての依存関係を含めてインストール

`gql`がサポートする全てのTransportやオプション機能を使いたい場合は、allを指定してインストールします。

# zshなどの特定のシェルではクォートが必要な場合があります
pip install "gql[all]"

コアコンセプト：gqlの構成要素

`gql`ライブラリを効果的に使うために、いくつかの重要なコンポーネントとその役割を理解しましょう。

1. Transport (トランスポート)

Transportは、GraphQLサーバーとの実際の通信方法を定義します。`gql`は様々なTransportを提供しており、ユースケースに応じて選択できます。

• AIOHTTPTransport: aiohttpライブラリを使用した非同期HTTP通信。
• RequestsHTTPTransport: requestsライブラリを使用した同期HTTP通信。
• WebsocketsTransport: WebSocketプロトコルを使用した通信。主にサブスクリプションに使用されますが、クエリやミューテーションも実行可能です。Apolloプロトコルとgraphql-wsプロトコルの両方をサポートします。
• AppSyncWebsocketsTransport: AWS AppSyncのリアルタイムプロトコルに対応したWebSocket通信（実験的）。
• PhoenixChannelTransport: Phoenix Channelsプロトコルに対応した通信（実験的）。

Transportを選択する際には、APIサーバーがサポートするプロトコルと、アプリケーションが同期処理か非同期処理（asyncio）かを確認する必要があります。

2. Client (クライアント)

Clientは`gql`の中心的なクラスです。Transportを指定してインスタンスを作成し、GraphQLリクエスト（クエリ、ミューテーション、サブスクリプション）の実行を担当します。

from gql import Client
from gql.transport.aiohttp import AIOHTTPTransport

# AIOHTTPTransportを使用してクライアントを作成
transport = AIOHTTPTransport(url="https://your-graphql-endpoint.com/graphql")
client = Client(transport=transport, fetch_schema_from_transport=True)

fetch_schema_from_transport=Trueを指定すると、クライアントは初期化時にサーバーからGraphQLスキーマを取得しようとします。これにより、後述するクエリのローカル検証が可能になります。

3. gqlタグ

gql関数（タグのように使われることが多い）は、GraphQLクエリ文字列を解析し、内部的な表現（DocumentNode）に変換します。これにより、構文エラーを早期に検出したり、クエリの検証を行ったりできます。

from gql import gql

# GraphQLクエリ文字列をgqlタグでラップ
query = gql("""
  query GetGreeting {
    hello
  }
""")

4. スキーマ (Schema)

GraphQLスキーマは、APIが提供するデータ型、クエリ、ミューテーション、サブスクリプションを定義します。`gql`クライアントは、スキーマ情報を利用して以下のことを行えます。

• クエリ検証: 実行前にクエリがスキーマに対して有効かどうかをチェックします。
• カスタムスカラー/Enumの処理: スキーマ定義に基づいて、カスタムデータ型を適切にシリアライズ/デシリアライズします。

スキーマは、Transport経由でサーバーから取得（Introspection Queryを使用）するか、ローカルの.graphqlファイルや文字列から読み込むことができます。

5. 同期 vs 非同期 (Sync vs Async)

`gql` バージョン3以降、asyncioを使用した非同期処理をネイティブにサポートしています。

• 同期 (Sync): RequestsHTTPTransportを使用する場合、または非同期Transportをasyncioイベントループ外で直接client.execute()で実行する場合。リクエストが完了するまで処理がブロックされます。
• 非同期 (Async): AIOHTTPTransportやWebsocketsTransportをasync def関数内でawait client.execute_async()やasync with client as session: await session.execute()のように使用する場合。複数のリクエストを並行して実行できます。

基本的な使い方：クエリとミューテーション

ここでは、`gql`を使った基本的なGraphQLのクエリ（データ取得）とミューテーション（データ変更）の実行方法を見ていきましょう。例として、公開されている国情報API (https://countries.trevorblades.com/) を使用します。

同期的なクエリ実行 (RequestsTransport)

まずは、同期的なHTTP通信を行うRequestsHTTPTransportを使った例です。

from gql import gql, Client
from gql.transport.requests import RequestsHTTPTransport

# 同期HTTPトランスポートを設定
transport = RequestsHTTPTransport(
    url="https://countries.trevorblades.com/",
    use_json=True,  # POSTリクエストボディをJSONとして送信
    headers={"Content-type": "application/json"},
    verify=True,    # SSL証明書の検証を行う
    retries=3,      # リトライ回数
)

# クライアントを作成
# fetch_schema_from_transport=True でスキーマを取得し、クエリ検証を有効にする
client = Client(transport=transport, fetch_schema_from_transport=True)

# 実行したいGraphQLクエリを定義
query = gql(
    """
    query GetContinentInfo($code: ID!) {
      continent(code: $code) {
        name
        countries {
          code
          name
          capital
        }
      }
    }
    """
)

# クエリ実行時に渡す変数を定義
params = {"code": "AS"} # アジア大陸のコード

try:
    # クエリを実行
    result = client.execute(query, variable_values=params)
    print(result)
except Exception as e:
    print(f"エラーが発生しました: {e}")

このコードは、アジア大陸（コード: AS）の名前と、その大陸に含まれる国々のコード、名前、首都を取得します。client.execute()メソッドがリクエストを送信し、結果が辞書形式で返されます。variable_values引数でクエリに変数を渡すことができます。

非同期的なクエリ実行 (AIOHTTPTransport)

次に、asyncioとAIOHTTPTransportを使った非同期処理の例です。

import asyncio
from gql import gql, Client
from gql.transport.aiohttp import AIOHTTPTransport

async def main():
    # 非同期HTTPトランスポートを設定
    transport = AIOHTTPTransport(url="https://countries.trevorblades.com/")

    # クライアントを作成
    client = Client(transport=transport, fetch_schema_from_transport=True)

    # 実行したいGraphQLクエリを定義
    query = gql(
        """
        query GetCountry($code: ID!) {
          country(code: $code) {
            name
            native
            emoji
            currency
            languages {
              code
              name
            }
          }
        }
        """
    )

    # クエリ実行時に渡す変数を定義
    params = {"code": "JP"} # 日本の国コード

    # 非同期でクライアントセッションを開始
    async with client as session:
        try:
            # クエリを非同期で実行
            result = await session.execute(query, variable_values=params)
            print(result)
        except Exception as e:
            print(f"エラーが発生しました: {e}")

# asyncioイベントループで main() コルーチンを実行
if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass # Ctrl+Cでの終了を許可

この例では、日本の国情報（名前、ネイティブ名、絵文字、通貨、言語）を取得します。主な違いは以下の点です。

• コード全体がasync def main(): のように非同期関数内で定義されています。
• クライアントの使用はasync with client as session: ブロックで行われ、セッション管理が自動化されます。
• クエリの実行はawait session.execute(...) のようにawaitキーワードを使って行われます。
• スクリプトの実行はasyncio.run(main()) を使って開始されます。

ミューテーション (Mutation)

ミューテーションはデータの作成、更新、削除を行う操作です。構文はクエリと似ていますが、通常mutationキーワードで始まります。`gql`での実行方法はクエリと全く同じです。

# (TransportとClientの設定は上記クエリ例と同様)

# 例：架空のタスク追加ミューテーション
mutation = gql(
    """
    mutation AddTask($title: String!, $description: String) {
      addTask(title: $title, description: $description) {
        id
        title
        completed
      }
    }
    """
)

# ミューテーションに渡す変数
task_data = {"title": "gqlライブラリを学ぶ", "description": "ブログ記事を読む"}

# 同期実行の場合
# result = client.execute(mutation, variable_values=task_data)

# 非同期実行の場合
# async with client as session:
#     result = await session.execute(mutation, variable_values=task_data)

# print(result) # -> {'addTask': {'id': 'some-generated-id', 'title': 'gqlライブラリを学ぶ', 'completed': False}}

ミューテーションを実行すると、通常は変更後のデータが返されます（スキーマ定義によります）。エラーハンドリングもクエリと同様に行います。

応用的な使い方 🛠️

`gql`は基本的なクエリ・ミューテーション以外にも、多くの高度な機能を提供しています。

サブスクリプション (Subscriptions)

サブスクリプションは、サーバーからのリアルタイムなデータ更新を受け取るための仕組みです。通常、WebSocketプロトコル上で動作します。`gql`ではWebsocketsTransportを使用してサブスクリプションを実行できます。

import asyncio
from gql import gql, Client
from gql.transport.websockets import WebsocketsTransport

async def subscribe_to_counter():
    # WebSocketトランスポートを設定 (URLはサーバーに合わせて変更)
    transport = WebsocketsTransport(url='ws://your-graphql-ws-endpoint/graphql')

    # クライアントを作成
    client = Client(transport=transport, fetch_schema_from_transport=True)

    # サブスクリプションクエリを定義 (例: 1秒ごとにカウントアップするカウンター)
    subscription_query = gql('''
      subscription CounterSubscription {
        counter
      }
    ''')

    print("サブスクリプションを開始します...")
    try:
        # 非同期イテレータとしてサブスクリプションを実行
        async for result in client.subscribe(subscription_query):
            print(f"サーバーからの更新: {result}")
            # ここで受け取ったデータに応じた処理を行う
            # 例: result['counter'] の値を利用する

    except asyncio.CancelledError:
        print("サブスクリプションがキャンセルされました。")
    except Exception as e:
        print(f"サブスクリプション中にエラーが発生しました: {e}")
    finally:
        print("サブスクリプションを終了します。")
        # 必要であればクリーンアップ処理

if __name__ == "__main__":
    try:
        asyncio.run(subscribe_to_counter())
    except KeyboardInterrupt:
        print("\n終了します...")

この例では、client.subscribe()メソッド（非同期版）を使用しています。async forループを使って、サーバーから新しいデータが送られてくるたびにresult変数で受け取ります。同期的にサブスクリプションを実行する場合は、client.subscribe()（同期版）を通常のforループで使用します。

ファイルアップロード

GraphQLの仕様にはファイルアップロードの標準的な方法はありませんが、一般的にGraphQL multipart request specificationが使われます。`gql`はこの仕様に準拠したファイルアップロードをサポートしています。

アップロードを行うには、ミューテーションの変数としてファイルオブジェクトを渡します。`gql`は自動的にリクエストをmultipart/form-data形式にエンコードします。

# (TransportとClientの設定は上記クエリ例と同様)
# AIOHTTPTransport または RequestsHTTPTransport を使用

mutation_upload = gql(
    """
    mutation UploadFile($file: Upload!) {
      uploadProfilePicture(file: $file) {
        success
        imageUrl
      }
    }
    """
)

# アップロードするファイルを開く (バイナリモード 'rb' で)
try:
    with open("path/to/your/image.jpg", "rb") as f:
        # 変数としてファイルオブジェクトを渡す
        params = {"file": f}

        # 同期実行の場合
        # result = client.execute(mutation_upload, variable_values=params, upload_files=True)

        # 非同期実行の場合
        # async with client as session:
        #     result = await session.execute(mutation_upload, variable_values=params, upload_files=True)

        # print(result)

except FileNotFoundError:
    print("エラー: 指定されたファイルが見つかりません。")
except Exception as e:
    print(f"ファイルアップロード中にエラーが発生しました: {e}")

# 注意: client.execute や session.execute を呼び出す際に upload_files=True を指定する必要があります。

重要なのは、executeメソッドを呼び出す際にupload_files=Trueフラグを設定することです。これにより、`gql`はファイルを含む変数を適切に処理します。

スキーマ検証 (Schema Validation)

クライアント初期化時にfetch_schema_from_transport=Trueを指定するか、schema引数にスキーマ文字列やファイルパスを指定することで、クエリ実行前にローカルで検証を行うことができます。

from gql import Client, gql
from gql.transport.requests import RequestsHTTPTransport

# スキーマをローカルファイルから読み込む場合
# with open("schema.graphql", "r") as f:
#     schema_str = f.read()
# client = Client(transport=transport, schema=schema_str)

# または、Transportから取得する場合
transport = RequestsHTTPTransport(url="https://countries.trevorblades.com/")
client = Client(transport=transport, fetch_schema_from_transport=True)

# 無効なフィールドを含むクエリ (スキーマに存在しない 'invalidField')
invalid_query = gql("""
    query {
        continent(code: "AS") {
            name
            invalidField # このフィールドはスキーマに存在しない
        }
    }
""")

try:
    # スキーマ検証が有効なため、ここでエラーが発生する
    result = client.execute(invalid_query)
    print(result)
except Exception as e:
    # 例: graphql.error.located_error.GraphQLLocatedError: Cannot query field 'invalidField' on type 'Continent'.
    print(f"クエリ検証エラー: {e}")

これにより、存在しないフィールドをクエリしたり、型が間違っていたりする問題を、実際にサーバーにリクエストを送る前に検出でき、開発効率が向上します。🚀

動的なクエリ構築 (DSL)

`gql`には、Pythonコードを使って動的にGraphQLクエリを構築するためのDSL（Domain Specific Language）モジュールも含まれています。これは、ユーザーの入力や条件に応じてクエリの内容を変更したい場合に便利です。

from gql.dsl import DSLSchema, DSLQuery, dsl_gql, DSLType, DSLField, DSLVariableDefinitions, DSLArgument

# スキーマ情報が必要 (例として手動で定義)
# 実際には introspection クエリ結果などから生成することが多い
class Country(DSLType):
    code = DSLField()
    name = DSLField()
    capital = DSLField()

class Query(DSLType):
    country = DSLField(
        args=DSLArgument(name="code", type="ID!"),
        type=Country
    )

class MySchema(DSLSchema):
    query: Query

# スキーマインスタンスを作成
ds = MySchema()

# 動的にフィールドを選択
selected_fields = [ds.Country.name]
if True: # 条件によってフィールドを追加
    selected_fields.append(ds.Country.capital)

# 変数定義
vars = DSLVariableDefinitions()
vars.code = DSLVariable(type="ID!")

# DSLを使ってクエリを構築
dsl_query = dsl_gql(
    DSLQuery(
        ds.Query.country(code=vars.code).select(
            *selected_fields
        )
    ).variable_definitions(vars)
)

# 構築されたクエリを表示 (DocumentNode オブジェクトが出力される)
# print(dsl_query)

# Client を使って実行 (変数も渡す)
# client = Client(...) # Client の初期化は省略
# result = client.execute(dsl_query, variable_values={"code": "FR"})
# print(result)

DSLを使うと、文字列操作に頼らずに型安全な方法でクエリを組み立てられますが、学習コストはやや高めです。シンプルなケースでは通常の文字列クエリで十分なことが多いでしょう。

Python GraphQLクライアントの選択肢 🤔

`gql`は非常に高機能で人気のあるPython GraphQLクライアントですが、他にもいくつかの選択肢が存在します。プロジェクトの要件によっては、他のライブラリが適している場合もあります。

選択のポイント：

• 機能性重視なら: `gql` が最も多機能です。サブスクリプションや複雑なユースケースが必要な場合は有力候補です。
• シンプルさ重視なら: `python-graphql-client` は学習コストが低く、手軽に始められます。
• 型安全性・データクラス連携重視なら: `gqlclient` や `sgqlc` が適しています。特に `sgqlc` はコード生成により開発体験が向上する可能性があります。

プロジェクトの規模、必要な機能、チームの好みなどを考慮して最適なライブラリを選択してください。

まとめ 🏁

`gql`は、PythonでGraphQL APIを扱うための強力で柔軟なクライアントライブラリです。この記事では、その基本的な使い方から、非同期処理、サブスクリプション、ファイルアップロード、スキーマ検証といった応用的な機能までを解説しました。

GraphQLはその効率性と柔軟性から、多くのモダンなアプリケーションで採用が進んでいます。`gql`ライブラリを使いこなすことで、PythonアプリケーションからこれらのAPIをスムーズかつ効果的に利用できるようになります。

ぜひ、あなたの次のプロジェクトで`gql`を活用してみてください！ Happy coding! 🎉