Dataloader's documentation

从链接的 DataLoader 文档开始，查看 User resolver 示例，包括 root.best_friend_id、root.friend_ids 和 User.objects.filter(id__in=keys)。将这些示例与所描述的辅助函数进行比较，并明确哪些代码可以运行，哪些代码仅用于说明。完成的标准是文档呈现一个自包含且定义明确的用例，并解释 batch 函数的预期行为。

The dataloader documentation is quite unclear in the current state, especially this piece of code:

class User(graphene.ObjectType):
    name = graphene.String()
    best_friend = graphene.Field(lambda: User)
    friends = graphene.List(lambda: User)

    def resolve_best_friend(root, info):
        return user_loader.load(root.best_friend_id)

    def resolve_friends(root, info):
        return user_loader.load_many(root.friend_ids)

It's not clear how this could work, because as it's presented here, neither root.best_friend_id nor root.friend_ids would be defined. Wouldn't it make sense to describe a real use case? The current documentation is more or less useless because when you first read it you can't tell what is an approximation and what is not (example of another non-working bit of code User.objects.filter(id__in=keys)). I'm not even sure yet about all this because I know graphene rewrites classes but I don't know all the details since I only used it (and never contributed). So maybe there is to trickery hidden in these statements, but I really feel there are several errors in the provided examples.

Maybe the solution to this is only to add empty objects/resolver (pass) but with well defined types. So everything is clearly defined (but unimplemented).

I've tried to make sense about graphene + aioloader by writing three simple helpers function that create some batch functions:

from collections import defaultdict
from typing import Callable, Dict, List, Optional, Tuple, TypeVar

class OneToOne:
    T = TypeVar("T")
    K = TypeVar("K", str, int)
    Query = Callable[[List[K]], List[T]]
    Key = Callable[[T], K]


OneToMany = OneToOne


class ManyToMany:
    T = TypeVar("T")
    Query = Callable[[List[str]], List[Tuple[T, str]]]


def one_to_one(query: OneToOne.Query, key: OneToOne.Key):

    async def batch_function(keys: List[str]) -> List[Optional[OneToOne.T]]:
        user_objects = {key(obj): obj for obj in query(keys)}
        return [user_objects.get(k) for k in keys]

    return batch_function


def one_to_many(query: OneToMany.Query, key: OneToMany.Key):

    async def batch_function(keys: List[str]) -> List[List[OneToMany.T]]:
        objects = query(keys)
        user_objects = defaultdict(list)
        for obj in objects:
            user_objects[key(obj)].append(obj)
        return [user_objects.get(k) or [] for k in keys]

    return batch_function


def many_to_many(query: ManyToMany.Query):

    async def batch_function(keys: List[str]) -> List[List[ManyToMany.T]]:
        obj_key_pairs = query(keys)
        user_objects: Dict[str, List[ManyToMany.T]] = defaultdict(list)
        for obj, k in obj_key_pairs:
            user_objects[k].append(obj)
        return [user_objects.get(k) or [] for k in keys]

    return batch_function

Describing this kind of functions might even help people understanding what is expected from them. I'm not sure yet these types (or even helper functions) are 100% correct, let me know if you find better solutions.