orionis.cache.stores.file.FileCacheBackend

THE ORIONIS API

Build with clarity.

Explore the building blocks of an async-first Python framework. Every module, class, and method — connected, searchable, and ready to build with.

Explore modules View classes

class documentation

class FileCacheBackend: (source)

Constructor: FileCacheBackend(path)

View In Hierarchy

Undocumented

Static Method	`__readEntryUnlocked`	Read a cache entry while its file lock is held.
Static Method	`__tempPath`	Return a unique staging path for file.
Method	`__createSync`	Write entry to file only when the file does not exist yet.
Method	`__file`	Return the filesystem path for key.
Method	`__incrementSync`	Update a counter while holding its file lock.
Method	`__init__`	Initialize the backend and create the cache directory if needed.
Method	`__lock`	Select the stable cross-process lock stripe for a cache file.
Method	`__readSync`	Read and decode a cache file synchronously.
Method	`__replaceIfPresent`	Replace a live entry while holding its cross-process lock.
Method	`__replaceSync`	Publish tmp as file, retrying a transient sharing violation.
Method	`__unlinkSync`	Remove file, ignoring missing-file errors.
Method	`__writeEntryUnlocked`	Publish a cache entry while its file lock is held.
Method	`__writeSync`	Atomically write entry to file via tmp-then-rename.
Async Method	`add`	Store value under key only if the key does not already exist.
Async Method	`clear`	Remove all cache entries from the store directory.
Async Method	`delete`	Remove key from the store.
Async Method	`exists`	Return True if key exists and has not expired.
Async Method	`get`	Return the cached value for key, or default when absent/expired.
Async Method	`increment`	Increment the integer stored at key by delta.
Async Method	`multiGet`	Return a list of values for keys in the same order.
Async Method	`multiSet`	Store multiple key/value pairs with an optional TTL.
Async Method	`replace`	Replace an existing live key without recreating a deleted entry.
Async Method	`set`	Store value under key with an optional TTL in seconds.
Class Variable	`__slots__`	Undocumented
Instance Variable	`_counter_lock`	Undocumented
Instance Variable	`_locks`	Undocumented
Instance Variable	`_path`	Undocumented
Instance Variable	`_rename_lock`	Undocumented
Property	`lockNamespace`	Return the canonical directory used to identify cache locks.

@staticmethod

def __readEntryUnlocked(file: Path) -> dict | None: (source) ¶

Read a cache entry while its file lock is held.

Parameters
file:`Path`	Path to the cache entry.

Returns
`dict \| None`	Live decoded entry, or None when missing, expired, or invalid.

@staticmethod

def __tempPath(file: Path) -> Path: (source) ¶

Return a unique staging path for file.

Parameters
file:`Path`	Target cache file.

Returns
`Path`	Sibling path ending in `.tmp` that no concurrent writer of the same key can collide with.

def __createSync(self, file: Path, entry: dict) -> bool: (source) ¶

Write entry to file only when the file does not exist yet.

Parameters
file:`Path`	Target cache file.
entry:`dict`	Serializable cache entry.

Returns
`bool`	`True` when the file was created by this call, `False` when another writer got there first.

def __file(self, key: str) -> Path: (source) ¶

Return the filesystem path for key.

Parameters
key:`str`	Cache key.

Returns
`Path`	Path to the corresponding `.json` file.

def __incrementSync(self, file: Path, delta: int) -> int: (source) ¶

Update a counter while holding its file lock.

Parameters
file:`Path`	Counter file to update.
delta:`int`	Amount to add to the current value.

Returns
`int`	Updated counter value.

def __init__(self, path: Path): (source) ¶

Initialize the backend and create the cache directory if needed.

Parameters
path:`Path`	Directory where cache files will be stored.

Returns
`None`	Complete the documented operation without returning a value.

def __lock(self, file: Path) -> BaseFileLock: (source) ¶

Select the stable cross-process lock stripe for a cache file.

Parameters
file:`Path`	Cache file protected by the lock.

Returns
`BaseFileLock`	Platform lock shared by writes, replacement and deletion.

def __readSync(self, file: Path) -> dict | None: (source) ¶

Read and decode a cache file synchronously.

Parameters
file:`Path`	Path to the cache file.

Returns
`dict \| None`	Decoded entry, or `None` on any error.

def __replaceIfPresent(self, file: Path, entry: dict) -> bool: (source) ¶

Replace a live entry while holding its cross-process lock.

Parameters
file:`Path`	Target cache file.
entry:`dict`	Encoded replacement entry.

Returns
`bool`	False when the previous entry has disappeared or expired.

def __replaceSync(self, tmp: Path, file: Path): (source) ¶

Publish tmp as file, retrying a transient sharing violation.

Parameters
tmp:`Path`	Staging file holding the encoded entry.
file:`Path`	Destination cache file.

Returns
`None`	Complete the documented operation without returning a value.

Raises
`OSError`	If the rename keeps failing after the last attempt.

def __unlinkSync(self, file: Path) -> int: (source) ¶

Remove file, ignoring missing-file errors.

Parameters
file:`Path`	File to remove.

Returns
`int`	One when a file was removed, otherwise zero.

def __writeEntryUnlocked(self, file: Path, entry: dict): (source) ¶

Publish a cache entry while its file lock is held.

Parameters
file:`Path`	Destination cache file.
entry:`dict`	Encodable cache entry.

Returns
`None`	Complete the documented operation without returning a value.

Raises
`OSError`	If staging or publishing the entry fails.

def __writeSync(self, file: Path, entry: dict): (source) ¶

Atomically write entry to file via tmp-then-rename.

Parameters
file:`Path`	Target cache file.
entry:`dict`	Serializable cache entry.

Returns
`None`	Complete the documented operation without returning a value.

Raises
`OSError`	If the entry cannot be staged or renamed into place.

async def add(self, key: str, value: Any, ttl: float | None = None) -> bool: (source) ¶

Store value under key only if the key does not already exist.

The entry is created with an exclusive open, so concurrent callers never both succeed and the operation is safe as a mutual-exclusion primitive.

Parameters
key:`str`	Cache key.
value:`Any`	Value to store.
ttl:`float \| None`	Optional TTL in seconds.

Returns
`bool`	True on success.

Raises
`ValueError`	If the key already exists and has not expired.

async def clear(self) -> bool: (source) ¶

Remove all cache entries from the store directory.

Returns
`bool`	Always True.

async def delete(self, key: str) -> int: (source) ¶

Remove key from the store.

Parameters
key:`str`	Cache key.

Returns
`int`	1 if the key existed, 0 otherwise.

async def exists(self, key: str) -> bool: (source) ¶

Return True if key exists and has not expired.

Parameters
key:`str`	Cache key.

Returns
`bool`	Undocumented

async def get(self, key: str, default: Any = None) -> Any: (source) ¶

Return the cached value for key, or default when absent/expired.

Expired entries are deleted on first read (lazy eviction).

Parameters
key:`str`	Cache key.
default:`Any`	Value returned when the key is not found or has expired.

Returns
`Any`	Stored value or default.

async def increment(self, key: str, delta: int = 1) -> int: (source) ¶

Increment the integer stored at key by delta.

Creates the key with value delta if it does not exist, and keeps the expiry of an existing entry untouched.

Parameters
key:`str`	Cache key.
delta:`int`	Amount to add (use negative values to decrement).

Returns
`int`	New value after increment.

async def multiGet(self, keys: list[str], default: Any = None) -> list[Any]: (source) ¶

Return a list of values for keys in the same order.

Parameters
keys:`list[str]`	Cache keys.
default:`Any`	Returned for each missing/expired key.

Returns
`list[Any]`	Undocumented

async def multiSet(self, pairs: list[tuple[str, Any]], ttl: float | None = None) -> bool: (source) ¶

Store multiple key/value pairs with an optional TTL.

Parameters
pairs:`list[tuple[str, Any]]`	Sequence of (key, value) pairs.
ttl:`float \| None`	Shared TTL applied to every pair.

Returns
`bool`	Always True.

async def replace(self, key: str, value: Any, ttl: float | None = None) -> bool: (source) ¶

Replace an existing live key without recreating a deleted entry.

Parameters
key:`str`	Existing cache key.
value:`Any`	Replacement value.
ttl:`float \| None`	Replacement lifetime in seconds.

Returns
`bool`	Whether the entry was replaced atomically.

async def set(self, key: str, value: Any, ttl: float | None = None) -> bool: (source) ¶

Store value under key with an optional TTL in seconds.

Parameters
key:`str`	Cache key.
value:`Any`	JSON-serializable value to cache.
ttl:`float \| None`	Time-to-live in seconds. None means no expiry.

Returns
`bool`	Always True.

__slots__: tuple[str, ...] = (source) ¶

Undocumented

_counter_lock = (source) ¶

Undocumented

_locks = (source) ¶

Undocumented

_path: Path = (source) ¶

Undocumented

_rename_lock = (source) ¶

Undocumented

@property

lockNamespace: Path = (source) ¶

Return the canonical directory used to identify cache locks.

Returns
`Path`	Absolute directory shared by backends storing the same files.