Django中Graphene结合DataLoader的可用实现示例求助

1. 安装依赖 #

pip install graphene-django graphql-sync-dataloaders

2. 定义自定义SyncDataLoader #

创建loaders.py，实现针对Django模型的批量加载逻辑：

from graphql_sync_dataloaders import SyncDataLoader
from django.db.models import Model
from .models import User  # 替换为你的实际模型

class ModelDataLoader(SyncDataLoader):
    def __init__(self, model: Model):
        self.model = model
        super().__init__()

    def load_many(self, keys):
        # 批量查询数据库，按主键分组
        queryset = self.model.objects.filter(id__in=keys)
        obj_map = {obj.id: obj for obj in queryset}
        # 保持传入keys的顺序返回，不存在的项返回None
        return [obj_map.get(key) for key in keys]

# 针对User模型的专用Loader
class UserLoader(ModelDataLoader):
    def __init__(self):
        super().__init__(model=User)

3. 将Loader注入GraphQL上下文 #

在Django视图中，为每个请求初始化Loader并传入上下文：

# views.py
from django.views.decorators.csrf import csrf_exempt
from graphene_django.views import GraphQLView
from .loaders import UserLoader
from .schema import schema  # 你的Graphene Schema

def get_request_context(request):
    context = {}
    # 每个请求初始化独立的Loader实例，避免跨请求数据污染
    context['user_loader'] = UserLoader()
    return context

@csrf_exempt
def graphql_view(request):
    return GraphQLView.as_view(
        schema=schema,
        context=get_request_context(request)
    )(request)

4. 在Graphene解析器中使用Loader #

在你的类型定义里，通过上下文调用Loader实现批量加载：

# schema.py
import graphene
from graphene_django import DjangoObjectType
from .models import Post, User
from .loaders import UserLoader

class UserType(DjangoObjectType):
    class Meta:
        model = User
        fields = ("id", "username", "email")

class PostType(DjangoObjectType):
    author = graphene.Field(UserType)

    def resolve_author(self, info):
        # 调用Loader批量加载关联的用户，自动合并同请求内的查询
        return info.context['user_loader'].load(self.author_id)

class Query(graphene.ObjectType):
    all_posts = graphene.List(PostType)

    def resolve_all_posts(self, info):
        return Post.objects.all()

schema = graphene.Schema(query=Query)

核心说明 #

• SyncDataLoader专为同步执行环境设计，完全适配Graphene默认的同步执行器，不会触发事件循环相关错误。
• 每个请求初始化独立的Loader实例，确保批量加载的范围仅限当前请求，避免数据交叉污染。
• load()方法会自动收集同一请求内的所有加载请求，在合适时机执行批量数据库查询，彻底解决N+1查询性能问题。

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