如何使用gql通过Apollo WebSocket协议发送指定消息？

针对你需要发送的start类型消息，有两种实现方式，可根据需求选择：

方式一：使用gql高层API（推荐，符合Apollo协议规范） #

gql库会自动处理Apollo WebSocket协议的消息格式，你只需定义对应的GraphQL查询并传递参数：

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

logging.basicConfig(level=logging.INFO)

# 定义GraphQL查询，operationName需与目标消息一致
QUERY = gql("""
# 替换为实际操作名称和查询内容
subscription YourTargetOperationName {
  # 示例：根据eventId订阅数据，可按需修改
  targetSubscription(eventId: 488) {
    id
    content
  }
}
""")

async def main():
    # 初始化WebSocket传输，包含握手所需的init_payload
    transport = WebsocketsTransport(
        url="你的WebSocket地址",
        init_payload={
            'accept-language': 'en',
            'ec-version': '5.1.88',
            'referrer': "你的referrer地址"  # 按需添加
        }
    )
    
    async with Client(transport=transport, fetch_schema_from_transport=False) as session:
        # 执行订阅操作，gql自动生成符合格式的start消息
        async for result in session.subscribe(QUERY):
            # 处理服务器返回结果
            logging.info(f"收到返回结果: {result}")

asyncio.run(main())

如果需要动态传递eventId等参数，可使用变量方式：

QUERY = gql("""
subscription YourTargetOperationName($eventId: Int!) {
  targetSubscription(eventId: $eventId) {
    id
    content
  }
}
""")

# 在subscribe时传入变量
async for result in session.subscribe(QUERY, variable_values={"eventId": 488}):
    logging.info(f"收到返回结果: {result}")

这种方式下，gql库自动生成的消息会包含id、type: "start"、operationName及参数内容，完全符合Apollo协议规范。

方式二：直接发送原始自定义消息 #

如果服务器要求的消息格式特殊（比如eventId作为顶层字段），可直接通过transport发送原始JSON消息：

import asyncio
import logging
from gql.transport.websockets import WebsocketsTransport

logging.basicConfig(level=logging.INFO)

async def main():
    transport = WebsocketsTransport(
        url="你的WebSocket地址",
        init_payload={
            'accept-language': 'en',
            'ec-version': '5.1.88',
            'referrer': "你的referrer地址"
        }
    )
    
    # 手动建立连接
    await transport.connect()
    
    try:
        # 构造需要发送的消息
        send_message = {
            "id": "1073897396",
            "type": "start",
            "operationName": "你的操作名称",
            "eventId": 488
            # 若服务器需要查询内容，添加"query": "对应的GraphQL查询语句"
        }
        
        # 发送消息
        await transport.send(send_message)
        
        # 循环接收服务器返回的消息
        while True:
            response = await transport.receive()
            logging.info(f"收到消息: {response}")
    finally:
        # 关闭连接
        await transport.close()

asyncio.run(main())

注意事项 #

• 替换代码中的你的WebSocket地址、你的referrer地址、你的操作名称为实际值
• 若服务器要求query字段（通常GraphQL WebSocket需要），需在消息中添加对应的查询语句

内容的提问来源于stack exchange，提问作者HJA24