wallet_helper.ledger module

Content-addressed ledger, so a heavy call runs once and is remembered.

The storage half of wallet-helper. A call is identified by a key built from a namespace plus a payload (the arguments, file, or bytes that determine the result). The ledger stores the result under that key, so an identical call is served from disk instead of running again, across process restarts.

The default store is a directory with one JSON file per entry: local, no server, easy to inspect and to delete. It is content-addressed, so a renamed input file still hits and two different inputs never collide.

Author

Warith HARCHAOUI, https://linkedin.com/in/warith-harchaoui

class wallet_helper.ledger.Ledger(cache_dir=None, max_entries=None)

Bases: object

A directory-backed store of results, one JSON file per entry.

Parameters:

• cache_dir (str or pathlib.Path, optional) – Where entries live. Defaults to $WALLET_HELPER_DIR then ~/.cache/wallet-helper.

• max_entries (int, optional) – A size cap. When set, each put() also evicts down to the newest max_entries entries, so the store cannot grow without bound. None (default) means no automatic bound; call evict() yourself.

Examples

>>> import os_helper as osh
>>> with osh.temporary_folder() as tmp:
...     Ledger(tmp).has("demo_x")
False

clear(namespace=None)

Delete entries, all or just one namespace (irreversible).

With a namespace, only its entries are removed (this is what a memoized function’s cache_clear() calls). Without one, the whole directory is removed and recreated lazily on the next put().

Parameters:

namespace (str | None)

Return type:

None

evict(*, max_entries=None, older_than=None)

Prune entries and return how many were removed.

Expired entries (past their ttl) are always removed. In addition:

Parameters:

• max_entries (int, optional) – Keep only the newest max_entries by creation time; remove the rest. This is a simple size cap.

• older_than (float, optional) – Remove entries created more than this many seconds ago.

Returns:

The number of entries removed.

Return type:

int

get(key)

Return just the stored result for key, or None if absent.

Parameters:

key (str)

Return type:

Any | None

get_record(key)

Return the full stored record, or None if absent.

Parameters:

key (str)

Return type:

dict | None

has(key)

Return True if a result is already stored for key.

Parameters:

key (str)

Return type:

bool

property location: str

The directory that holds the JSON entries (for display).

put(key, result, *, ttl=None)

Store result for key (overwrites any previous entry).

Parameters:

• key (str) – The ledger key (see make_key()).

• result (Any) – The JSON-serialisable result to store.

• ttl (float, optional) – Seconds until the entry is considered stale. None (default) means it never expires. A stale entry is treated as a miss on the next call and is removed by evict().

Return type:

None

register_hit(key)

Count one reuse of the cached result for key (no-op if absent).

The read-modify-write runs under a lock so concurrent hits do not lose a count: threads in this process take turns, and separate processes take turns too where advisory file locks are available (POSIX). On Windows the cross-process count is best-effort; use SqliteLedger for an exact count under heavy multi-process concurrency.

Parameters:

key (str)

Return type:

None

stats(namespace=None)

Count stored entries and their reuses, for the store or one namespace.

Parameters:

namespace (str, optional) – Restrict the count to entries under this namespace. None (default) counts the whole ledger.

Returns:

{"entries": int, "hits": int} where hits is how many times a cached result was reused, that is, how many real calls were saved.

Return type:

dict

class wallet_helper.ledger.LedgerLike(*args, **kwargs)

Bases: Protocol

The storage contract every ledger backend fulfils.

The default Ledger keeps one JSON file per entry, which is ideal for a single process. wallet_helper.sqlite_ledger.SqliteLedger keeps everything in one SQLite file for a shared, concurrency-safe store. A wallet_helper.guard.Wallet and the command-line tools accept either.

clear(namespace=None)

Remove entries, all of them or just one namespace (irreversible).

Parameters:

namespace (str | None)

Return type:

None

evict(*, max_entries=None, older_than=None)

Prune entries and return how many were removed (see Ledger.evict()).

Parameters:

• max_entries (int | None)

• older_than (float | None)

Return type:

int

get(key)

Return the stored result for key, or None if absent.

Parameters:

key (str)

Return type:

Any | None

get_record(key)

Return the full stored record, or None if absent.

Parameters:

key (str)

Return type:

dict | None

has(key)

Return True if a result is already stored for key.

Parameters:

key (str)

Return type:

bool

property location: str

A readable pointer to where entries live (a directory or a file).

put(key, result, *, ttl=None)

Store result for key (overwrites), expiring after ttl seconds.

Parameters:

• key (str)

• result (Any)

• ttl (float | None)

Return type:

None

register_hit(key)

Count one reuse of the cached result for key (no-op if absent).

Parameters:

key (str)

Return type:

None

stats(namespace=None)

Return {entries, hits} for the whole store or one namespace.

Parameters:

namespace (str | None)

Return type:

dict

wallet_helper.ledger.is_fresh(record, *, now=None)

Return True if a record has not expired.

Parameters:

• record (dict) – A stored record, which may carry an expires_at timestamp.

• now (float, optional) – The current time; defaults to time.time(). Passing it lets a caller judge many records against one instant.

Returns:

True when there is no expiry, or the expiry is still in the future.

Return type:

bool

wallet_helper.ledger.make_key(namespace, payload)

Build a ledger key from a namespace and a content payload.

Parameters:

• namespace (str) – A scope for the call, for example "transcribe" or "openai.chat", so unrelated calls do not collide even if their payloads hash alike.

• payload (Any) – See _digest().

Returns:

"<namespace>_<hash>", safe to use as a filename.

Return type:

str

Notes

A str or os.PathLike argument that names an existing file is keyed by the file’s content, by design, so the same file reached by a different path still hits (this applies to a path used as an argument or as a value; a plain str used as a dict key keeps its text). Two consequences worth knowing:

• A string you meant as plain data (a title, an id) that happens to match an existing file in the working directory is content-addressed too. If that is not what you want, wrap the value so it is not a bare path, or exclude the argument with ignore= / a key= builder.

• If a file is deleted in the brief moment between detection and hashing, the key falls back to the path text for that call. The wrapped function would fail to read the missing file anyway, so no stale result is served.

Examples

>>> make_key("demo", {"b": 2, "a": 1}) == make_key("demo", {"a": 1, "b": 2})
True