Python脚本GraphQL查询如何构造

本文目录导读：

1. 基础方法：使用纯字符串
2. 带变量的查询
3. 使用Python GraphQL客户端库
4. 使用GraphQL-Core库
5. 使用AIOHTTP异步查询
6. Mutation操作示例
7. 封装成通用函数
8. 动态构建查询
9. 关键注意事项

我来详细介绍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
#   }
# }

关键注意事项

1. 变量名：使用$variableName格式
2. 参数类型：确保类型匹配（ID, String, Int, Boolean等）
3. 错误处理：检查返回的errors字段
4. 认证：根据需要添加认证头
5. 查询复杂度：避免嵌套过深的查询

选择哪种方法取决于你的具体需求：

• 简单场景：使用requests直接发送字符串
• 复杂应用：使用gql等专业库
• 性能要求：考虑异步方案