Skip to content

Valkey Client

The Valkey module provides async access to a Valkey (Redis-compatible) server through the official valkey-py async client and integrates with the FastAPI lifespan dependency-injection pattern.

Overview

valkey_lifespan is an async context manager that constructs a valkey.asyncio.Valkey client from settings.Valkey().url and ensures the client is closed on shutdown. The Client type alias wraps the underlying client with fastapi.Depends so it can be injected into route handlers.

Basic Usage

from imbi_common import valkey

async with valkey.valkey_lifespan() as client:
    await client.set('greeting', 'hello')
    value = await client.get('greeting')

FastAPI Dependency Injection

import fastapi

from imbi_common import lifespan, valkey

app = fastapi.FastAPI(
    lifespan=lifespan.Lifespan(valkey.valkey_lifespan),
)


@app.get('/greeting')
async def get_greeting(*, client: valkey.Client) -> dict[str, str | None]:
    value = await client.get('greeting')
    return {'value': value.decode() if value else None}

Configuration

The Valkey URL is read from VALKEY_URL (or the [valkey] section of a config.toml). The default is valkey://localhost:6379/0. Supported schemes are valkey://, valkeys://, and unix://.

API Reference

valkey_lifespan async 

valkey_lifespan() -> abc.AsyncIterator[valkey.Valkey]

Open a Valkey client and ensure it is closed on shutdown.

Source code in src/imbi_common/valkey.py 
22
23
24
25
26
27
28
29
30
31
32
@contextlib.asynccontextmanager
async def valkey_lifespan() -> abc.AsyncIterator[valkey.Valkey]:
    """Open a Valkey client and ensure it is closed on shutdown."""
    global _client
    _client = valkey.Valkey.from_url(str(settings.Valkey().url))
    try:
        yield _client
    finally:
        client, _client = _client, None
        if client is not None:
            await client.aclose()

Client module-attribute 

Client = Annotated[Valkey, Depends(_inject_client)]