Django GraphQL查询深度限制及Graphene动态字段实现咨询

你可以通过自定义**验证规则（ValidationRule）**实现查询深度限制，支持全局统一限制或按特定查询单独配置：

1. 全局统一深度限制 #

自定义验证类追踪查询深度，超出阈值时抛出错误：

from graphene.validation import ValidationRule
from graphql import GraphQLError

class DepthLimitRule(ValidationRule):
    def __init__(self, max_depth):
        self.max_depth = max_depth
        self.current_depth = 0

    def enter_field(self, node, key, parent, path, ancestors):
        self.current_depth += 1
        if self.current_depth > self.max_depth:
            raise GraphQLError(f"查询深度超过最大限制：{self.max_depth}")
        return super().enter_field(node, key, parent, path, ancestors)

    def leave_field(self, node, key, parent, path, ancestors):
        self.current_depth -= 1
        return super().leave_field(node, key, parent, path, ancestors)

在Django GraphQL视图中应用规则：

from django.views.decorators.csrf import csrf_exempt
from graphene_django.views import GraphQLView
from .schema import schema

@csrf_exempt
def graphql_view(request):
    # 根据请求路径区分公开/私有查询，设置不同深度
    max_depth = 3 if request.path == "/public-graphql/" else 6
    return GraphQLView.as_view(
        schema=schema,
        validation_rules=[DepthLimitRule(max_depth)]
    )(request)

2. 按查询名称单独设置深度 #

若需给不同查询接口配置不同深度，修改验证类识别查询名称：

class PerQueryDepthLimitRule(ValidationRule):
    def __init__(self, query_limits):
        # query_limits格式：{查询名称: 最大深度}
        self.query_limits = query_limits
        self.current_depth = 0
        self.current_query = None

    def enter_operation_definition(self, node, key, parent, path, ancestors):
        if node.name:
            self.current_query = node.name.value
        return super().enter_operation_definition(node, key, parent, path, ancestors)

    def enter_field(self, node, key, parent, path, ancestors):
        self.current_depth += 1
        if self.current_query in self.query_limits:
            max_depth = self.query_limits[self.current_query]
            if self.current_depth > max_depth:
                raise GraphQLError(f"查询「{self.current_query}」深度超过最大限制：{max_depth}")
        return super().enter_field(node, key, parent, path, ancestors)

    def leave_field(self, node, key, parent, path, ancestors):
        self.current_depth -= 1
        return super().leave_field(node, key, parent, path, ancestors)

视图中配置不同查询的深度：

@csrf_exempt
def graphql_view(request):
    query_depth_config = {
        "publicJobs": 2,
        "privateJobs": 5,
        "userProfile": 3
    }
    return GraphQLView.as_view(
        schema=schema,
        validation_rules=[PerQueryDepthLimitRule(query_depth_config)]
    )(request)

你提供的jobs = graphene.List(JobGQLType, fields=['title', 'location'])写法无法直接实现动态字段（Graphene类型定义为静态），但可以通过以下两种方式达成类似效果：

1. 在resolver中根据参数过滤字段 #

通过graphene.Argument接收字段列表，在resolver中只返回指定字段的数据：

import graphene
from .models import Job
from .types import JobGQLType

class Query(graphene.ObjectType):
    jobs = graphene.List(
        JobGQLType,
        fields=graphene.Argument(graphene.List(graphene.String), required=False)
    )

    def resolve_jobs(self, info, fields=None):
        queryset = Job.objects.all()
        if fields:
            # 仅加载指定字段，减少数据库查询开销
            queryset = queryset.only(*fields)
        # 若需严格只返回指定字段，可改用values()返回字典
        # return queryset.values(*fields)
        return queryset

2. 动态生成GQL类型（进阶） #

如果需要严格返回指定字段的结构，可动态生成对应的GQL类型：

import graphene
from graphene_django.converter import convert_django_field
from .models import Job

def create_dynamic_job_type(fields):
    # 根据传入字段生成动态类型
    dynamic_fields = {}
    for field_name in fields:
        model_field = Job._meta.get_field(field_name)
        dynamic_fields[field_name] = convert_django_field(model_field)
    return type("DynamicJobGQLType", (graphene.ObjectType,), dynamic_fields)

class Query(graphene.ObjectType):
    jobs = graphene.Field(
        graphene.List(graphene.ObjectType),
        fields=graphene.Argument(graphene.List(graphene.String), required=True)
    )

    def resolve_jobs(self, info, fields):
        dynamic_type = create_dynamic_job_type(fields)
        # 查询并转换为动态类型实例
        job_data = Job.objects.values(*fields)
        return [dynamic_type(**item) for item in job_data]

注意：动态生成类型可能影响性能，建议对生成的类型进行缓存，避免每次请求都创建新类型。

内容的提问来源于stack exchange，提问作者Manzurul Hoque Rumi