Skip to content

Async

_async

AsyncCache 

AsyncCache(
    max_size: int = 100,
    model: type[BaseModel] | None = None,
    name: str = "",
    endpoint_key_is_id: bool = False,
    url_suffix: str = "",
    is_list: bool = False,
)

Bases: BaseCacheState[_KT, _VT, 'PokeLanceAsyncClient'], Generic[_KT, _VT]

Asynchronous cache partition supporting async readiness, non-blocking I/O, and batch loading.

Extends BaseCacheState with an asyncio.Event readiness signal, async JSON serialization via aiofiles, and asynchronous batch network pre-fetching.

Parameters:

Name Type Description Default
max_size int

Maximum number of items to keep in memory before evicting least recently used items.

 100
model type[BaseModel] | None

The PokeLance model class to instantiate when deserializing payloads.

 None
name str

The name of this cache partition (e.g. 'pokemon', 'berry').

 ''
endpoint_key_is_id bool

Whether the primary key for endpoints in this cache is the numeric ID rather than the name.

 False
url_suffix str

URL suffix to append when building endpoint URLs (e.g. '/encounters').

 ""
is_list bool

Whether this cache stores lists of models rather than single model instances.

 False

Examples:

# Wait until the pokemon cache has pre-loaded all endpoint metadata
await client.cache.pokemon.pokemon.wait_until_ready()

# Save cached pokemon to disk asynchronously
await client.cache.pokemon.pokemon.save("./cache_backup")

# Load cached pokemon from disk asynchronously
await client.cache.pokemon.pokemon.load("./cache_backup")
Source code in pokelance/cache/_async/base.py 
def __init__(
    self,
    max_size: int = 100,
    model: type[BaseModel] | None = None,
    name: str = "",
    endpoint_key_is_id: bool = False,
    url_suffix: str = "",
    is_list: bool = False,
) -> None:
    super().__init__(
        max_size=max_size,
        model=model,
        name=name,
        endpoint_key_is_id=endpoint_key_is_id,
        url_suffix=url_suffix,
        is_list=is_list,
    )
    self._ready = asyncio.Event()

is_ready property 

is_ready: bool

Whether this cache partition has finished loading its endpoint identifiers.

wait_until_ready async 

wait_until_ready() -> None

Waits asynchronously until this cache partition has loaded its endpoint identifiers.

Examples:

await client.cache.pokemon.pokemon.wait_until_ready()
Source code in pokelance/cache/_async/base.py 
async def wait_until_ready(self) -> None:
    """Waits asynchronously until this cache partition has loaded its endpoint identifiers.

    Examples
    --------
    ```python
    await client.cache.pokemon.pokemon.wait_until_ready()
    ```
    """
    await self._ready.wait()

set_ready 

set_ready() -> None

Sets this cache partition as ready and unblocks any waiting tasks.

Source code in pokelance/cache/_async/base.py 
@override
def set_ready(self) -> None:
    """Sets this cache partition as ready and unblocks any waiting tasks."""
    super().set_ready()
    self._ready.set()

reset_endpoints 

reset_endpoints() -> None

Clears endpoint registries and resets the readiness event.

Source code in pokelance/cache/_async/base.py 
@override
def reset_endpoints(self) -> None:
    """Clears endpoint registries and resets the readiness event."""
    super().reset_endpoints()
    self._ready.clear()

save async 

save(path: str = '.') -> None

Saves all cached data in this partition to a JSON file asynchronously.

Parameters:

Name Type Description Default
path str

Directory where {name}.json will be saved.

 "."

Examples:

await client.cache.pokemon.pokemon.save("./cache")
Source code in pokelance/cache/_async/base.py 
async def save(self, path: str = ".") -> None:
    """Saves all cached data in this partition to a JSON file asynchronously.

    Parameters
    ----------
    path : str, default: "."
        Directory where `{name}.json` will be saved.

    Examples
    --------
    ```python
    await client.cache.pokemon.pokemon.save("./cache")
    ```
    """
    pathlib.Path(path).mkdir(parents=True, exist_ok=True)
    data = self.serialize()
    async with aiofiles.open(pathlib.Path(f"{path}/{self._name}.json"), "w", encoding="utf-8") as f:
        await f.write("{\n")
        for n, (k, v) in enumerate(data.items()):
            await f.write("    " + f'"{k}": {json.dumps(v, indent=4)}')
            if n != len(data) - 1:
                await f.write(",\n")
        await f.write("\n}")

load async 

load(path: str = '.') -> None

Loads cached data into this partition from a JSON file asynchronously.

Parameters:

Name Type Description Default
path str

Directory containing {name}.json.

 "."

Examples:

await client.cache.pokemon.pokemon.load("./cache")
Source code in pokelance/cache/_async/base.py 
async def load(self, path: str = ".") -> None:
    """Loads cached data into this partition from a JSON file asynchronously.

    Parameters
    ----------
    path : str, default: "."
        Directory containing `{name}.json`.

    Examples
    --------
    ```python
    await client.cache.pokemon.pokemon.load("./cache")
    ```
    """
    async with aiofiles.open(pathlib.Path(f"{path}/{self._name}.json"), encoding="utf-8") as f:
        self.deserialize(json.loads(await f.read()))

load_all async 

load_all() -> None

