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

pip install gql

pip install gql[aiohttp]

pip install gql[requests]

pip install gql[websockets]

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

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)

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

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}")

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での終了を許可

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

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終了します...")

# (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 を指定する必要があります。

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}")

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)