如何将Strawberry异步订阅方法精简为一行代码？

要实现像Query类那样用一行赋值定义订阅字段，核心是利用Strawberry的字段描述符机制——node()这类函数本质是返回一个Strawberry能识别的Field对象，在类被@strawberry.type处理时自动注册为GraphQL字段。我们可以封装一个自定义工厂函数，动态生成符合要求的订阅字段。

步骤1：封装订阅字段工厂函数 #

这个函数接收模型作为参数，动态生成对应订阅的解析逻辑，并返回Strawberry订阅字段对象：

from typing import Type, AsyncGenerator
import strawberry
from strawberry.types import Info
# 替换成你的项目导入
from your_app.models import Company, User
from your_app.types import CompanyType, UserType

def model_subscription(model: Type) -> strawberry.Field:
    # 动态生成订阅解析器，复用原有逻辑
    async def resolver(info: Info, pk: str) -> AsyncGenerator[object, None]:
        async for item in model_subscribe_publisher(info=info, pk=pk, model=model):
            yield item
    
    # 维护模型与GraphQL类型的映射（可优化为自动绑定）
    model_type_map = {
        Company: CompanyType,
        User: UserType,
        # 其他40+模型的映射
    }
    
    return strawberry.subscription_field(
        resolver=resolver,
        type_=model_type_map[model],
        arguments={"pk": strawberry.argument(str)},
    )

步骤2：简化订阅类写法 #

现在可以像Query类一样，用一行赋值定义每个订阅字段：

@strawberry.type(name="Subscription")
class ContactsSubscription:
    company: AsyncGenerator[CompanyType, None] = model_subscription(Company)
    user: AsyncGenerator[UserType, None] = model_subscription(User)
    # 其他模型订阅直接追加一行即可

优化：自动绑定模型与GraphQL类型 #

手动维护映射字典比较繁琐，可以给模型添加类属性直接绑定对应的GraphQL类型：

# 模型定义示例
class Company(models.Model):
    # 模型字段...
    graphql_type = CompanyType

class User(models.Model):
    # 模型字段...
    graphql_type = UserType

然后修改工厂函数，直接读取模型的graphql_type属性：

def model_subscription(model: Type) -> strawberry.Field:
    async def resolver(info: Info, pk: str) -> AsyncGenerator[object, None]:
        async for item in model_subscribe_publisher(info=info, pk=pk, model=model):
            yield item
    
    return strawberry.subscription_field(
        resolver=resolver,
        type_=model.graphql_type,
        arguments={"pk": strawberry.argument(str)},
    )

这样新增模型时，只要给模型绑定graphql_type，再在订阅类里加一行赋值即可，完全消除重复代码。

内容的提问来源于stack exchange，提问作者S.D.