使用Graphene Relay创建Subscription根类型是否可行？

嗨，很高兴你开始探索Graphene和GraphQL订阅！答案是完全可以用Graphene创建Subscription根类型，而且实现起来也不算复杂，下面我给你详细讲讲怎么操作：

Graphene从较早版本就支持Subscription功能了，核心是定义一个订阅根类型，然后在里面实现流式返回数据的解析逻辑。

基本实现步骤 #

• 定义一个继承自graphene.ObjectType的类作为订阅根
• 在订阅根中定义订阅字段，指定返回类型
• 为每个订阅字段编写异步解析方法（方法名以subscribe_开头，或者通过resolver参数指定），该方法需要返回AsyncIterable类型，用于持续推送数据
• 将订阅根类型关联到GraphQL Schema中

代码示例 #

假设我们要实现一个简单的实时消息订阅功能：

import graphene
import asyncio
from graphene import ObjectType, Schema

# 先定义一个消息类型，作为订阅返回的数据结构
class MessageType(ObjectType):
    content = graphene.String(description="消息内容")
    sender = graphene.String(description="发送者")

# 定义订阅根类型
class Subscription(ObjectType):
    # 定义订阅字段，返回MessageType
    new_message = graphene.Field(MessageType)

    # 订阅的异步解析方法，返回AsyncIterable用于流式推送
    async def subscribe_new_message(root, info):
        # 这里模拟实时生成消息的逻辑，实际项目中可能是从消息队列/WebSocket连接获取数据
        for i in range(5):
            await asyncio.sleep(1)
            yield MessageType(content=f"实时消息 {i}", sender="系统通知")

# 创建Schema，关联订阅根类型
schema = Schema(subscription=Subscription)

测试订阅功能 #

你可以在异步环境中执行以下代码验证订阅效果：

async def test_subscription():
    subscription_query = """
    subscription {
        newMessage {
            content
            sender
        }
    }
    """
    # 执行订阅，获取异步迭代器
    result = schema.subscribe(subscription_query)
    # 遍历接收推送的数据
    async for item in result:
        print(item.data)

# 运行测试
asyncio.run(test_subscription())

生产环境注意事项 #

• 订阅依赖持久化连接，生产环境中通常需要结合WebSocket服务器（比如FastAPI配合WebSocket、Django Channels）来处理客户端连接
• 如果使用Django框架，可以用graphene-django扩展，它对订阅的支持和原生Graphene一致，只需额外配置Django Channels来管理WebSocket连接
• 解析方法里的逻辑需要适配你的业务场景，比如从Redis队列、Kafka等消息中间件获取实时数据，再推送给客户端

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