Fetches and caches all known resources for this endpoint sequentially.

Raises:

Type Description
RuntimeError

If endpoints have not yet been registered or no model class is configured.

Examples:

await client.cache.berry.berry_firmness.load_all()
Source code in pokelance/cache/_async/base.py 
async def load_all(self) -> None:
    """Fetches and caches all known resources for this endpoint sequentially.

    Raises
    ------
    RuntimeError
        If endpoints have not yet been registered or no model class is configured.

    Examples
    --------
    ```python
    await client.cache.berry.berry_firmness.load_all()
    ```
    """
    if not self._endpoints_cached or self._model is None:
        raise RuntimeError("Endpoints not loaded or model not set")
    logger.info(f"Loading {self._name}...")
    self._max_size = len(self._endpoints)
    for endpoint in self._endpoints.values():
        route = Route.from_raw_url(endpoint.url)
        data = self.get(t.cast("_KT", route), None)
        if not data:
            res = await self._client.http.request(route)
            self.setdefault(t.cast("_KT", route), self.from_payload(res))
    logger.info(f"Loaded {self._name}.")

load_all_batch async 

load_all_batch(batch_size: int = 20) -> None

Fetches and caches all known resources for this endpoint in concurrent batches.

Parameters:

Name Type Description Default
batch_size int

The number of parallel HTTP requests to send in each batch.

 20

Raises:

Type Description
RuntimeError

If endpoints have not yet been registered or no model class is configured.

Examples:

await client.cache.berry.berry.load_all_batch(batch_size=15)
Source code in pokelance/cache/_async/base.py 
async def load_all_batch(self, batch_size: int = 20) -> None:
    """Fetches and caches all known resources for this endpoint in concurrent batches.

    Parameters
    ----------
    batch_size : int, default: 20
        The number of parallel HTTP requests to send in each batch.

    Raises
    ------
    RuntimeError
        If endpoints have not yet been registered or no model class is configured.

    Examples
    --------
    ```python
    await client.cache.berry.berry.load_all_batch(batch_size=15)
    ```
    """
    if not self._endpoints_cached or self._model is None:
        raise RuntimeError("Endpoints not loaded or model not set")
    logger.info(f"Loading {self._name}...")
    self._max_size = len(self._endpoints)
    endpoints = list(self._endpoints.values())
    total_endpoints = len(endpoints)
    for i in range(0, total_endpoints, batch_size):
        batch = endpoints[i : i + batch_size]
        tasks = [
            self._fetch_and_cache(t.cast("_KT", Route.from_raw_url(ep.url)))
            for ep in batch
            if not self.get(t.cast("_KT", Route.from_raw_url(ep.url)))
        ]
        if tasks:
            await asyncio.gather(*tasks)
        current_batch = i // batch_size + 1
        total_batches = (total_endpoints + batch_size - 1) // batch_size
        logger.debug(f"Loaded batch {current_batch}/{total_batches} for {self._name}")
    logger.info(f"Loaded {self._name} - {len(self._cache)}/{total_endpoints} items.")

AsyncCacheGroup

Bases: BaseCacheGroup['PokeLanceAsyncClient', AsyncCache[Route, Any]]

Category cache aggregate grouping multiple asynchronous sub-caches.

Examples:

# Wait until all pokemon category sub-caches are ready
await client.cache.pokemon.wait_until_ready()

# Clear all cached pokemon category models
client.cache.pokemon.clear()

wait_until_ready async 

wait_until_ready() -> None

Waits asynchronously until all sub-caches in this group are ready.

Examples:

await client.cache.pokemon.wait_until_ready()
Source code in pokelance/cache/_async/base.py 
async def wait_until_ready(self) -> None:
    """Waits asynchronously until all sub-caches in this group are ready.

    Examples
    --------
    ```python
    await client.cache.pokemon.wait_until_ready()
    ```
    """
    tasks = [cache.wait_until_ready() for cache in self._walk_caches()]
    await asyncio.gather(*tasks)

AsyncCacheManager

Bases: BaseCacheManager['PokeLanceAsyncClient', AsyncCacheGroup]

Top-level asynchronous cache manager.

Coordinates category cache aggregates and provides centralized configuration, cache clearance, readiness synchronization, and aggregated metrics across all sub-caches.

Attributes:

Name Type Description
client PokeLanceAsyncClient

The parent async client instance.
max_size int, default: 100

The maximum number of items allowed in each cache partition.

Examples:

# Check total hits and hit ratio across all endpoints
stats = client.cache.stats
print(f"Total lookups: {stats.total_lookups}, Hit ratio: {stats.hit_ratio:.1%}")

# Set cache capacity globally
client.cache.set_size(200)

# Clear all cached data
client.cache.clear()

wait_until_ready async 

wait_until_ready() -> None

Waits asynchronously until all sub-caches in all aggregates are ready.

Examples:

await client.cache.wait_until_ready()
Source code in pokelance/cache/_async/manager.py 
async def wait_until_ready(self) -> None:
    """Waits asynchronously until all sub-caches in all aggregates are ready.

    Examples
    --------
    ```python
    await client.cache.wait_until_ready()
    ```"""
    tasks = [aggregate.wait_until_ready() for aggregate in self._walk_aggregates()]
    await asyncio.gather(*tasks)

Comments