kv-store-py 1.0.1

A simple wrapper for key-value stores supporting Python dict, Redis, Valkey, PostgreSQL, Memcached, and SQLite and other backend clients.

kv-store

Easy and practical key value store.

• get, set, get_many, set_many methods with handy defaults
• Implemented as Asynchronous context manager
• Supports distributed locking in all implementations
• Supports multiple value types: strings, numbers, booleans, lists, and dicts (JSON-serializable types)

Installation

pip install kv-store-py

Quick Start

from kv_store_py import AbstractStore, RedisStore, PostgresStore

# Use Redis
store: AbstractStore = await RedisStore.from_url("redis://localhost:6379/0")
await store.set("key", "value", ttl=3600)
await store.get("key")

# Supports multiple value types
await store.set("name", "alice", ttl=3600)          # string
await store.set("count", 42, ttl=3600)              # number
await store.set("active", True, ttl=3600)           # boolean
await store.set("tags", ["a", "b", "c"], ttl=3600)  # list
await store.set("user", {"name": "alice", "age": 30}, ttl=3600)  # dict

# Swap to Postgres - same interface
store: AbstractStore = await PostgresStore.from_url("postgresql://localhost/mydb")
await store.set_many({"k1": "v1", "k2": "v2"}, ttl=3600)
await store.get_many(["k1", "k2"])

# Distributed locking (supported by both Redis and Postgres)
async with store.lock("my-resource", timeout=30):
    # critical section
    pass

API Reference

All store implementations share the same async interface:

Core Operations

# Get a value (returns default if key doesn't exist or is expired)
value = await store.get(key: str, default: Any = None) -> Any

# Set a value with TTL in seconds
success = await store.set(key: str, value: Any, ttl: int) -> bool

# Delete a key
existed = await store.delete(key: str) -> bool

# Get multiple keys at once
results = await store.get_many(
    keys: list[str],
    default: Any = None  # Single value or list of defaults per key
) -> dict[str, Any]

# Set multiple keys atomically
results = await store.set_many(items: dict[str, Any], ttl: int) -> dict[str, bool]

# Delete multiple keys
count = await store.delete_many(keys: list[str]) -> int

Distributed Locking

# Acquire a lock for mutual exclusion
async with store.lock("resource:id", timeout=30, blocking_timeout=10):
    # Critical section - only one process can execute this
    await perform_exclusive_operation()

PostgreSQL Options

When using PostgresStore.from_url(), you can customize the connection pool and table name:

store = await PostgresStore.from_url(
    "postgresql://localhost/mydb",
    table_name="custom_kv",  # Default: "kv_store"
    min_size=2,              # Minimum pool connections
    max_size=10              # Maximum pool connections
)

PostgreSQL uses table-based locking with auto-expiration tokens (not advisory locks).

Supported Backends

Backend	Status	Class
In-Memory	Available	InMemoryStore
Redis	Available	RedisStore
PostgreSQL	Available	PostgresStore

Note: PostgreSQL support was tested only on PostgreSQL 16. There is no guarantee it works on other versions.

| Valkey | Planned | - | | SQLite | Planned | - | | MySQL | Planned | - |

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

License

MIT License - see LICENSE for details.