storage-verse-avneesh

Async storage router across Redis, Upstash, PostgreSQL, and future backends.

Backends are grouped by category, not forced into one shape: redis and upstash live under backends/cache/ sharing a get/set interface; postgres lives under backends/database/ with its own connection-pool-and-query API, since SQL access doesn't fit a key-value shape. Each backend has its own folder and its own self-contained code.

Unlike an LLM call (one shape: prompt in, text out), storage operations are heterogeneous — get, set, incr, delete all return different native types. So instead of one Router.get_response(...) wrapping every call in a uniform envelope, this library gives you a small, cached client per backend that returns plain Python types and raises typed exceptions on failure.

Install

pip install "storage-verse-avneesh[redis]"     # Redis only
pip install "storage-verse-avneesh[upstash]"   # Upstash only
pip install "storage-verse-avneesh[database]"  # PostgreSQL (asyncpg) only
pip install "storage-verse-avneesh[all]"       # all three

Usage

import asyncio
from storage_verse_avneesh import get_store

async def main():
    store = get_store("redis", url="redis://localhost:6379/0")

    await store.set("greeting", "hello", ttl_seconds=60)
    print(await store.get("greeting"))     # "hello"
    print(await store.incr("visits"))      # 1
    await store.delete("greeting")

asyncio.run(main())

For Upstash: get_store("upstash", url="...", token="...") (both from the Upstash console).

PostgreSQL (a database, not a cache, backend)

postgres doesn't implement get/set — it's a connection-pool manager with a query API, since SQL access needs queries and transactions, not key-value operations. Configuration comes from environment variables (ENVIRONMENT, DATABASE_URL or DB_HOST/DB_PORT/DB_USER/ DB_PASSWORD/DB_NAME), read lazily the first time a pool is actually created — not from get_store() kwargs:

import asyncio
from storage_verse_avneesh import get_store

async def main():
    db = get_store("postgres")   # no config kwargs

    row = await db.fetchrow("SELECT * FROM users WHERE id = $1", 1)
    count = await db.fetchval("SELECT count(*) FROM users")
    await db.execute("UPDATE users SET last_seen = now() WHERE id = $1", 1)

    async with db.transaction() as conn:
        await conn.execute("UPDATE accounts SET balance = balance - $1 WHERE id = $2", 100, 1)
        await conn.execute("UPDATE accounts SET balance = balance + $1 WHERE id = $2", 100, 2)
        # commits on clean exit, rolls back automatically if either line raises

asyncio.run(main())

execute/fetch/fetchrow/fetchval each pull a connection from the managed pool automatically and raise StorageOperationError on a Postgres failure (matching how the cache backends report failures). Need something these don't cover (LISTEN/NOTIFY, prepared statements, cursors)? Drop to pool = await db.get_pool() (or await db.wait_for_connection_pool() for retry-with-backoff) for direct asyncpg access.

The manager handles pool health checks (dead pools are transparently recreated), Postgres date/time codec registration, and clean shutdown via await db.close().

Connection reuse

Call get_store(name, **config) from anywhere in your codebase — you don't need to construct a client once and pass it around manually. The first call for a given (name, config) pair constructs the backend (and its connection pool); every later call with the same name and config, from any module, returns that exact same cached instance instead of reconnecting:

# file_a.py
store = get_store("redis", url=REDIS_URL)

# file_b.py — same instance as file_a.py, no new connection made
store = get_store("redis", url=REDIS_URL)

Each backend's underlying client also pools connections internally (redis-py's async client, Upstash's HTTP client), so concurrent calls through the same cached instance don't open a new connection per operation either — reuse happens at both the instance level (this library's cache) and the transport level (the backend's own client).

Call await close_all() once, on app shutdown, to close every cached backend and clear the cache.

Discovering backends

import storage_verse_avneesh as sv

sv.help()                    # documents list_backends(), backend_info(), get_store()
sv.list_backends()           # [{"name": "redis", "display_name": "Redis", "category": "cache"}, ...]
sv.backend_info("upstash")   # what it needs to construct, and what it supports

backend_info(name) tells you what a backend needs (e.g. redis needs url; upstash needs url and token) and what it supports — e.g. upstash has no pub/sub or multi-command transactions, since its REST protocol has no persistent connection for either. Raises BackendNotFoundError for an unknown name.

Registered backends

The CacheBackend protocol

Every key-value backend implements the same structural interface (protocols.py):

async def get(self, key: str) -> Optional[str]: ...
async def set(self, key: str, value: str, ttl_seconds: Optional[int] = None) -> bool: ...
async def delete(self, key: str) -> int: ...
async def exists(self, key: str) -> bool: ...
async def expire(self, key: str, ttl_seconds: int) -> bool: ...
async def incr(self, key: str, amount: int = 1) -> int: ...
async def ping(self) -> bool: ...
async def close(self) -> None: ...

This is why redis and upstash are genuinely interchangeable for key-value use — code written against CacheBackend works with either. postgres deliberately does not implement this protocol — it's a database-category backend with its own shape (see above). Future non-key-value backends (document stores, etc.) will get their own protocol too, rather than being forced into CacheBackend.

Exceptions

All exceptions inherit from StorageError:

• BackendNotFoundError — name isn't registered.
• StorageConnectionError — a backend couldn't be reached (connect/ping failure).
• StorageOperationError — a specific operation (get/set/...) failed.

Development

pip install -e ".[dev]"
pytest