Skip to main content

cache_store

cache_impl_for_policy​

def cache_impl_for_policy(policy: PolicyT) -> CacheStoreImpl[PolicyT]

Get a cache implementation depending on the policy

CacheScopeStore​

from dara.core.internal.cache_store.cache_store import CacheScopeStore

class CacheScopeStore(Generic[PolicyT])

A Cache Scope aware store. Depending on the policy will store a different cache implementation per entry. Keeps entries scoped to the cache scope of current execution.

Methods​

delete​

async def delete(key: str) -> Any

Delete an entry from the cache.

Arguments:

  • key: The key of the entry to delete.

get​

async def get(key: str, unpin: bool = False) -> Optional[Any]

Retrieve an entry from the cache.

Arguments:

  • key: The key of the entry to retrieve.
  • unpin: If true, the entry will be unpinned if it is pinned.

set​

async def set(key: str, value: Any, pin: bool = False)

Add an entry to the cache. Depending on the implementation might evict other entries.

Arguments:

  • key: The key of the entry to set.
  • value: The value of the entry to set.
  • pin: If true, the entry will not be evicted until read.

clear​

async def clear()

Empty the store.

CacheStore​

from dara.core.internal.cache_store.cache_store import CacheStore

class CacheStore()

Key-value store class which stores a separate CacheScopeStore per registry entry.

Methods​

delete​

async def delete(registry_entry: CachedRegistryEntry, key: str) -> Any

Delete an entry from the cache for the given registry entry and cache key.

Arguments:

  • registry_entry: The registry entry to delete the value for.
  • key: The key of the entry to delete.

get​

async def get(registry_entry: CachedRegistryEntry,
              key: str,
              unpin: bool = False) -> Optional[Any]

Retrieve an entry from the cache for the given registry entry and cache key.

Arguments:

  • registry_entry: The registry entry to retrieve the value for.
  • key: The key of the entry to retrieve.
  • unpin: If true, the entry will be unpinned if it is pinned.

get_or_wait​

async def get_or_wait(registry_entry: CachedRegistryEntry, key: str)

Retrieve an entry from the cache for the given registry entry and cache key.

If the entry is a pending value or a pending task, wait for it to resolve.

Arguments:

  • registry_entry: The registry entry to retrieve the value for.
  • key: The key of the entry to retrieve.

set​

async def set(registry_entry: CachedRegistryEntry,
              key: str,
              value: Any,
              error: Optional[Exception] = None,
              pin: bool = False)

Add an entry to the cache for the given registry entry and cache key.

Arguments:

  • registry_entry: The registry entry to store the value for.
  • key: The key of the entry to set.
  • value: The value of the entry to set.
  • error: If set, the value is a PendingValue that will resolve to this error.
  • pin: If true, the entry will not be evicted until read.

set_pending​

async def set_pending(registry_entry: CachedRegistryEntry, key: str)

Set a pending value for the given registry entry and cache key.

Arguments:

  • registry_entry: The registry entry to store the value for.
  • key: The key of the entry to set.

clear​

async def clear()

Empty all stores.