fastfoundry-redis 1.0.1

Redis connection building blocks for the fastfoundry ecosystem.

fastfoundry-redis

Redis connection building blocks for the fastfoundry ecosystem.

Part of the fastfoundry.* namespace (fastfoundry-settings, fastfoundry-postgres, …).

An async redis-py connection pool and client lifecycle for a single Redis server, and a configuration model that plugs into fastfoundry.settings.

Install

pip install fastfoundry-redis
# or
uv add fastfoundry-redis

Requires Python 3.12+.

Configuration

RedisCfg holds the connection parameters. It is designed to nest inside a fastfoundry.settings.BaseAppSettings subclass, so the values load from the environment with the rest of your service configuration, then get handed to redis_pool_ctx:

from fastfoundry.settings import BaseAppSettings
from fastfoundry.redis import RedisCfg


class Settings(BaseAppSettings):
    redis: RedisCfg


settings = Settings()

With the FF_ prefix and __ nested delimiter that BaseAppSettings applies, the fields are read from the environment as FF_REDIS__HOST, FF_REDIS__PORT, FF_REDIS__DB, and so on. password is optional (omit it for a server without auth).

Connection pool & client lifecycle

Open redis_pool_ctx(settings.redis) once for the application lifetime — it builds the connection pool, publishes it for ambient access (get_pool() and redis_cfg()), and disconnects it on exit. For FastAPI, wire it into the lifespan:

from contextlib import asynccontextmanager
from collections.abc import AsyncGenerator

from fastapi import FastAPI
from fastfoundry.redis import redis_pool_ctx

from app.settings import settings


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
    async with redis_pool_ctx(settings.redis):
        yield


app = FastAPI(lifespan=lifespan)

Acquire a client with redis_client; it draws a connection from the pool and returns it on exit. Responses are decoded to str by default (decode_responses=True):

from fastfoundry.redis import redis_client


async with redis_client() as rc:
    await rc.set("key", "value")
    value = await rc.get("key")

Development

mise run install     # uv sync --dev
mise run lint        # ruff check
mise run typecheck   # mypy --strict
mise run test        # pytest
mise run build       # uv build (sdist + wheel)

License

MIT