如何为Graphene中所有Mutation批量应用认证或检查机制？

当然可以在Mutations类层级统一配置批量装饰，以下是三种实用方案，可根据你的场景选择：

方法1：自定义带认证的Field生成函数 #

通过封装原Field()方法，自动为目标Mutation的核心方法添加装饰器，无需修改原有Mutation类的定义：

from functools import wraps
import graphene
from graphene import relay

# 示例认证装饰器
def login_required(func):
    @wraps(func)
    def wrapper(cls, root, info, *args, **kwargs):
        if not info.context.user.is_authenticated:
            raise Exception("用户未登录，请先认证")
        return func(cls, root, info, *args, **kwargs)
    return wrapper

# 自定义批量装饰的Field生成函数
def AuthenticatedMutationField(mutation_cls, **kwargs):
    # 自动装饰Mutation的核心方法
    if hasattr(mutation_cls, 'mutate_and_get_payload'):
        mutation_cls.mutate_and_get_payload = login_required(mutation_cls.mutate_and_get_payload)
    elif hasattr(mutation_cls, 'mutate'):
        mutation_cls.mutate = login_required(mutation_cls.mutate)
    return mutation_cls.Field(**kwargs)

# 使用示例
class SomeMutation(relay.ClientIDMutation):
    class Input:
        some_uuid = graphene.UUID()
    success = graphene.Boolean()
    @classmethod
    def mutate_and_get_payload(cls, root, info, **inputs):
        return cls(success=True)

class Mutations(graphene.ObjectType):
    # 用自定义的AuthenticatedMutationField替代原Field
    some_mutation = AuthenticatedMutationField(SomeMutation)
    some_mutation2 = AuthenticatedMutationField(SomeMutation2)
    some_mutation3 = AuthenticatedMutationField(SomeMutation3)

graphql_schema = graphene.Schema(query=Queries, mutation=Mutations)

方法2：利用元类自动批量装饰 #

通过元类遍历Mutations类的所有属性，自动识别并装饰所有Mutation的核心方法，完全无需修改现有字段定义：

class AuthenticatedMutationMeta(type):
    def __new__(cls, name, bases, attrs):
        # 遍历类内所有属性，识别Mutation字段
        for attr_name, attr_value in attrs.items():
            if hasattr(attr_value, 'type'):
                mutation_cls = attr_value.type
                # 装饰目标方法
                if hasattr(mutation_cls, 'mutate_and_get_payload'):
                    mutation_cls.mutate_and_get_payload = login_required(mutation_cls.mutate_and_get_payload)
                elif hasattr(mutation_cls, 'mutate'):
                    mutation_cls.mutate = login_required(mutation_cls.mutate)
        return super().__new__(cls, name, bases, attrs)

# 为Mutations类指定元类
class Mutations(graphene.ObjectType, metaclass=AuthenticatedMutationMeta):
    some_mutation = SomeMutation.Field()
    some_mutation2 = SomeMutation2.Field()
    some_mutation3 = SomeMutation3.Field()

graphql_schema = graphene.Schema(query=Queries, mutation=Mutations)

方法3：定义带认证的Mutation基类 #

创建一个包含认证装饰器的基类，所有业务Mutation继承该基类即可自动获得认证检查，适合新项目统一规范：

class AuthenticatedClientIDMutation(relay.ClientIDMutation):
    @classmethod
    @login_required
    def mutate_and_get_payload(cls, root, info, **inputs):
        raise NotImplementedError("子类必须实现mutate_and_get_payload方法")

# 业务Mutation继承认证基类
class SomeMutation(AuthenticatedClientIDMutation):
    class Input:
        some_uuid = graphene.UUID()
    success = graphene.Boolean()
    @classmethod
    def mutate_and_get_payload(cls, root, info, **inputs):
        return cls(success=True)

# Mutations类正常定义即可
class Mutations(graphene.ObjectType):
    some_mutation = SomeMutation.Field()
    some_mutation2 = SomeMutation2.Field()
    some_mutation3 = SomeMutation3.Field()

方案对比 #

• 方法1：灵活性高，可针对单个Mutation选择是否启用认证，适合混合场景
• 方法2：完全自动化，无需修改现有代码，适合所有Mutation都需要认证的场景
• 方法3：侵入性低，通过继承实现规范，适合新项目统一约束

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