wallet_helper package

Submodules

Module contents

wallet-helper: never run the same heavy call twice.

Persistent, content-addressed memoization for expensive calls (a paid API request, a slow model, any heavy function). An identical call is served from a local ledger instead of running again, across process restarts, and two identical calls made at the same time collapse into one (single-flight), so the second waits for the first and reuses its result.

The front door is Wallet and the memoize() decorator. Storage is a Ledger (a folder of JSON files) or a SqliteLedger (one shared, concurrency-safe file, with a cross-process lease).

Author

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

class wallet_helper.Ledger(cache_dir=None, max_entries=None)[source]

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)[source]

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)[source]

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)[source]

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

Parameters:: key (str)
Return type:: Any | None

get_record(key)[source]

Return the full stored record, or None if absent.

Parameters:: key (str)
Return type:: dict | None

has(key)[source]

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)[source]

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)[source]

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)[source]

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.LedgerLike(*args, **kwargs)[source]

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)[source]

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

Parameters:: namespace (str | None)
Return type:: None

evict(*, max_entries=None, older_than=None)[source]

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)[source]

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

Parameters:: key (str)
Return type:: Any | None

get_record(key)[source]

Return the full stored record, or None if absent.

Parameters:: key (str)
Return type:: dict | None

has(key)[source]

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)[source]

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

Parameters:

key (str)
result (Any)
ttl (float | None)

Return type:

None

register_hit(key)[source]

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

Parameters:: key (str)
Return type:: None

stats(namespace=None)[source]

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

Parameters:: namespace (str | None)
Return type:: dict

class wallet_helper.RemoteLedger(base_url, *, timeout=30.0, request=None)[source]

Bases: object

A ledger backed by a wallet-helper HTTP server.

Parameters:

base_url (str) – Root URL of the server, for example "http://127.0.0.1:8000". A trailing slash is fine; it is trimmed.
timeout (float, optional) – Per-request timeout in seconds. Defaults to 30.
request (callable, optional) – A custom transport request(method, path, body) -> dict | None used instead of the built-in urllib one. Mainly for tests, where it can route to a FastAPI TestClient.

claim(key, lease_seconds=300.0)[source]

Get the cached result, or lease the right to compute it (see the server).

Parameters:

key (str)
lease_seconds (float)

Return type:

dict

clear(namespace=None)[source]

Delete results on the server, all of them or just one namespace.

Parameters:: namespace (str | None)
Return type:: None

evict(*, max_entries=None, older_than=None)[source]

Prune results on the server and return how many were removed.

Parameters:

max_entries (int | None)
older_than (float | None)

Return type:

int

extend(key, token=None)[source]

Renew a lease on the server for a long-running job.

Parameters:

key (str)
token (str | None)

Return type:

bool

get(key)[source]

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

Parameters:: key (str)
Return type:: Any | None

get_record(key)[source]

Return a partial record {"key", "result"} for key, or None.

Parameters:: key (str)
Return type:: dict | None

has(key)[source]

Return True if the server has a result stored for key.

Parameters:: key (str)
Return type:: bool

property location: str: The server URL that backs this ledger (for display).

put(key, result, *, ttl=None)[source]

Store result for key on the server (an alias for submit).

Parameters:

key (str)
result (Any)
ttl (float | None)

Return type:

None

register_hit(key)[source]

No-op: the server counts reuses itself, on claim hits and result reads.

Parameters:: key (str)
Return type:: None

release(key, token=None)[source]

Drop a lease on the server so a waiter can take over (your own, if fenced).

Parameters:

key (str)
token (str | None)

Return type:

None

stats(namespace=None)[source]

Return {entries, hits} from the server (namespace filter optional).

Parameters:: namespace (str | None)
Return type:: dict

submit(key, result, *, token=None, ttl=None)[source]

Store a leader’s result on the server and release its own lease.

Parameters:

key (str)
result (Any)
token (str | None)
ttl (float | None)

Return type:

dict

class wallet_helper.SqliteLedger(db_path=None, max_entries=None)[source]

Bases: object

A single-file SQLite store of results, with an in-flight lease table.

Parameters:

db_path (str or pathlib.Path, optional) – The database file. Defaults to <default ledger dir>/ledger.db (the same base as Ledger, honouring $WALLET_HELPER_DIR). Parent directories are created if missing.
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.

Examples

>>> import os_helper as osh
>>> with osh.temporary_folder() as tmp:
...     lg = SqliteLedger(tmp + "/ledger.db")
...     lg.put("demo_x", {"ok": True})
...     lg.get("demo_x")
{'ok': True}

