aiodataloader

If you need to ensure every .load() call results in a new fetch, set cache=False in the constructor or as a class attribute.

Warning: When cache=False, your batch_load_fn may receive duplicate keys in a single batch if .load() was called multiple times for the same key within the same event loop tick. Your batch function must be able to handle these duplicates.

class MyLoader(DataLoader):
    cache = False
    async def batch_load_fn(self, keys):
        # If .load('A') was called twice, keys might be ['A', 'B', 'A']
        return keys

my_loader = MyLoader()
my_loader.load('A')
my_loader.load('B')
my_loader.load('A')

If a resource can be accessed via different keys (e.g., an id or a username), you can use the .prime(key, value) method to manually populate the cache of one loader from the results of another. This ensures that if a user is loaded by ID, they are also immediately available in the cache when searched by username, avoiding redundant batch requests.

async def user_by_id_batch_fn(ids):
    users = await gen_users_by_id(ids)
    for user in users:
        username_loader.prime(user.username, user)
    return users

user_by_id_loader = DataLoader(user_by_id_batch_fn)

async def username_batch_fn(names):
    users = await gen_usernames(names)
    for user in users:
        user_by_id_loader.prime(user.id, user)
    return users

username_loader = DataLoader(username_batch_fn)

To ensure security and data isolation between different users (e.g., in a web server), it is recommended to create a new instance of DataLoader for every incoming request. This prevents one user's cached data from being served to another user.

A common pattern is to return a dictionary of loaders that can be passed through the application context or root_value.

def create_loaders(auth_token):
  return {
    'users': DataLoader(lambda ids: gen_users(auth_token, ids)),
    'cdn_urls': DataLoader(lambda raw_urls: gen_cdn_urls(auth_token, raw_urls)),
    'stories': DataLoader(lambda keys: gen_stories(auth_token, keys)),
  }

# When handling an incoming web request:
loaders = create_loaders(request.query.auth_token)

# Then, within application logic:
user = await loaders.users.load(4)

Install the aiodataloader package using pip to use it in your AsyncIO environment (requires Python 3.5+).

pip install aiodataloader

DataLoader is highly effective in GraphQL environments to prevent the N+1 query problem. Instead of each field resolver issuing a separate database request, you can use load() for single items and load_many() for lists. This collapses multiple field resolutions into a single batch request.

When using graphene, you can implement resolvers that call the loader instead of querying the database directly.

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

    def resolve_best_friend(self, args, context, info):
        return user_loader.load(self.best_friend_id)

    def resolve_friends(self, args, context, info):
        return user_loader.load_many(self.friend_ids)

By default, DataLoader uses a standard Python dict for its memoization cache, which grows indefinitely until the loader is released.

You can provide a custom cache object during initialization to implement specific cache eviction algorithms (like LRU). The custom object must implement the following methods:

• get(key)
• set(key, value)
• delete(key)
• clear()

The DataLoader class provides the following methods for interacting with the data layer:

• DataLoader(batch_load_fn, **options): Constructor.
• load(key): Returns a Future for the value represented by key.
• load_many(keys): Returns a list of values for the provided list of keys. Equivalent to await asyncio.gather(*[loader.load(k) for k in keys]).
• clear(key): Clears the value at key from the cache. Returns self for chaining.
• clear_all(): Clears the entire cache. Returns self for chaining.
• prime(key, value): Manually adds a key and value to the cache. If the key exists, no change is made. To overwrite, use loader.clear(key).prime(key, value). Returns self for chaining.

# Using load_many
a, b = await my_loader.load_many(['a', 'b'])

# Using prime for manual cache injection
my_loader.prime('existing_key', 'some_value')

When initializing DataLoader(batch_load_fn, **options), you can provide the following configuration keys: