Asyncio DataLoader

https://github.com/syrusakbary/aiodataloader
Admin
Asyncio DataLoader is a Python utility that provides a simplified and consistent API for data...
Tokens:9,388
Snippets:73
Trust Score:10
License:MIT
Update:4 months ago
Show doc for...
Tokens:
Raw
### Install aiodataloader

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Install the aiodataloader package using pip.

```sh
pip install aiodataloader
```

--------------------------------

### Load single values with batching and caching

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Load a key and get a Future. Multiple loads in one tick are batched. Repeated loads of the same key use the cache.

```python
import asyncio
from aiodataloader import DataLoader

async def batch_fn(keys):
    # Simulated DB: returns values matching keys
    db = {1: {"id": 1, "name": "Alice"}, 2: {"id": 2, "name": "Bob"}}
    return [db.get(k) for k in keys]

async def main():
    loader = DataLoader(batch_fn)

    # Both loads are issued before either is awaited,
    # so they are coalesced into ONE batch_fn call: batch_fn([1, 2])
    future1 = loader.load(1)
    future2 = loader.load(2)

    user1 = await future1
    user2 = await future2

    print(user1)  # {'id': 1, 'name': 'Alice'}
    print(user2)  # {'id': 2, 'name': 'Bob'}

    # Repeated load of key 1 returns the cached Future — no second DB call
    user1_again = await loader.load(1)
    print(user1_again)  # {'id': 1, 'name': 'Alice'}  (from cache)

asyncio.run(main())
```

--------------------------------

### Batch function return value example

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

The batch loading function must return a list of values that is the same length as the list of keys provided. Each value in the returned list must correspond to the key at the same index in the input list. Use None for keys that do not have a corresponding value.

```python
[
  { 'id': 2, 'name': 'San Francisco' },
  { 'id': 9, 'name': 'Chicago' },
  None,
  { 'id': 1, 'name': 'New York' }
]
```

--------------------------------

### Define User Type with DataLoader Integration

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Integrate DataLoader with GraphQL using `graphene`. This example shows how to define a User type with fields that use DataLoader for efficient data fetching.

```python
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)
```

--------------------------------

### Implement LRUCache with aiodataloader

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Demonstrates how to use a custom LRUCache class with the `cache_map` option in DataLoader for custom caching strategies. Ensure the custom cache object implements `get`, `__setitem__`, `__getitem__`, `pop`, and `clear` methods.

```python
import asyncio
from functools import lru_cache
from collections import OrderedDict
from aiodataloader import DataLoader

class LRUCache:
    """Simple LRU cache with a fixed capacity."""
    def __init__(self, capacity=128):
        self._data = OrderedDict()
        self._capacity = capacity

    def get(self, key, default=None):
        if key not in self._data:
            return default
        self._data.move_to_end(key)
        return self._data[key]

    def __setitem__(self, key, value):
        if key in self._data:
            self._data.move_to_end(key)
        self._data[key] = value
        if len(self._data) > self._capacity:
            self._data.popitem(last=False)

    def __getitem__(self, key):
        return self._data[key]

    def __contains__(self, key):
        return key in self._data

    def pop(self, key, *args):
        return self._data.pop(key, *args)

    def clear(self):
        self._data.clear()

async def batch_fn(keys):
    return keys

async def main():
    lru = LRUCache(capacity=10)
    loader = DataLoader(batch_fn, cache_map=lru)

    await loader.load_many([1, 2, 3])
    print(list(lru._data.keys()))  # [1, 2, 3]

asyncio.run(main())
```

--------------------------------

### Clear Cache Immediately After Batch Load

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

An example of a DataLoader that disables its cache but immediately clears it after each batch load. This ensures that while duplicates might be sent to the batch function, subsequent requests for the same key will always trigger a new load.

```python
class MyLoader(DataLoader):
    cache = False
    async def batch_load_fn(self, keys):
        self.clear_all()
        return keys
```

--------------------------------

### Initialize DataLoader with Sanic

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Demonstrates how to create DataLoader instances within the context of a web request using the Sanic framework. Loaders are typically created per request.

```python
def create_loaders(auth_token) {
    return {
      'users': user_loader,
    }
}


app = Sanic(__name__)

@app.route("/")
async def test(request):
    auth_token = authenticate_user(request)
    loaders = create_loaders(auth_token)
    return render_page(request, loaders)
```

--------------------------------

### Loading by Alternative Keys with DataLoader

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Demonstrates priming the cache of one DataLoader with data loaded by another. This is useful when a single data entity can be accessed by multiple unique identifiers.

```python
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)
```

--------------------------------

### DataLoader(batch_load_fn, **options)

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Instantiate a DataLoader by subclassing it and overriding batch_load_fn, or by passing a coroutine function directly. The batch function receives a list of keys and must return a list of values of the same length and in the same order. Options include batch, max_batch_size, cache, get_cache_key, and cache_map.

```APIDOC
## DataLoader(batch_load_fn, **options) — Create a loader

Instantiate a `DataLoader` by subclassing it and overriding `batch_load_fn`, or by passing a coroutine function directly. The batch function receives a list of keys and must return a list of values **of the same length and in the same order**. Options include `batch` (bool, default `True`), `max_batch_size` (int), `cache` (bool, default `True`), `get_cache_key` (callable), and `cache_map` (dict-like object).

```python
import asyncio
from aiodataloader import DataLoader

# --- Approach 1: subclass ---
class UserLoader(DataLoader):
    async def batch_load_fn(self, user_ids):
        # Simulate a DB call returning users in the same order as keys
        rows = await fetch_users_from_db(user_ids)
        user_map = {u["id"]: u for u in rows}
        return [user_map.get(uid) for uid in user_ids]

# --- Approach 2: pass coroutine directly ---
async def batch_load_users(user_ids):
    rows = await fetch_users_from_db(user_ids)
    user_map = {u["id"]: u for u in rows}
    return [user_map.get(uid) for uid in user_ids]

user_loader = DataLoader(batch_load_users)

# --- Approach 3: functools.partial for context injection ---
from functools import partial

async def batch_load_with_auth(auth_token, user_ids):
    rows = await fetch_users_from_db(user_ids, token=auth_token)
    user_map = {u["id"]: u for u in rows}
    return [user_map.get(uid) for uid in user_ids]

authed_loader = DataLoader(partial(batch_load_with_auth, "secret-token"))
```
```

--------------------------------

### Create DataLoader instances per request

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Illustrates the per-request loader factory pattern for web applications. A new DataLoader instance should be created for each incoming request to ensure data isolation between users and cache separation.

```python
import asyncio
from aiodataloader import DataLoader

# --- Loader factory ---
async def gen_users(auth_token, ids):
    # auth_token scopes the query to the requesting user's permissions
    print(f"[{auth_token}] Fetching users: {ids}")
    db = {1: "Alice", 2: "Bob", 3: "Carol"}
    return [db.get(i) for i in ids]

def create_loaders(auth_token):
    return {
        "users": DataLoader(lambda ids: gen_users(auth_token, ids)),
    }

# --- Sanic-style request handler ---
# from sanic import Sanic
# app = Sanic(__name__)
#
# @app.route("/profile/<user_id:int>")
# async def profile(request, user_id):
#     loaders = create_loaders(request.token)      # fresh loader per request
#     user = await loaders["users"].load(user_id)
#     return json({"user": user})

async def simulate_request(auth_token, user_id):
    loaders = create_loaders(auth_token)
    return await loaders["users"].load(user_id)

async def main():
    r1 = await simulate_request("token-alice", 1)
    r2 = await simulate_request("token-bob",   2)
    print(r1)  # Alice
    print(r2)  # Bob

asyncio.run(main())
```

--------------------------------

### Pre-populate cache with `prime(key, value)`

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Use `loader.prime(key, value)` to manually insert a key-value pair into the cache, bypassing the batch function. If the key exists, it's a no-op unless cleared first. Accepts exceptions for pre-caching failures and returns `self` for chaining.

```python
import asyncio
from aiodataloader import DataLoader

async def batch_fn(keys):
    print(f"batch_fn called with: {keys}")
    return keys

async def main():
    loader = DataLoader(batch_fn)

    # Pre-populate cache for key "A"
    loader.prime("A", "cached_value")

    a, b = await asyncio.gather(loader.load("A"), loader.load("B"))
    # batch_fn called with: ['B']  <-- "A" was served from primed cache
    print(a)  # cached_value
    print(b)  # B

    # prime() is a no-op if key already exists
    loader.prime("A", "ignored_value")
    a2 = await loader.load("A")
    print(a2)  # cached_value  (not overwritten)

    # Force-update with clear().prime()
    loader.clear("A").prime("A", "updated_value")
    a3 = await loader.load("A")
    print(a3)  # updated_value

asyncio.run(main())
```

--------------------------------

### Integrate DataLoader with Graphene for GraphQL

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Shows how to integrate DataLoader with Graphene resolvers to efficiently fetch data and avoid N+1 query problems. Each resolver should return a Future from `.load()` or `.load_many()`, which Graphene will await automatically.

```python
import asyncio
import graphene
from aiodataloader import DataLoader

# --- Loader setup (one instance per request) ---
async def batch_load_users(ids):
    db = {
        1: {"id": 1, "name": "Alice", "best_friend_id": 2},
        2: {"id": 2, "name": "Bob",   "best_friend_id": 1},
        3: {"id": 3, "name": "Carol", "best_friend_id": 1},
    }
    return [db.get(i) for i in ids]

user_loader = DataLoader(batch_load_users)

# --- Graphene types ---
class User(graphene.ObjectType):
    id   = graphene.Int()
    name = graphene.String()
    best_friend = graphene.Field(lambda: User)
    friends     = graphene.List(lambda: User)

    def resolve_best_friend(root, info):
        # Returns a Future — graphql-core awaits it
        return user_loader.load(root["best_friend_id"])

    def resolve_friends(root, info):
        friend_ids = [2, 3] if root["id"] == 1 else []
        return user_loader.load_many(friend_ids)

class Query(graphene.ObjectType):
    me = graphene.Field(User)

    async def resolve_me(root, info):
        return await user_loader.load(1)

schema = graphene.Schema(query=Query)

# For the query { me { name bestFriend { name } friends { name } } }
# only 2 batch calls are made instead of up to 13 naive DB round-trips.
```

--------------------------------

### Create DataLoader instances

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Instantiate DataLoader by subclassing or passing a coroutine function. The batch function must return values in the same order as the input keys.

```python
import asyncio
from aiodataloader import DataLoader

# --- Approach 1: subclass ---
class UserLoader(DataLoader):
    async def batch_load_fn(self, user_ids):
        # Simulate a DB call returning users in the same order as keys
        rows = await fetch_users_from_db(user_ids)
        user_map = {u["id"]: u for u in rows}
        return [user_map.get(uid) for uid in user_ids]

# --- Approach 2: pass coroutine directly ---
async def batch_load_users(user_ids):
    rows = await fetch_users_from_db(user_ids)
    user_map = {u["id"]: u for u in rows}
    return [user_map.get(uid) for uid in user_ids]

user_loader = DataLoader(batch_load_users)

# --- Approach 3: functools.partial for context injection ---
from functools import partial

async def batch_load_with_auth(auth_token, user_ids):
    rows = await fetch_users_from_db(user_ids, token=auth_token)
    user_map = {u["id"]: u for u in rows}
    return [user_map.get(uid) for uid in user_ids]

authed_loader = DataLoader(partial(batch_load_with_auth, "secret-token"))
```

--------------------------------

### Create DataLoaders per Request

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

This pattern demonstrates creating a new set of DataLoaders for each incoming web request to ensure data isolation and security. It returns a dictionary of DataLoaders for different data types.

```python
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)
pic = await loaders.cdn_urls.load(user.raw_pic_url)
```

--------------------------------

### DataLoader Constructor

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Initializes a new DataLoader instance with a batch loading function and optional configuration settings.

```APIDOC
## DataLoader(batch_load_fn, **options)

### Description
Create a new `DataLoader` given a batch loading function and options.

### Parameters
- *batch_load_fn*: An async function (coroutine) which accepts an List of keys and returns a Future which resolves to an List of values.
- *options*:
  - *batch*: Default `True`. Set to `False` to disable batching, instead immediately invoking `batch_load_fn` with a single load key.
  - *max_batch_size*: Default `Infinity`. Limits the number of items that get passed in to the `batch_load_fn`.
  - *cache*: Default `True`. Set to `False` to disable memoization caching, instead creating a new Promise and new key in the `batch_load_fn` for every load of the same key.
  - *cache_key_fn*: A function to produce a cache key for a given load key. Defaults to `key => key`. Useful to provide when Python objects are keys and two similarly shaped objects should be considered equivalent.
  - *cache_map*: An instance of [dict][] (or an object with a similar API) to be used as the underlying cache for this loader. Default `{}`.
```

--------------------------------

### load(key)

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Loads a key and returns an asyncio.Future representing the eventual value. All .load() calls within the same event loop tick are automatically batched together into a single batch_load_fn call. The key must not be None.

```APIDOC
## load(key) — Load a single value

Loads a key and returns an `asyncio.Future` representing the eventual value. All `.load()` calls within the same event loop tick are automatically batched together into a single `batch_load_fn` call. The key must not be `None`.

```python
import asyncio
from aiodataloader import DataLoader

async def batch_fn(keys):
    # Simulated DB: returns values matching keys
    db = {1: {"id": 1, "name": "Alice"}, 2: {"id": 2, "name": "Bob"}}
    return [db.get(k) for k in keys]

async def main():
    loader = DataLoader(batch_fn)

    # Both loads are issued before either is awaited,
    # so they are coalesced into ONE batch_fn call: batch_fn([1, 2])
    future1 = loader.load(1)
    future2 = loader.load(2)

    user1 = await future1
    user2 = await future2

    print(user1)  # {'id': 1, 'name': 'Alice'}
    print(user2)  # {'id': 2, 'name': 'Bob'}

    # Repeated load of key 1 returns the cached Future — no second DB call
    user1_again = await loader.load(1)
    print(user1_again)  # {'id': 1, 'name': 'Alice'}  (from cache)

asyncio.run(main())
```
```

--------------------------------

### prime(key, value)

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Inserts a key/value pair directly into the memoization cache, bypassing the batch function. If the key already exists, no change is made (use `loader.clear(key).prime(key, value)` to force-update). Returns `self` for method chaining. Accepts `Exception` instances as values to pre-cache failures.

```APIDOC
## `prime(key, value)` — Pre-populate the cache

Inserts a key/value pair directly into the memoization cache, bypassing the batch function. If the key already exists, no change is made (use `loader.clear(key).prime(key, value)` to force-update). Returns `self` for method chaining. Accepts `Exception` instances as values to pre-cache failures.

```python
import asyncio
from aiodataloader import DataLoader

async def batch_fn(keys):
    print(f"batch_fn called with: {keys}")
    return keys

async def main():
    loader = DataLoader(batch_fn)

    # Pre-populate cache for key "A"
    loader.prime("A", "cached_value")

    a, b = await asyncio.gather(loader.load("A"), loader.load("B"))
    # batch_fn called with: ['B']  <-- "A" was served from primed cache
    print(a)  # cached_value
    print(b)  # B

    # prime() is a no-op if key already exists
    loader.prime("A", "ignored_value")
    a2 = await loader.load("A")
    print(a2)  # cached_value  (not overwritten)

    # Force-update with clear().prime()
    loader.clear("A").prime("A", "updated_value")
    a3 = await loader.load("A")
    print(a3)  # updated_value

asyncio.run(main())
```
```

--------------------------------

### Define a UserLoader

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Create a custom DataLoader by subclassing DataLoader and implementing the batch_load_fn method. This method should accept a list of keys and return a future that resolves to a list of values.

```python
from aiodataloader import DataLoader

class UserLoader(DataLoader):
    async def batch_load_fn(self, keys):
        return await my_batch_get_users(keys)

user_loader = UserLoader()
```

--------------------------------

### load(key)

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Loads a single key and returns a Future for its corresponding value.

```APIDOC
## load(key)

### Description
Loads a key, returning a `Future` for the value represented by that key.

### Parameters
- *key*: An key value to load.
```

--------------------------------

### Load multiple values at once

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Load multiple keys efficiently using load_many. Handles empty input gracefully.

```python
import asyncio
from aiodataloader import DataLoader

async def batch_fn(keys):
    db = {"a": 100, "b": 200, "c": 300}
    return [db.get(k) for k in keys]

async def main():
    loader = DataLoader(batch_fn)

    # Load multiple keys in a single call
    values = await loader.load_many(["a", "b", "c"])
    print(values)  # [100, 200, 300]

    # Empty list is handled gracefully
    empty = await loader.load_many([])
    print(empty)  # []

asyncio.run(main())
```

--------------------------------

### Verify DataLoader Caching

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Asserts that loading the same key twice within a single request context returns the same future, demonstrating DataLoader's memoization.

```python
user_future1 = user_loader.load(1)
user_future2 = user_loader.load(1)

assert user_future1 == user_future2
```

--------------------------------

### Load Many Keys with DataLoader

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Use `load_many` to efficiently load multiple keys. This is equivalent to using `asyncio.gather` on individual `load` calls but is more concise.

```python
a, b = await my_loader.load_many([ 'a', 'b' ]);
```

```python
from asyncio import gather
a, b = await gather(
    my_loader.load('a'),
    my_loader.load('b')
)
```

--------------------------------

### max_batch_size option

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Caps the number of keys sent to `batch_load_fn` in a single call. When more keys than the cap are pending, the queue is split into multiple chunks of at most `max_batch_size` and dispatched as separate batch calls.

```APIDOC
## `max_batch_size` option — Limit batch size

Caps the number of keys sent to `batch_load_fn` in a single call. When more keys than the cap are pending, the queue is split into multiple chunks of at most `max_batch_size` and dispatched as separate batch calls.

```python
import asyncio
from aiodataloader import DataLoader

batches_issued = []

async def batch_fn(keys):
    batches_issued.append(list(keys))
    return keys

async def main():
    # Limit each batch to 2 keys at most
    loader = DataLoader(batch_fn, max_batch_size=2)

    results = await asyncio.gather(
        loader.load(1),
        loader.load(2),
        loader.load(3),
        loader.load(4),
        loader.load(5),
    )

    print(results)        # [1, 2, 3, 4, 5]
    print(batches_issued) # [[1, 2], [3, 4], [5]]

asyncio.run(main())
```
```

--------------------------------

### Limit batch size with `max_batch_size`

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Configure `max_batch_size` in `DataLoader` to cap the number of keys sent in a single batch call. This splits larger key sets into multiple smaller batches, controlling network load and downstream processing.

```python
import asyncio
from aiodataloader import DataLoader

batches_issued = []

async def batch_fn(keys):
    batches_issued.append(list(keys))
    return keys

async def main():
    # Limit each batch to 2 keys at most
    loader = DataLoader(batch_fn, max_batch_size=2)

    results = await asyncio.gather(
        loader.load(1),
        loader.load(2),
        loader.load(3),
        loader.load(4),
        loader.load(5),
    )

    print(results)        # [1, 2, 3, 4, 5]
    print(batches_issued) # [[1, 2], [3, 4], [5]]

asyncio.run(main())
```

--------------------------------

### prime(key, value)

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Manually adds or updates a key-value pair in the cache.

```APIDOC
## prime(key, value)

### Description
Primes the cache with the provided key and value. If the key already exists, no change is made. (To forcefully prime the cache, clear the key first with `loader.clear(key).prime(key, value)`.) Returns itself for method chaining.

### Parameters
- *key*: The key to add to the cache.
- *value*: The value to associate with the key.
```

--------------------------------

### clear_all()

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Clears all entries from the cache.

```APIDOC
## clear_all()

### Description
Clears the entire cache. To be used when some event results in unknown invalidations across this particular `DataLoader`. Returns itself for method chaining.
```

--------------------------------

### load_many(keys)

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Accepts an iterable of keys and returns a single Future that resolves to a list of values. Equivalent to asyncio.gather(*[loader.load(k) for k in keys]), but more concise. Returns an empty list for an empty input.

```APIDOC
## load_many(keys) — Load multiple values at once

Accepts an iterable of keys and returns a single `Future` that resolves to a list of values. Equivalent to `asyncio.gather(*[loader.load(k) for k in keys])`, but more concise. Returns an empty list for an empty input.

```python
import asyncio
from aiodataloader import DataLoader

async def batch_fn(keys):
    db = {"a": 100, "b": 200, "c": 300}
    return [db.get(k) for k in keys]

async def main():
    loader = DataLoader(batch_fn)

    # Load multiple keys in a single call
    values = await loader.load_many(["a", "b", "c"])
    print(values)  # [100, 200, 300]

    # Empty list is handled gracefully
    empty = await loader.load_many([])
    print(empty)  # []

asyncio.run(main())
```
```

--------------------------------

### Load individual values with DataLoader

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Load individual values from a DataLoader instance. DataLoader will automatically batch concurrent load requests within a single event loop tick and call the batch_load_fn with all requested keys.

```python
user1_future = user_loader.load(1)
user2_future = user_loader.load(2)

user1 = await user1_future
user2 = await user2_future

user1_invitedby = user_loader.load(user1.invited_by_id)
user2_invitedby = user_loader.load(user2.invited_by_id)

print("User 1 was invited by", await user1_invitedby)
print("User 2 was invited by", await user2_invitedby)
```

--------------------------------

### cache=False option

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Disables the memoization cache so every `.load()` call produces a fresh `Future` and forces a new key into the batch, even for duplicate keys. Useful when data changes frequently and stale reads are not acceptable.

```APIDOC
## `cache=False` option — Disable memoization

Disables the memoization cache so every `.load()` call produces a fresh `Future` and forces a new key into the batch, even for duplicate keys. Useful when data changes frequently and stale reads are not acceptable.

```python
import asyncio
from aiodataloader import DataLoader

batches = []

async def batch_fn(keys):
    batches.append(list(keys))
    return keys

async def main():
    loader = DataLoader(batch_fn, cache=False)

    # Duplicate key "A" is NOT deduplicated when cache is off
    await asyncio.gather(loader.load("A"), loader.load("B"), loader.load("A"))
    print(batches)  # [['A', 'B', 'A']]  -- 'A' appears twice

asyncio.run(main())
```
```

--------------------------------

### clear_all()

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Clears every entry in the memoization cache at once. Returns `self` for method chaining. Use when a broad mutation or event makes all cached data suspect.

```APIDOC
## `clear_all()` — Invalidate the entire cache

Clears every entry in the memoization cache at once. Returns `self` for method chaining. Use when a broad mutation or event makes all cached data suspect.

```python
import asyncio
from aiodataloader import DataLoader

call_count = 0

async def batch_fn(keys):
    global call_count
    call_count += 1
    return keys  # identity loader

async def main():
    loader = DataLoader(batch_fn)

    await loader.load_many(["A", "B"])
    print(call_count)  # 1 — one batch

    await loader.load_many(["A", "B"])
    print(call_count)  # 1 — served from cache

    loader.clear_all()

    await loader.load_many(["A", "B"])
    print(call_count)  # 2 — cache was cleared, re-fetched

asyncio.run(main())
```
```

--------------------------------

### load_many(keys)

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Loads multiple keys concurrently and returns a Future for a list of their corresponding values.

```APIDOC
## load_many(keys)

### Description
Loads multiple keys, promising an array of values.

### Parameters
- *keys*: A list of key values to load.

### Example
```python
a, b = await my_loader.load_many([ 'a', 'b' ]);
```
```

--------------------------------

### Disable DataLoader Caching

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Demonstrates disabling the memoization cache for a DataLoader instance by setting `cache=False`. This results in every `.load()` call producing a new Future and potentially duplicate keys in the batch function.

```python
class MyLoader(DataLoader):
    cache = False
    async def batch_load_fn(self, keys):
        print(keys)
        return keys

my_loader = MyLoader()

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

# > [ 'A', 'B', 'A' ]
```

--------------------------------

### Clear Individual Cached Exception

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Shows how to clear the cache for an individual error that was previously cached by DataLoader. This is useful if the condition causing the error has been resolved.

```python
try:
    user_loader.load(1)
except Exception as e:
    user_loader.clear(1)
    raise
```

--------------------------------

### Clear specific cache entries with `clear()`

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Use `loader.clear(key)` to invalidate a single cache entry. This is useful after a specific data mutation to ensure the next load fetches fresh data. It supports method chaining.

```python
import asyncio
from aiodataloader import DataLoader

db = {4: {"id": 4, "username": "old_name"}}

async def batch_fn(keys):
    return [db.get(k) for k in keys]

async def main():
    loader = DataLoader(batch_fn)

    user = await loader.load(4)
    print(user)  # {'id': 4, 'username': 'old_name'}

    # Simulate a mutation
    db[4] = {"id": 4, "username": "new_name"}

    # Without clear(), the stale cached value would be returned
    loader.clear(4)

    user_fresh = await loader.load(4)
    print(user_fresh)  # {'id': 4, 'username': 'new_name'}

    # clear() returns self — supports chaining
    loader.clear(1).clear(2).clear(3)

asyncio.run(main())
```

--------------------------------

### Invalidate entire cache with `clear_all()`

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Call `loader.clear_all()` to invalidate all entries in the cache at once. This is recommended when a broad data mutation occurs, making all cached data potentially stale. It returns `self` for chaining.

```python
import asyncio
from aiodataloader import DataLoader

call_count = 0

async def batch_fn(keys):
    global call_count
    call_count += 1
    return keys  # identity loader

async def main():
    loader = DataLoader(batch_fn)

    await loader.load_many(["A", "B"])
    print(call_count)  # 1 — one batch

    await loader.load_many(["A", "B"])
    print(call_count)  # 1 — served from cache

    loader.clear_all()

    await loader.load_many(["A", "B"])
    print(call_count)  # 2 — cache was cleared, re-fetched

asyncio.run(main())
```

--------------------------------

### Clear DataLoader Cache After Mutation

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Illustrates clearing the cache for a specific key after a data mutation within the same request. This ensures subsequent loads retrieve the updated data.

```python
# Request begins...
user_loader = ...

# And a value happens to be loaded (and cached).
user4 = await user_loader.load(4)

# A mutation occurs, invalidating what might be in cache.
await sql_run('UPDATE users WHERE id=4 SET username="zuck"')
user_loader.clear(4)

# Later the value load is loaded again so the mutated data appears.
user4 = await user_loader.load(4)

# Request completes.
```

--------------------------------

### clear(key)

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Removes a specific key from the memoization cache. Returns self for method chaining. Use this after a mutation to ensure a subsequent load fetches fresh data from the backend.

```APIDOC
## clear(key) — Invalidate a single cache entry

Removes a specific key from the memoization cache. Returns `self` for method chaining. Use this after a mutation to ensure a subsequent load fetches fresh data from the backend.

```python
import asyncio
from aiodataloader import DataLoader


```

--------------------------------

### clear(key)

Source: https://github.com/syrusakbary/aiodataloader/blob/master/README.md

Removes a specific key and its associated value from the cache.

```APIDOC
## clear(key)

### Description
Clears the value at `key` from the cache, if it exists. Returns itself for method chaining.

### Parameters
- *key*: An key value to clear.
```

--------------------------------

### clear(key)

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Invalidates a single entry from the memoization cache. Returns `self` for method chaining. Use this when a specific piece of data has changed and needs to be re-fetched on the next access.

```APIDOC
## `clear(key)` — Invalidate a single cache entry

Clears a specific entry from the memoization cache. Returns `self` for method chaining.

```python
import asyncio
from aiodataloader import DataLoader

db = {4: {"id": 4, "username": "old_name"}}

async def batch_fn(keys):
    return [db.get(k) for k in keys]

async def main():
    loader = DataLoader(batch_fn)

    user = await loader.load(4)
    print(user)  # {'id': 4, 'username': 'old_name'}

    # Simulate a mutation
    db[4] = {"id": 4, "username": "new_name"}

    # Without clear(), the stale cached value would be returned
    loader.clear(4)

    user_fresh = await loader.load(4)
    print(user_fresh)  # {'id': 4, 'username': 'new_name'}

    # clear() returns self — supports chaining
    loader.clear(1).clear(2).clear(3)

asyncio.run(main())
```
```

--------------------------------

### Clear a single cache entry

Source: https://context7.com/syrusakbary/aiodataloader/llms.txt

Invalidate a specific key from the memoization cache to ensure fresh data is fetched on subsequent loads.

```python
import asyncio
from aiodataloader import DataLoader

```
Tokens:
Raw