claim(key, lease_seconds=300.0)[source]

Get the cached result, or lease the right to compute it.

Parameters:

key (str) – The ledger key (see wallet_helper.ledger.make_key()).
lease_seconds (float, optional) – How long a lease is honoured before it counts as abandoned, so a crashed leader cannot block waiters forever. Defaults to 300 s.

Returns:

{"status": "hit", "result": ...} if it is already computed, {"status": "leased", "token": ...} if you are the leader (compute, then submit() with the token), or {"status": "pending"} if another caller is computing it (wait and claim again).

Return type:

dict

clear(namespace=None)[source]

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

The database file itself remains; only rows are removed.

Parameters:: namespace (str | None)
Return type:: None

evict(*, max_entries=None, older_than=None)[source]

Prune entries and return how many were removed.

Expired entries are always removed. With older_than, entries created more than that many seconds ago go too. With max_entries, only the newest max_entries by creation time are kept.

Parameters:

max_entries (int | None)
older_than (float | None)

Return type:

int

extend(key, token=None)[source]

Renew a lease so a long job is not treated as abandoned.

A leader running longer than lease_seconds calls this (directly or through heartbeat()) to reset the lease clock. With token the renewal only applies to a lease you still own, so a revived stale leader cannot extend the lease a new leader now holds.

Returns:

True if a lease you may renew existed and was renewed.

Return type:

bool

Parameters:

key (str)
token (str | None)

get(key)[source]

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

Parameters:: key (str)
Return type:: Any | None

get_record(key)[source]

Return the full stored record, or None if absent.

Parameters:: key (str)
Return type:: dict | None

has(key)[source]

Return True if a result is already stored for key.

Parameters:: key (str)
Return type:: bool

heartbeat(key, token=None, *, interval=60.0)[source]

Renew key’s lease every interval seconds for the duration of a block.

Pass the token from claim() so the renewal is fenced to the lease you hold. Wrap a long computation in this so its lease never lapses:

>>> import os_helper as osh
>>> with osh.temporary_folder() as tmp:
...     lg = SqliteLedger(tmp + "/ledger.db")
...     lease = lg.claim("job_1")
...     with lg.heartbeat("job_1", lease["token"], interval=0.05):
...         result = 6 * 7            # a long job, kept alive meanwhile
...     _ = lg.submit("job_1", result, token=lease["token"])
...     lg.get("job_1")
42

Parameters:

key (str)
token (str | None)
interval (float)

Return type:

Iterator[None]

property location: str: The database file that holds the entries (for display).

put(key, result, *, ttl=None)[source]

Store result for key (overwrites, resetting the hit counter).

With ttl set, the entry expires that many seconds from now and is then treated as a miss on the next claim() and removed by evict().

Parameters:

key (str)
result (Any)
ttl (float | None)

Return type:

None

register_hit(key)[source]

Atomically count one reuse of key (no-op if absent).

The hits = hits + 1 runs as a single statement, so concurrent reuses never lose a count. This is the reason to prefer this backend over the read-modify-write of the JSON one under real concurrency.

Parameters:: key (str)
Return type:: None

release(key, token=None)[source]

Drop a lease so a waiter can take over (only your own, if token is given).

Parameters:

key (str)
token (str | None)

Return type:

None

stats(namespace=None)[source]

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

Parameters:: namespace (str | None)
Return type:: dict

submit(key, result, *, token=None, ttl=None)[source]

Store a leader’s result and release its own lease; return the record.

The result is stored unconditionally (it is deterministic, so a late submit is harmless), but only a lease held by token is released, so a revived stale leader cannot drop a new leader’s lease. Passing token fences the release: duplicates coalesce as long as a leader finishes within its lease or keeps a heartbeat (a leader that silently overruns its lease can still be run twice, as with any time-based lease).

Parameters:

key (str)
result (Any)
token (str | None)
ttl (float | None)

Return type:

dict

class wallet_helper.Wallet(ledger=None, poll_interval=0.05)[source]

Bases: object

A ledger plus single-flight around heavy calls.

Parameters:

ledger (wallet_helper.ledger.LedgerLike, optional) – The result store, any backend satisfying LedgerLike. With the default JSON Ledger, single-flight is in-process. With a claim-capable backend (SqliteLedger or RemoteLedger) it is cross-process. A default Ledger is created when omitted.
poll_interval (float, optional) – How often a waiter re-checks a claim-based backend while another caller computes a key. Defaults to 0.05 s.

async acall(namespace, key_payload, fn, *, ttl=None, stale_while_revalidate=False)[source]

