Skip to content

Cache

cache

Classes

Empty

Sentinel type representing an empty/missing value.

Used to distinguish between None (a valid value) and a truly missing value. Always evaluates to False in boolean context.

Example 
async def get_value(key: str) -> str | Empty:
    if key not in data:
        return EMPTY
    return data[key]


value = await get_value("missing")
if value is EMPTY:
    print("Value not found")

Placeholder

Sentinel type representing a placeholder value.

Used to mark positions where a value will be provided later. Always evaluates to True in boolean context.

CacheMiss

Sentinel type representing a cache miss.

Used to distinguish between cached None values and keys that don't exist. Always evaluates to False in boolean context.

Negative

Sentinel type representing a negative cache entry.

Used to mark keys that are known to be absent, preventing cache penetration. Always evaluates to False in boolean context.

KeyBuilder 

KeyBuilder(split_char: str = ':', prefix: str = replace(' ', '_'))

Utility class for building cache keys with a consistent format.

Provides methods to construct namespaced cache keys and validate key formats.

Attributes:

Name Type Description
split_char

Character used to split parts of the key.
prefix

Prefix to prepend to all keys.

Parameters:

Name Type Description Default
split_char str

Separator character for key parts. Defaults to ":".

 ':'
prefix str

Namespace prefix for all keys. Defaults to normalized project title.

 replace(' ', '_')
Example 
builder = KeyBuilder(prefix="myapp")

# Build keys
user_key = builder.build("user", "123")  # "myapp:user:123"
session_key = builder.build(
    "session", "abc"
)  # "myapp:session:abc"

# Validate keys
builder.validate("myapp:user:123")  # True
builder.validate("other:user:123")  # False
Source code in audex/lib/cache/__init__.py 
181
182
183
184
185
186
187
def __init__(
    self,
    split_char: str = ":",
    prefix: str = __title__.lower().replace(" ", "_"),
) -> None:
    self.split_char = split_char
    self.prefix = prefix
Functions
build 
build(*parts: str) -> str

Build a cache key by joining the prefix and parts.

Parameters:

Name Type Description Default
*parts str

Parts to include in the key.

 ()

Returns:

Type Description
str

The constructed cache key.
Source code in audex/lib/cache/__init__.py 
189
190
191
192
193
194
195
196
197
198
def build(self, *parts: str) -> str:
    """Build a cache key by joining the prefix and parts.

    Args:
        *parts: Parts to include in the key.

    Returns:
        The constructed cache key.
    """
    return self.split_char.join((self.prefix, *parts))
validate 
validate(key: str) -> bool

Validate if a given key starts with the defined prefix.

Parameters:

Name Type Description Default
key str

The cache key to validate.

 required

Returns:

Type Description
bool

True if the key starts with the prefix, False otherwise.
Source code in audex/lib/cache/__init__.py 
200
201
202
203
204
205
206
207
208
209
def validate(self, key: str) -> bool:
    """Validate if a given key starts with the defined prefix.

    Args:
        key: The cache key to validate.

    Returns:
        True if the key starts with the prefix, False otherwise.
    """
    return key.startswith(self.prefix + self.split_char)

KVCache 

KVCache()

Bases: LoggingMixin, AsyncContextMixin, ABC

Abstract base class for async key-value cache implementations.

Provides an async dictionary-like interface for caching with support for TTL, atomic operations (incr/decr), and key validation. Implementations can use in-memory storage, Redis, or other backends.

The cache supports: - Async dict operations (get, set, delete, etc.) - Time-to-live (TTL) for automatic expiration - Atomic increment/decrement operations - Key namespace management via KeyBuilder

Example 
# Using a cache implementation
cache = await make_cache(config, logger)

# Basic operations
await cache.set("user:123", {"name": "Alice"})
user = await cache.get("user:123")

# With TTL
await cache.setx("session:abc", {"data": "..."}, ttl=3600)

# Atomic operations
await cache.incr("counter:visits")
await cache.decr("counter:remaining", amount=5)
Source code in audex/helper/mixin.py 
64
65
def __init__(self) -> None:
    self.logger = loguru.logger.bind(tag=self.__logtag__)
Attributes
key_builder abstractmethod property 
key_builder: KeyBuilder

Get the KeyBuilder instance used for constructing cache keys.

