wallet_helper.sqlite_ledger module

SQLite-backed ledger: one file, concurrency-safe, with a cross-process lease.

An interchangeable backend for wallet_helper.ledger.Ledger. The default keeps one JSON file per entry, which is great for a single process. This backend keeps everything in one SQLite file with write-ahead logging, so many processes or hosts can share one store and update it without clobbering each other’s reuse counters.

It also adds a lease table (claim / submit / release) used for cross-process single-flight: while one caller computes a key, others see it as pending and wait, so the same heavy call is not run twice at the same time.

sqlite3 ships with Python, so this stays dependency-light: the shared, concurrency-safe store without running a database server.

Author

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

class wallet_helper.sqlite_ledger.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:
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:
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:
Return type:

dict