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.connect() 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.

Example: wiring this into a project

A minimal FastAPI project using this library end to end:

myproject/
  .env
  app/
    __init__.py
    db.py               <- the ONE file that talks to storage-verse-avneesh
    main.py               <- wires startup/shutdown
    services/
      user_service.py     <- uses the DB, never calls get_store() directly
    routers/
      users.py             <- uses the service, never touches the DB directly

.env — you own this file; the library never reads it directly, only os.environ after something loads it:

ENVIRONMENT=local_environment
DATABASE_URL=postgresql://user:pass@host/dbname?sslmode=require

app/db.py — the only place in the project that calls get_store():

import os
from storage_verse_avneesh import get_store

def get_db():
    """
    Postgres reads its own config from env vars, so this is a thin
    passthrough - but every caller going through this one function
    guarantees they all hit the exact same get_store() cache entry.
    """
    return get_store("postgres")

def get_cache():
    """
    Here it DOES matter that this is the only place url= is passed - two
    files independently calling get_store("redis", url=...) with even a
    slightly different value would silently get two different pools.
    """
    return get_store("redis", url=os.environ["REDIS_URL"])

app/main.py — loads .env, connects eagerly at startup, closes at shutdown:

from contextlib import asynccontextmanager
from dotenv import load_dotenv
from fastapi import FastAPI

load_dotenv()   # populates os.environ - this line is the entire env setup step

from storage_verse_avneesh import close_all
from app.db import get_db
from app.routers import users

@asynccontextmanager
async def lifespan(app: FastAPI):
    db = get_db()
    await db.connect()   # retries with backoff; fails fast if the DB is unreachable
    yield
    await close_all()      # closes postgres, redis, everything cached

app = FastAPI(lifespan=lifespan)
app.include_router(users.router)

app/services/user_service.py — a different file, deep in the app, never received db as an argument:

from app.db import get_db

async def get_user_by_id(user_id: int):
    db = get_db()
    return await db.fetchrow("SELECT * FROM users WHERE id = $1", user_id)

async def create_user(name: str, email: str):
    db = get_db()
    async with db.transaction() as conn:
        await conn.execute("INSERT INTO users (name, email) VALUES ($1, $2)", name, email)

app/routers/users.py:

from fastapi import APIRouter
from app.services import user_service

router = APIRouter()

@router.get("/users/{user_id}")
async def read_user(user_id: int):
    user = await user_service.get_user_by_id(user_id)
    return dict(user) if user else {"error": "not found"}

user_service.py and main.py never share a variable — every get_db() call, from any file at any depth, hits the same get_store() cache entry and returns the identical pooled connection manager. No dependency injection required (though Depends(get_db) composes fine on top if you want it swappable in tests).

Two things this doesn't do: the cache is per-process, not per-cluster — uvicorn --workers 4 gives each worker its own pool, which is what you want; and it's safe under asyncio but not guaranteed thread-safe if you call get_store() concurrently from separate OS threads rather than asyncio tasks.

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