本文目录导读：
我来详细介绍Python中构造GraphQL查询的几种方法：
基础方法：使用纯字符串
最简单的直接拼接查询字符串：
import requests
import json
# 简单查询
query = """
{
user(id: "1") {
name
email
age
}
}
"""
url = "https://api.example.com/graphql"
response = requests.post(url, json={'query': query})
print(response.json())
带变量的查询
import requests
query = """
query GetUser($id: ID!, $includeEmail: Boolean) {
user(id: $id) {
name
email @include(if: $includeEmail)
age
}
}
"""
variables = {
"id": "1",
"includeEmail": True
}
url = "https://api.example.com/graphql"
response = requests.post(
url,
json={
'query': query,
'variables': variables
}
)
print(response.json())
使用Python GraphQL客户端库
安装库
pip install gql requests
使用GQL库
from gql import gql, Client
from gql.transport.requests import RequestsHTTPTransport
# 配置传输
transport = RequestsHTTPTransport(
url="https://api.example.com/graphql",
headers={"Authorization": "Bearer YOUR_TOKEN"}
)
# 创建客户端
client = Client(transport=transport, fetch_schema_from_transport=True)
# 定义查询
query = gql("""
query GetUser($id: ID!) {
user(id: $id) {
name
email
posts {
title
content
}
}
}
""")
# 执行查询
variables = {"id": "1"}
result = client.execute(query, variable_values=variables)
print(result)
使用GraphQL-Core库
from graphql_core import graphql_sync
# 定义schema和resolver
schema = '''
type Query {
hello: String
user(id: ID!): User
}
type User {
id: ID!
name: String
email: String
}
'''
# 自定义查询
query = '''
query {
user(id: "1") {
name
email
}
}
'''
# 执行查询（本地模式）
result = graphql_sync(schema, query)
print(result.data)
使用AIOHTTP异步查询
import aiohttp
import asyncio
async def graphql_query(url, query, variables=None):
headers = {"Content-Type": "application/json"}
payload = {"query": query}
if variables:
payload["variables"] = variables
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload, headers=headers) as response:
return await response.json()
# 使用示例
async def main():
query = """
query GetUser($id: ID!) {
user(id: $id) {
name
email
posts {
title
}
}
}
"""
url = "https://api.example.com/graphql"
variables = {"id": "1"}
result = await graphql_query(url, query, variables)
print(result)
# 运行异步函数
asyncio.run(main())
Mutation操作示例
import requests
# 带参数的mutation
mutation = """
mutation CreateUser($input: CreateUserInput!) {
createUser(input: $input) {
user {
id
name
email
}
errors {
field
message
}
}
}
"""
variables = {
"input": {
"name": "张三",
"email": "zhangsan@example.com",
"password": "securepassword123"
}
}
response = requests.post(
"https://api.example.com/graphql",
json={"query": mutation, "variables": variables}
)
data = response.json()
if "errors" in data:
print("Error:", data["errors"])
else:
print("Created user:", data["data"]["createUser"]["user"])
封装成通用函数
import requests
from typing import Optional, Dict, Any
class GraphQLClient:
def __init__(self, endpoint: str, headers: Optional[Dict] = None):
self.endpoint = endpoint
self.headers = headers or {}
self.headers.setdefault("Content-Type", "application/json")
def execute(self, query: str, variables: Optional[Dict] = None) -> Dict[str, Any]:
"""执行GraphQL查询"""
payload = {"query": query}
if variables:
payload["variables"] = variables
response = requests.post(
self.endpoint,
json=payload,
headers=self.headers
)
if response.status_code != 200:
raise Exception(f"GraphQL request failed: {response.status_code}")
return response.json()
def execute_query(self, query: str, variables: Optional[Dict] = None) -> Dict:
"""执行查询操作"""
result = self.execute(query, variables)
if "errors" in result:
raise Exception(f"GraphQL errors: {result['errors']}")
return result.get("data", {})
# 使用示例
client = GraphQLClient(
endpoint="https://api.example.com/graphql",
headers={"Authorization": "Bearer YOUR_TOKEN"}
)
query = """
query GetUsers {
users {
id
name
email
}
}
"""
try:
data = client.execute_query(query)
print(data["users"])
except Exception as e:
print(f"Error: {e}")
动态构建查询
from typing import List, Dict
def build_graphql_query(
operation_type: str = "query",
operation_name: str = "",
fields: List[str] = None,
nested_fields: Dict[str, List[str]] = None,
arguments: Dict[str, str] = None
) -> str:
"""动态构建GraphQL查询字符串"""
# 构建字段选择
def build_selection(fields, nested=None):
selections = []
if fields:
selections.extend(fields)
if nested:
for name, subfields in nested.items():
sub_selection = build_selection(subfields)
selections.append(f"{name} {sub_selection}")
return "{\n" + "\n".join(f" {s}" for s in selections) + "\n}"
# 构建参数
args_str = ""
if arguments:
args = ", ".join(f"{k}: {v}" for k, v in arguments.items())
args_str = f"({args})"
# 构建完整查询
selection = build_selection(fields, nested_fields)
query = f"{operation_type} {operation_name}{args_str} {selection}"
return query
# 使用示例
query = build_graphql_query(
operation_type="query",
operation_name="GetUser",
fields=["name", "email"],
nested_fields={
"posts": ["title", "content"],
"profile": ["bio", "avatar"]
},
arguments={"id": "\"1\""}
)
print(query)
# 输出: query GetUser(id: "1") {
# name
# email
# posts {
# title
# content
# }
# profile {
# bio
# avatar
# }
# }
关键注意事项
- 变量名：使用
$variableName格式 - 参数类型：确保类型匹配（ID, String, Int, Boolean等）
- 错误处理：检查返回的
errors字段 - 认证：根据需要添加认证头
- 查询复杂度：避免嵌套过深的查询
选择哪种方法取决于你的具体需求：
- 简单场景：使用requests直接发送字符串
- 复杂应用：使用gql等专业库
- 性能要求：考虑异步方案