Functions
get_item abstractmethod async 
get_item(key: str) -> VT | Empty | Negative

Retrieve an item from the cache by key.

Source code in audex/lib/cache/__init__.py 
256
257
258
@abc.abstractmethod
async def get_item(self, key: str) -> VT | Empty | Negative:
    """Retrieve an item from the cache by key."""
set_item abstractmethod async 
set_item(key: str, value: VT) -> None

Set an item in the cache with the specified key and value.

Source code in audex/lib/cache/__init__.py 
260
261
262
@abc.abstractmethod
async def set_item(self, key: str, value: VT) -> None:
    """Set an item in the cache with the specified key and value."""
del_item abstractmethod async 
del_item(key: str) -> None

Delete an item from the cache by key.

Source code in audex/lib/cache/__init__.py 
264
265
266
@abc.abstractmethod
async def del_item(self, key: str) -> None:
    """Delete an item from the cache by key."""
iter_keys abstractmethod 
iter_keys() -> AsyncIterator[str]

Return an async iterator over the keys in the cache.

Source code in audex/lib/cache/__init__.py 
268
269
270
@abc.abstractmethod
def iter_keys(self) -> t.AsyncIterator[str]:
    """Return an async iterator over the keys in the cache."""
len abstractmethod async 
len() -> int

Return the number of items in the cache.

Source code in audex/lib/cache/__init__.py 
272
273
274
@abc.abstractmethod
async def len(self) -> int:
    """Return the number of items in the cache."""
contains abstractmethod async 
contains(key: str) -> bool

Check if the cache contains a specific key.

Source code in audex/lib/cache/__init__.py 
276
277
278
@abc.abstractmethod
async def contains(self, key: str) -> bool:
    """Check if the cache contains a specific key."""
get abstractmethod async 
get(key: str, /, default: VT | T | None = None) -> VT | T | None

Get an item from the cache, returning default if the key does not exist.

Source code in audex/lib/cache/__init__.py 
280
281
282
283
@abc.abstractmethod
async def get(self, key: str, /, default: VT | T | None = None) -> VT | T | None:
    """Get an item from the cache, returning default if the key does
    not exist."""
setdefault abstractmethod async 
setdefault(key: str, default: VT | None = None) -> VT | None

Set a default value for a key if it does not exist in the cache.

Source code in audex/lib/cache/__init__.py 
285
286
287
288
@abc.abstractmethod
async def setdefault(self, key: str, default: VT | None = None, /) -> VT | None:
    """Set a default value for a key if it does not exist in the
    cache."""
clear abstractmethod async 
clear() -> None

Clear all items from the cache.

Source code in audex/lib/cache/__init__.py 
290
291
292
@abc.abstractmethod
async def clear(self) -> None:
    """Clear all items from the cache."""
pop abstractmethod async 
pop(key: str, /, default: VT | T | None = None) -> VT | T | None

Remove and return an item from the cache by key.

Source code in audex/lib/cache/__init__.py 
294
295
296
@abc.abstractmethod
async def pop(self, key: str, /, default: VT | T | None = None) -> VT | T | None:
    """Remove and return an item from the cache by key."""
popitem abstractmethod async 
popitem() -> tuple[str, VT]

Remove and return an arbitrary (key, value) pair from the cache.

Source code in audex/lib/cache/__init__.py 
298
299
300
301
@abc.abstractmethod
async def popitem(self) -> tuple[str, VT]:
    """Remove and return an arbitrary (key, value) pair from the
    cache."""
set abstractmethod async 
set(key: str, value: VT) -> None

Set an item in the cache with the specified key and value.

Source code in audex/lib/cache/__init__.py 
303
304
305
@abc.abstractmethod
async def set(self, key: str, value: VT) -> None:
    """Set an item in the cache with the specified key and value."""
setx abstractmethod async 
setx(key: str, value: VT, ttl: int | None = None) -> None

Set an item in the cache with the specified key, value, and optional TTL.

Parameters:

Name Type Description Default
key str

Cache key.

 required
value VT

Value to store.

 required
ttl int | None

Time-to-live in seconds. None means use default TTL.

 None
