Examples & Recipes

1# SPDX-License-Identifier: Apache-2.0
2# Copyright 2026 Hemanshu Vaidya
3"""session_store.py - a realistic session-store example.
4
5Shows: TTL-based session expiry, reactive logging of session lifecycle
6events (useful for auditing/security monitoring), and atomic() for a
7per-session request counter that's safe even if a session is hit by
8concurrent requests.
9
10Run: python examples/session_store.py
11"""
12import time
13
14from nimbuskv import NimbusKV
15
16SESSION_TTL_SECONDS = 60 * 30  # 30 minutes
17
18
19def create_session(store: NimbusKV, user_id: str) -> str:
20    session_id = f"sess_{user_id}_{int(time.time())}"
21    store.set(
22        f"session:{session_id}",
23        {"user_id": user_id, "created_at": time.time(), "requests": 0},
24        ttl=SESSION_TTL_SECONDS,
25    )
26    return session_id
27
28
29def touch_session(store: NimbusKV, session_id: str) -> None:
30    """Bump the request counter for a session -- safe under concurrent
31    requests for the same session, no lock held."""
32    key = f"session:{session_id}"
33
34    def _bump(session):
35        if session is None:
36            return session  # already expired/gone -- nothing to bump
37        session = dict(session)
38        session["requests"] += 1
39        return session
40
41    store.atomic(key, _bump, default=None)
42
43
44def main():
45    with NimbusKV() as store:
46        # Log every session lifecycle event -- this is the kind of thing
47        # you'd wire into real audit logging or a security dashboard.
48        def audit_log(key, event, old, new):
49            if event == "expire":
50                print(f"[audit] session expired: {key}")
51            elif event == "set" and old is None:
52                print(f"[audit] session created: {key}")
53
54        store.subscribe(audit_log, key=None)
55
56        session_id = create_session(store, user_id="alvin")
57        print("Created session:", session_id)
58
59        for _ in range(5):
60            touch_session(store, session_id)
61
62        session = store.get(f"session:{session_id}")
63        print("Session state:", session)
64        assert session["requests"] == 5
65
66        # Simulate a short-lived demo session to show expiry firing --
67        # not something you'd normally do with a real 30-minute TTL.
68        demo_id = "demo_session"
69        store.set(f"session:{demo_id}", {"user_id": "demo"}, ttl=1)
70        print("Waiting for demo session to expire...")
71        time.sleep(1.5)
72        print("Demo session after expiry:", store.get(f"session:{demo_id}"))
73
74
75if __name__ == "__main__":
76    main()

1# SPDX-License-Identifier: Apache-2.0
2# Copyright 2026 Hemanshu Vaidya
3"""worker_pool.py - producer/consumer coordination without polling.
4
5This is the example that actually demonstrates why NimbusKV's reactivity
6matters, not just its caching: multiple worker threads pick up jobs and
7report results, a dispatcher waits on completion with wait_for() instead
8of a sleep-and-poll loop, and atomic() coordinates a shared "jobs done"
9counter safely across threads.
10
11Run: python examples/worker_pool.py
12"""
13import threading
14import time
15
16from nimbuskv import NimbusKV
17
18
19def worker(store: NimbusKV, worker_id: int, job_queue: list, lock: threading.Lock):
20    """A worker thread that pulls jobs off a shared list and reports
21    results into the store. (The job queue itself uses a plain lock
22    here for simplicity -- NimbusKV isn't a queue, it's the shared
23    state / coordination layer around one.)
24    """
25    while True:
26        with lock:
27            if not job_queue:
28                return
29            job_id = job_queue.pop()
30
31        time.sleep(0.1)  # simulate work
32        result = job_id * job_id
33
34        store.set(f"job:{job_id}:result", result)
35        store.atomic("jobs_completed", lambda v: (v or 0) + 1)
36        print(f"[worker {worker_id}] finished job {job_id} -> {result}")
37
38
39def main():
40    with NimbusKV() as store:
41        store.set("jobs_completed", 0)
42
43        n_jobs = 8
44        job_queue = list(range(1, n_jobs + 1))
45        queue_lock = threading.Lock()
46
47        workers = [
48            threading.Thread(target=worker, args=(store, i, job_queue, queue_lock))
49            for i in range(3)
50        ]
51        for w in workers:
52            w.start()
53
54        # The dispatcher doesn't poll or sleep-loop to know when all jobs
55        # are done -- it blocks on wait_for(), which is driven by the
56        # subscription system and unblocks the instant the condition is
57        # actually met.
58        print("Dispatcher waiting for all jobs to complete...")
59        store.wait_for("jobs_completed", lambda v: v == n_jobs, timeout=30)
60        print("All jobs completed.")
61
62        for w in workers:
63            w.join()
64
65        results = {job_id: store.get(f"job:{job_id}:result") for job_id in range(1, n_jobs + 1)}
66        print("Results:", results)
67
68
69if __name__ == "__main__":
70    main()

1from nimbuskv import NimbusKV
2import threading
3
4store = NimbusKV()
5store.set("hit_count", 0)
6
7def increment_hits():
8    for _ in range(1000):
9        store.atomic("hit_count", lambda current: (current or 0) + 1)
10
11threads = [threading.Thread(target=increment_hits) for _ in range(10)]
12for t in threads: t.start()
13for t in threads: t.join()
14
15print("Final Hits:", store.get("hit_count"))  # Exactly 10000

1import asyncio
2from nimbuskv import NimbusKV
3
4async def main():
5    # SQLite persistence: in-memory live reads + async background write-through
6    with NimbusKV(backend="sqlite:cache.db") as store:
7        await store.aset("user:101", {"name": "Alice", "role": "admin"}, ttl=300)
8        user = await store.aget("user:101")
9        print(f"Loaded: {user['name']} (TTL active)")
10
11asyncio.run(main())

1from nimbuskv import NimbusKV
2
3with NimbusKV() as store:
4    # Readers will either see both new values or neither, never partial
5    store.mset({
6        "order:402:status": "COMPLETED",
7        "order:402:paid_at": "2026-10-11T09:00:00Z",
8        "inventory:item_9:stock": 42
9    })
10    
11    results = store.mget(["order:402:status", "inventory:item_9:stock"])
12    print(results)  # {'order:402:status': 'COMPLETED', 'inventory:item_9:stock': 42}