graphql-sync-dataloaders

在需要同步上下文（例如Django）运行的Python GraphQL服务器中使用DataLoaders。

需求

• Python 3.8+
• graphql-core >=3.2.0

安装

此软件包可以从PyPi通过运行

pip install graphql-sync-dataloaders

Strawberry设置

在创建您的Strawberry Schema时，将DeferredExecutionContext作为execution_context_class参数传递

# schema.py
import strawberry
from graphql_sync_dataloaders import DeferredExecutionContext

schema = strawberry.Schema(Query, execution_context_class=DeferredExecutionContext)

然后使用SyncDataLoader类创建您的dataloaders

from typing import List

from graphql_sync_dataloaders import SyncDataLoader

from .app import models  # your Django models

def load_users(keys: List[int]) -> List[User]:
    qs = models.User.objects.filter(id__in=keys)
    user_map = {user.id: user for user in qs}
    return [user_map.get(key, None) for key in keys]

user_loader = SyncDataLoader(load_users)

然后您可以在解析器中使用加载器，它将自动批量处理以减少SQL查询数量

import strawberry

@strawberry.type
class Query:
    @strawberry.field
    def get_user(self, id: strawberry.ID) -> User:
        return user_loader.load(id)

注意：您可能需要在上下文中设置加载器。有关更多详细信息，请参阅https://strawberry.rocks/docs/guides/dataloaders#usage-with-context

以下查询将仅执行1个SQL查询

fragment UserDetails on User {
  username
}

query {
  user1: getUser(id: '1') {
    ...UserDetails
  }
  user2: getUser(id: '2') {
    ...UserDetails
  }
  user3: getUser(id: '3') {
    ...UserDetails
  }
}

Graphene-Django设置

需要graphene-django >=3.0.0b8

在设置GraphQLView时，将DeferredExecutionContext作为execution_context_class参数传递

# urls.py
from django.urls import path
from graphene_django.views import GraphQLView
from graphql_sync_dataloaders import DeferredExecutionContext

from .schema import schema

urlpatterns = [
    path(
        "graphql",
        csrf_exempt(
            GraphQLView.as_view(
                schema=schema, 
                execution_context_class=DeferredExecutionContext
            )
        ),
    ),
]

然后使用SyncDataLoader类创建您的dataloaders

from typing import List

from graphql_sync_dataloaders import SyncDataLoader

from .app import models  # your Django models

def load_users(keys: List[int]) -> List[User]:
    qs = models.User.objects.filter(id__in=keys)
    user_map = {user.id: user for user in qs}
    return [user_map.get(key, None) for key in keys]

user_loader = SyncDataLoader(load_users)

然后您可以在解析器中使用加载器，它将自动批量处理以减少SQL查询数量

import graphene

class Query(graphene.ObjectType):
    get_user = graphene.Field(User, id=graphene.ID)

    def resolve_get_user(root, info, id):
        return user_loader.load(id)

以下查询将仅执行1个SQL查询

fragment UserDetails on User {
  username
}

query {
  user1: getUser(id: '1') {
    ...UserDetails
  }
  user2: getUser(id: '2') {
    ...UserDetails
  }
  user3: getUser(id: '3') {
    ...UserDetails
  }
}

工作原理

此库实现了一个自定义版本的graphql-core ExecutionContext类，该类了解库中定义的SyncFuture对象。一个SyncFuture代表一个尚未解析为值的值（类似于asyncio的Futures或JavaScript的Promises），当您调用.load函数时，SyncDataLoader返回的就是这样的值。

当自定义的ExecutionContext遇到一个由解析器返回的SyncFuture时，它会跟踪它们。然后在执行的第一遍之后，它触发SyncFuture回调，直到没有剩余的回调。一旦没有剩余的回调，数据就完全解析完毕，可以同步返回给调用者。这使我们能够实现一个DataLoader模式，该模式批量调用加载函数，并且可以完全同步地执行。

致谢

@Cito为graphql-core，并在https://github.com/graphql-python/graphql-core/pull/155中实现了此库的第一个版本。