Source code in audex/lib/cache/__init__.py 
307
308
309
310
311
312
313
314
315
316
@abc.abstractmethod
async def setx(self, key: str, value: VT, ttl: int | None = None) -> None:
    """Set an item in the cache with the specified key, value, and
    optional TTL.

    Args:
        key: Cache key.
        value: Value to store.
        ttl: Time-to-live in seconds. None means use default TTL.
    """
set_negative abstractmethod async 
set_negative(key: str) -> None

Store cache miss marker to prevent cache penetration.

Parameters:

Name Type Description Default
key str

Cache key.

 required
Source code in audex/lib/cache/__init__.py 
318
319
320
321
322
323
324
@abc.abstractmethod
async def set_negative(self, key: str, /) -> None:
    """Store cache miss marker to prevent cache penetration.

    Args:
        key: Cache key.
    """
ttl abstractmethod async 
ttl(key: str) -> int | None

Get the time-to-live (TTL) for a specific key in the cache.

Parameters:

Name Type Description Default
key str

Cache key.

 required

Returns:

Type Description
int | None

Remaining TTL in seconds, or None if no TTL is set.
Source code in audex/lib/cache/__init__.py 
326
327
328
329
330
331
332
333
334
335
@abc.abstractmethod
async def ttl(self, key: str) -> int | None:
    """Get the time-to-live (TTL) for a specific key in the cache.

    Args:
        key: Cache key.

    Returns:
        Remaining TTL in seconds, or None if no TTL is set.
    """
expire abstractmethod async 
expire(key: str, ttl: int | None = None) -> None

Set the time-to-live (TTL) for a specific key in the cache.

Parameters:

Name Type Description Default
key str

Cache key.

 required
ttl int | None

Time-to-live in seconds. None means use default TTL.

 None
Source code in audex/lib/cache/__init__.py 
337
338
339
340
341
342
343
344
@abc.abstractmethod
async def expire(self, key: str, ttl: int | None = None) -> None:
    """Set the time-to-live (TTL) for a specific key in the cache.

    Args:
        key: Cache key.
        ttl: Time-to-live in seconds. None means use default TTL.
    """
incr abstractmethod async 
incr(key: str, amount: int = 1) -> int

Increment the integer value of a key by the given amount.

Parameters:

Name Type Description Default
key str

Cache key.

 required
amount int

Amount to increment by. Defaults to 1.

 1

Returns:

Type Description
int

The new value after incrementing.
Source code in audex/lib/cache/__init__.py 
346
347
348
349
350
351
352
353
354
355
356
@abc.abstractmethod
async def incr(self, key: str, amount: int = 1) -> int:
    """Increment the integer value of a key by the given amount.

    Args:
        key: Cache key.
        amount: Amount to increment by. Defaults to 1.

    Returns:
        The new value after incrementing.
    """
decr abstractmethod async 
decr(key: str, amount: int = 1) -> int

Decrement the integer value of a key by the given amount.

Parameters:

Name Type Description Default
key str

Cache key.

 required
amount int

Amount to decrement by. Defaults to 1.

 1

Returns:

Type Description
int

The new value after decrementing.
Source code in audex/lib/cache/__init__.py 
358
359
360
361
362
363
364
365
366
367
368
@abc.abstractmethod
async def decr(self, key: str, amount: int = 1) -> int:
    """Decrement the integer value of a key by the given amount.

    Args:
        key: Cache key.
        amount: Amount to decrement by. Defaults to 1.

    Returns:
        The new value after decrementing.
    """
keys abstractmethod async 
keys() -> list[str]

Return a list of cache keys.

Source code in audex/lib/cache/__init__.py 
370
371
372
@abc.abstractmethod
async def keys(self) -> list[str]:
    """Return a list of cache keys."""
values abstractmethod async 
values() -> list[VT]

Return all cache values.

Source code in audex/lib/cache/__init__.py 
374
375
376
@abc.abstractmethod
async def values(self) -> list[VT]:
    """Return all cache values."""
items abstractmethod async 
items() -> list[tuple[str, VT]]

Return all cache items as (key, value) pairs.

Source code in audex/lib/cache/__init__.py 
378
379
380
@abc.abstractmethod
async def items(self) -> list[tuple[str, VT]]:
    """Return all cache items as (key, value) pairs."""

options: show_root_heading: true show_source: true heading_level: 2 members_order: source show_signature_annotations: true separate_signature: true