SNKV Examples

from snkv import KVStore, NotFoundError

with KVStore("mydb.db") as db:
    # Write
    db["user:1"] = "Alice"
    db.put(b"user:2", b"Bob")

    # Read
    print(db["user:1"].decode())       # Alice
    print(db.get("user:2").decode())   # Bob
    print(db.get("missing", b"n/a"))   # b'n/a'

    # Update
    db["user:1"] = "Alice Smith"

    # Delete
    del db["user:1"]

    # Existence
    print("user:2" in db)   # True
    print(db.exists("user:1"))  # False

># Arbitrary binary keys and values
db.put(bytes([0x00, 0x01, 0xFF]), bytes(range(256)))

# In-memory store — no file created, data gone after close
with KVStore(None) as db:
    db["temp"] = "ephemeral"
    print(db["temp"])   # b'ephemeral'

>with KVStore("bank.db") as db:
    # Atomic batch — all writes commit together
    db.begin(write=True)
    try:
        db["account:alice"] = "800"
        db["account:bob"]   = "700"
        db.commit()
    except Exception:
        db.rollback()   # both writes discarded
        raise

    # Large batch — 1000 writes in one transaction
    db.begin(write=True)
    for i in range(1000):
        db[f"item:{i:05d}"] = f"value_{i}"
    db.commit()

>with KVStore("store.db") as db:
    # Create namespaces
    with db.create_column_family("users") as users:
        users["alice"] = "alice@example.com"
        users["bob"]   = "bob@example.com"

    with db.create_column_family("products") as products:
        products["prod:100"] = "Laptop:$999"

    # Same key, different values per CF
    ns_a = db.create_column_family("ns_a")
    ns_b = db.create_column_family("ns_b")
    ns_a["key"] = "from_a"
    ns_b["key"] = "from_b"
    ns_a.close(); ns_b.close()

    # List and drop
    print(db.list_column_families())   # ['users', 'products', 'ns_a', 'ns_b']
    db.drop_column_family("ns_a")
    db.drop_column_family("ns_b")

>with KVStore("store.db") as db:
    # Full scan — for loop uses iterator() under the hood
    for key, value in db:
        print(key.decode(), "->", value.decode())

    # Prefix scan — only keys starting with "user:"
    for key, value in db.prefix_iterator(b"user:"):
        print(key, value)

    # Manual control
    it = db.iterator()
    it.first()
    while not it.eof:
        print(it.key, it.value)
        it.next()
    it.close()

>with KVStore("store.db") as db:
    # Reverse full scan — largest key first
    for key, value in db.reverse_iterator():
        print(key, value)

    # Reverse prefix scan — only "score:" keys, descending
    for key, value in db.reverse_prefix_iterator(b"score:"):
        print(key, value)

    # Unified API — reverse=True on iterator()
    for key, value in db.iterator(reverse=True, prefix=b"user:"):
        print(key, value)

    # Seek in reverse — positions at last key <= target
    with db.iterator(reverse=True) as it:
        it.last()
        it.seek(b"user:m")   # jump to last key <= "user:m"
        while not it.eof:
            print(it.key)
            it.prev()

>from snkv import KVStore, NotFoundError, NO_TTL

with KVStore("store.db") as db:
    # Put with TTL — expires in 60 seconds
    db.put(b"session", b"tok123", ttl=60)
    db[b"token", 30] = b"bearer-xyz"   # dict-style shorthand

    # Inspect remaining TTL
    remaining = db.ttl(b"session")   # e.g. 59.98 (float seconds)
    print(f"{remaining:.2f}s remaining")

    # Lazy expiry — get() returns None if expired
    val = db.get(b"session")   # None if expired, bytes if alive

    # Bulk purge — deletes all expired keys, returns count
    n = db.purge_expired()
    print(f"Removed {n} expired keys")

>class RateLimiter:
    def __init__(self, db, limit, window_s):
        self._cf = db.create_column_family("rl")
        self._limit = limit; self._window = window_s

    def is_allowed(self, user_id):
        key = user_id.encode()
        raw = self._cf.get(key)
        if raw is None:
            self._cf.put(key, b"1", ttl=self._window)
            return True
        count = int(raw)
        if count >= self._limit: return False
        ttl = self._cf.ttl(key) or self._window
        self._cf.put(key, str(count+1).encode(), ttl=ttl)
        return True

rl = RateLimiter(db, limit=5, window_s=60)
print(rl.is_allowed("alice"))   # True (1/5)

>with KVStore("store.db") as db:
    # Forward seek — first key >= target
    with db.iterator() as it:
        it.seek(b"cherry")
        while not it.eof:
            print(it.key); it.next()

    # Reverse seek — last key <= target
    with db.iterator(reverse=True) as it:
        it.last()
        it.seek(b"bbc")   # lands on b"bbb"
        print(it.key)

    # Prefix iterator + seek — boundary enforced
    with db.iterator(prefix=b"user:") as it:
        it.seek(b"user:m")   # skip to "user:m..."
        while not it.eof:
            print(it.key); it.next()

    # Chaining — seek() returns self
    key = db.iterator().seek(b"target").key

>with KVStore("store.db") as db:
    # Absent → inserted=True
    inserted = db.put_if_absent(b"lock", b"owner:alice")
    print(inserted)   # True

    # Present → inserted=False, value unchanged
    inserted = db.put_if_absent(b"lock", b"owner:bob")
    print(inserted)   # False
    print(db.get(b"lock"))   # b'owner:alice'

    # With TTL — auto-releasing lock
    db.put_if_absent(b"job:1", b"worker-A", ttl=30)

    # CF variant — message deduplication
    with db.create_column_family("dedup") as cf:
        if cf.put_if_absent(b"msg:001", b"hello"):
            process()   # only first caller

>with KVStore("store.db") as db:
    # count — O(pages) via sqlite3BtreeCount
    print(db.count())   # e.g. 10000

    # clear — O(pages) bulk truncation; TTL index cleared atomically
    db.clear()
    print(db.count())   # 0

    # Extended stats — 12 counters
    db.put(b"k", b"v"); db.get(b"k")
    st = db.stats()
    print(st["puts"], st["gets"],
          st["bytes_written"], st["db_pages"])

    # Reset cumulative counters; db_pages stays live
    db.stats_reset()
    st = db.stats()
    print(st["puts"])      # 0
    print(st["db_pages"])  # still > 0

>with KVStore("store.db") as db:
    with db.create_column_family("sessions") as cf:
        # Create — expires in 30 minutes
        cf.put(b"sess:abc", b"user_id=42", ttl=1800)

        # Validate
        data = cf.get(b"sess:abc")   # None if expired

        # Idempotent job scheduler via put_if_absent + TTL
        claimed = cf.put_if_absent(b"job:001", b"worker-A", ttl=30)
        if claimed:
            run_job()   # only one worker proceeds

>import multiprocessing
from snkv import KVStore, JOURNAL_WAL

def worker(db_path, worker_id):
    # Each process opens its own connection
    with KVStore(db_path, journal_mode=JOURNAL_WAL,
                busy_timeout=5000) as db:
        for i in range(100):
            db.put(f"w{worker_id}:key:{i}".encode(), b"value")

procs = [multiprocessing.Process(
             target=worker, args=("shared.db", i))
         for i in range(5)]
for p in procs: p.start()
for p in procs: p.join()
# All 500 keys durable on disk