解决 aiohttp 中 HTTP 头部换行符错误的指南：深入理解与实践

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

# 假设你的GraphQL API端点
GRAPHQL_URL = "https://your.graphql.api/endpoint"

# 模拟从文件或环境变量加载API Key
# 假设 api_key_file.txt 内容为 "your_api_key_value
"
def load_api_key_from_source_problematic():
    """模拟从源加载API Key，可能包含未处理的换行符。"""
    # 实际场景可能是 f.read() 或 os.getenv()
    # 例如：with open("api_key.txt", "r") as f: return f.read()
    return "your_api_key_value_from_file
"

# --- 调试步骤：在传递给 aiohttp 之前检查头部值 ---
print("--- 调试：检查原始API Key ---")
raw_api_key = load_api_key_from_source_problematic()
print(f"从源加载的原始API Key: '{raw_api_key}'")
print(f"原始API Key的长度: {len(raw_api_key)}")
print(f"原始API Key是否包含换行符: {'\n' in raw_api_key or '\r' in raw_api_key}")

# 构造头部字典
# 假设认证头部是 'Authorization: Bearer <API_KEY>'
headers_to_send = {
    "Authorization": f"Bearer {raw_api_key}"
}

print(f"即将传递给 aiohttp 的 Authorization 头部值: '{headers_to_send['Authorization']}'")
print(f"头部值长度: {len(headers_to_send['Authorization'])}")
print("-" * 30)

# 尝试使用这个头部创建 gql 客户端 (此操作会触发 aiohttp 错误)
try:
    # 实际项目中，这里会是 transport = AIOHTTPTransport(url=GRAPHQL_URL, headers=headers_to_send)
    # client = Client(transport=transport)
    # client.execute(...)
    print("注意：如果上述头部包含换行符，此处将抛出 ValueError。")
    # 模拟 aiohttp 抛出错误
    if '
' in headers_to_send['Authorization'] or '
' in headers_to_send['Authorization']:
        raise ValueError("模拟: Newline or carriage return character detected in HTTP status message or header.")
except ValueError as e:
    print(f"成功捕获到预期错误: {e}")
except Exception as e:
    print(f"捕获到其他错误: {e}")

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

GRAPHQL_URL = "https://your.graphql.api/endpoint"

# 模拟从文件或环境变量加载API Key
def load_api_key_from_source_correct():
    """模拟从源加载API Key，并进行正确清理。"""
    # 假设从文件读取的内容是 "your_api_key_value_from_file
"
    raw_key = "your_api_key_value_from_file
"
    return raw_key.strip() # 使用 .strip() 移除空白字符

# --- 正确的使用方法 ---
print("
--- 解决方案：使用 .strip() 清理API Key ---")
cleaned_api_key = load_api_key_from_source_correct()
print(f"清理后的API Key: '{cleaned_api_key}'")
print(f"清理后API Key的长度: {len(cleaned_api_key)}")
print(f"清理后API Key是否包含换行符: {'\n' in cleaned_api_key or '\r' in cleaned_api_key}")

# 构造头部字典
headers_clean = {
    "Authorization": f"Bearer {cleaned_api_key}"
}

print(f"即将传递给 aiohttp 的干净 Authorization 头部值: '{headers_clean['Authorization']}'")
print(f"干净头部值长度: {len(headers_clean['Authorization'])}")
print("-" * 30)

# 使用清理后的头部创建 gql 客户端 (此操作将正常工作)
try:
    transport = AIOHTTPTransport(url=GRAPHQL_URL, headers=headers_clean)
    client = Client(transport=transport)

    # 假设一个简单的 GraphQL 查询
    query = gql("query { __typename }")
    # result = client.execute(query)
    # print("GraphQL 查询成功执行。")
    print("（此处代码在实际运行时将正常工作，不会触发 ValueError）")

except Exception as e:
    print(f"捕获到意外错误: {e}")