Async twin of call(): await fn once and cache its result.

This is what memoizes an async def correctly: it awaits the coroutine and stores the result, never the coroutine object. Concurrent identical awaits coalesce (async single-flight), and a claim-capable backend still gives cross-process dedup (its blocking calls run in a worker thread).

Parameters mirror call(), except fn is a zero-argument callable returning an awaitable.

Parameters:

namespace (str)
key_payload (Any)
fn (Callable[[], Awaitable[Any]])
ttl (float | None)
stale_while_revalidate (bool)

Return type:

tuple[Any, bool]

call(namespace, key_payload, fn, *, ttl=None, stale_while_revalidate=False)[source]

Return a cached result, or run fn once and store it.

Parameters:

namespace (str) – Scope for the call, for example the provider or endpoint name.
key_payload (Any) – What determines the result: arguments, a file path, or bytes. Hashed to the ledger key (see wallet_helper.ledger.make_key()).
fn (callable) – Zero-argument callable doing the heavy work, run at most once per key even under concurrency.
ttl (float, optional) – Seconds the stored result stays fresh. After it expires the next call recomputes. None (default) means it never expires.
stale_while_revalidate (bool, optional) – Only for the in-process Ledger. When an entry is expired, return the stale result at once and refresh it in a background thread, so callers never wait on the recompute.

Returns:

The result and from_cache: True when served from the ledger, False when this call did the real work. A raising fn stores nothing, so a failed call is never cached.

Return type:

tuple of (Any, bool)

paid(namespace, *, key=None, ignore=(), ttl=None, stale_while_revalidate=False)[source]

Decorator memoizing a function through this wallet.

Parameters:

namespace (str) – Scope for the call.
key (callable, optional) – key(*args, **kwargs) returning the payload that identifies the result. Overrides the default (all args and kwargs).
ignore (tuple of str, optional) – Parameter names to exclude from the cache key, the tidy alternative to a key= lambda when you just need to drop a volatile handle (for example ignore=("client",)).
ttl (float, optional) – Seconds the stored result stays fresh (see call()).
stale_while_revalidate (bool, optional) – Serve a stale result and refresh in the background (in-process Ledger only; see call()).

Returns:

The wrapped function (repeat identical calls are free). It carries .cache_info() (this namespace’s {entries, hits}) and .cache_clear() (drop this namespace’s entries), like functools.lru_cache.

Return type:

callable

Examples

>>> import os_helper as osh
>>> from wallet_helper.ledger import Ledger
>>> with osh.temporary_folder() as tmp:
...     w = Wallet(Ledger(tmp))
...     @w.paid("square")
...     def square(n):
...         return n * n
...     square(9), square(9)                  # second is free
...     square.cache_info()["entries"]        # one entry stored
(81, 81)
1

wallet_helper.default_wallet()[source]

Return the process-wide default Wallet, created on first use.

It uses the default Ledger location ($WALLET_HELPER_DIR then ~/.cache/wallet-helper). Assign wallet_helper.guard._default_wallet yourself to point it elsewhere.

Return type:: Wallet

wallet_helper.make_key(namespace, payload)[source]

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

wallet_helper.memoize(fn=None, *, namespace=None, key=None, ignore=(), ttl=None, stale_while_revalidate=False, wallet=None)[source]

Persistent memoization: a cache that survives restarts, plus single-flight.

Drop it on any function and its results are content-addressed to disk, reused across process restarts, and shared between concurrent callers so the same heavy call never runs twice. Works bare (@memoize) or configured (@memoize(ttl=3600, ignore=("client",))).

Parameters:

fn (callable, optional) – The function, when used bare as @memoize (filled in by Python).
namespace (str, optional) – Cache scope; defaults to the function’s module.qualname so distinct functions never collide.
key (callable, optional) – Custom key builder key(*args, **kwargs).
ignore (tuple of str, optional) – Parameter names to exclude from the key.
ttl (float, optional) – Seconds the stored result stays fresh.
stale_while_revalidate (bool, optional) – Serve a stale result and refresh in the background (in-process store only).
wallet (Wallet, optional) – The wallet to use; defaults to the shared default_wallet().

Returns:

The memoized function, carrying .cache_info() and .cache_clear().

Return type:

callable

Examples

>>> import os_helper as osh
>>> from wallet_helper.ledger import Ledger
>>> with osh.temporary_folder() as tmp:
...     w = Wallet(Ledger(tmp))
...     @memoize(wallet=w)
...     def double(n):
...         return n * 2
...     double(21), double(21)
(42, 42)