NimbusKV Recipes & Quickstart

Verified production patterns for caching, atomic counters, worker coordination, reactive session stores, and FastAPI integration.

5-Minute Quickstart

From installation to production concurrency in 7 progressive steps.

Quickstart

Step 1Installation

Install the core NimbusKV library from PyPI. Zero compulsory external dependencies.

pip install nimbuskv

Step 2Import & Initialize Store

Instantiate your reactive, lock-free store. Thread-safe across all CPU cores.

from nimbuskv import NimbusKV

# Create an in-memory store instance
store = NimbusKV()

Step 3Basic Reads, Writes & Existence

Readers never lock or block writers; reads achieve millions of ops/sec.

store.set("app:mode", "production")
store.set("user:42", {"name": "Priya", "role": "admin"})

print("Mode:", store.get("app:mode"))
print("Has user?", store.exists("user:42"))

Step 4Generational TTL & Auto-Expiry

Generational priority queue guarantees accurate cleanup without memory leaks.

# Key automatically evicts after 10 seconds
store.set("session:token_xyz", {"uid": 42}, ttl=10)

# Background priority queue manages cleanup without memory leaks

Step 5Atomic Compound Updates

Single deterministic mutex prevents lost updates and CAS retry storms.

store.set("counter:hits", 0)

# Read-modify-write executed safely under the write mutex
new_count = store.atomic("counter:hits", lambda c: (c or 0) + 1)
print("Updated count:", new_count)

Step 6Reactive Subscriptions

Copy-on-write event notifications dispatched without holding write locks.

def on_state_change(key, action, old_val, new_val):
    print(f"Key '{key}' was {action}: {old_val} -> {new_val}")

store.subscribe("system:status", on_state_change)
store.set("system:status", "online")

Step 7Worker Thread Coordination (wait_for)

Condition-variable coordination without wasteful polling loops.

# Wait up to 5 seconds until key meets the condition
is_ready = store.wait_for("job:101", lambda val: val == "COMPLETED", timeout=5.0)
if is_ready:
    print("Worker finished task successfully!")

Practical Production Recipes

Production-ready code patterns verified against NimbusKV v3.1.

1. Caching Repeated Computations (Function Memoization)

Performance

Problem: Expensive database queries or CPU-heavy calculations executed repeatedly with the same arguments.

API Reference

import time
from nimbuskv import NimbusKV

cache = NimbusKV()

def get_user_profile(user_id: int):
    cache_key = f"user_profile:{user_id}"
    cached = cache.get(cache_key)
    if cached is not None:
        return cached

    # Simulate expensive computation / DB query
    time.sleep(0.05)
    data = {"id": user_id, "name": f"User_{user_id}", "status": "active"}

    # Cache result for 60 seconds
    cache.set(cache_key, data, ttl=60)
    return data

# First call: computes & caches
profile1 = get_user_profile(42)
# Second call: instant lock-free cache hit
profile2 = get_user_profile(42)
print("Profile retrieved:", profile2["name"])

Expected Output:

Profile retrieved: User_42

Concurrency Considerations:

Reads are completely lock-free. If two threads simultaneously compute for an uncached key, the last set() wins cleanly without corrupting memory.

2. High-Concurrency Atomic Counter

Concurrency

Problem: Concurrent threads incrementing request counts, metrics, or rate limit buckets losing updates due to race conditions.

API Reference

from concurrent.futures import ThreadPoolExecutor
from nimbuskv import NimbusKV

store = NimbusKV()
store.set("requests:total", 0)

def record_request():
    # store.atomic executes under the write mutex exactly once
    store.atomic("requests:total", lambda c: (c or 0) + 1)

# Simulate 20 concurrent threads executing 50 increments each (1,000 total)
with ThreadPoolExecutor(max_workers=20) as executor:
    list(executor.map(lambda _: record_request(), range(1000)))

final_count = store.get("requests:total")
print(f"Total requests recorded: {final_count} (Expected: 1000)")

Expected Output:

Total requests recorded: 1000 (Expected: 1000)

Concurrency Considerations:

In v3.1.0+, atomic() uses a deterministic short mutex instead of CAS retries. Every call runs its mutation exactly once without wasting CPU.

3. Worker Pool Task Coordination with wait_for()

Coordination

Problem: Worker threads waiting for an asynchronous background job to finish without burning CPU in a sleep/poll loop.

API Reference

import threading
import time
from nimbuskv import NimbusKV

store = NimbusKV()
store.set("pipeline:stage_1", "RUNNING")

def background_worker():
    time.sleep(0.3)
    # Mark task as completed
    store.set("pipeline:stage_1", "COMPLETED")

threading.Thread(target=background_worker, daemon=True).start()

# Consumer thread blocks cleanly until condition is satisfied or timeout expires
ready = store.wait_for("pipeline:stage_1", lambda status: status == "COMPLETED", timeout=2.0)

if ready:
    print("Pipeline stage 1 finished! Proceeding to stage 2.")
else:
    print("Timed out waiting for worker.")

Expected Output:

Pipeline stage 1 finished! Proceeding to stage 2.

Concurrency Considerations:

Uses threading.Condition internally. Zero CPU cycles wasted while waiting; awakens immediately upon publication.

4. Reactive Web Session Store with Auto-Expiry

Web & API

Problem: Managing authenticated session tokens that must expire automatically when idle.

API Reference

import uuid
from nimbuskv import NimbusKV

sessions = NimbusKV()

def create_session(user_id: str, ttl_seconds: int = 1800) -> str:
    token = f"sess_{uuid.uuid4().hex[:16]}"
    sessions.set(token, {"user_id": user_id, "authenticated": True}, ttl=ttl_seconds)
    return token

def authenticate_request(token: str):
    sess = sessions.get(token)
    if not sess:
        return None
    # Sliding window: refresh TTL on activity
    sessions.set(token, sess, ttl=1800)
    return sess["user_id"]

token = create_session("admin_user", ttl_seconds=1800)
user = authenticate_request(token)
print("Authenticated user:", user)

Expected Output:

Authenticated user: admin_user

Concurrency Considerations:

Overwriting a key with a new TTL updates the generational priority queue without leaking stale timer references.

5. Durable SQLite Persistence

Persistence

Problem: Requiring fast in-memory performance while surviving process restarts with durable on-disk backup.

API Reference

# Install optional extra: pip install "nimbuskv[sqlite]"
from nimbuskv import NimbusKV
from nimbuskv.backends.sqlite import SQLiteBackend

# Configure WAL (Write-Ahead-Log) SQLite storage
backend = SQLiteBackend("app_cache.db")
store = NimbusKV(backend=backend)

store.set("config:theme", "dark")
store.set("config:locale", "en_US")

# On next restart with the same file, initial state is restored automatically
print("Config restored:", store.get("config:theme"))

Expected Output:

Config restored: dark

Concurrency Considerations:

Reads hit in-memory HAMT at full speed. Writes commit to the SQLite WAL backend under the write lock.

6. FastAPI Lifespan State Integration

Web & API

Problem: Sharing a single, thread-safe NimbusKV store instance across all asynchronous route handlers in FastAPI.

API Reference

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from nimbuskv import NimbusKV

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize shared store in application state
    app.state.store = NimbusKV()
    yield
    # Cleanup / flush backends if applicable
    app.state.store.close()

app = FastAPI(lifespan=lifespan)

@app.get("/items/{item_id}")
async def read_item(item_id: str, request: Request):
    store = request.app.state.store
    cached = store.get(f"item:{item_id}")
    if cached:
        return {"data": cached, "source": "cache"}

    # Mock lookup
    data = {"id": item_id, "name": f"Item_{item_id}"}
    store.set(f"item:{item_id}", data, ttl=300)
    return {"data": data, "source": "origin"}

Expected Output:

{"data": {"id": "10", "name": "Item_10"}, "source": "origin"}

Concurrency Considerations:

NimbusKV is completely thread-safe and non-blocking for reads, making it ideal for FastAPI worker threads.