Greyhorse Redis library

Greyhorse framework library for Redis support.

The primary API is the pieces, not a ready-made module:

An application lists the ones it needs on its own Module, alongside pieces from any other storage library. RedisSyncModule/RedisAsyncModule bundle the trio for the common single-storage case -- sugar, not the primary surface. See examples/.

Usage

Every snippet below is a runnable program, checked against a live Redis. The longer, commented versions live in examples/ and are executed by the test suite, so they cannot rot silently.

One engine, one client

RedisSyncModule is the ready-made bundle for the single-storage case. The config reaches the engine's constructor through the same args={Type: value} door every greyhorse.strand resource uses -- there is no redis-specific wiring.

from greyhorse.run import wrap_sync
from greyhorse.strand import running

from greyhorse_redis import EngineConf, RedisSyncClientCtx, RedisSyncModule


def main() -> None:
    conf = EngineConf(dsn='redis://localhost:6379/0')

    with running(RedisSyncModule, args={EngineConf: conf}) as module:
        client_ctx = module.get(RedisSyncClientCtx).unwrap()
        with client_ctx as client:
            client.set('greeting', 'hello')
            print(client.get('greeting'))


wrap_sync(main)

The engine starts when the module starts and stops when it stops; client is an ordinary redis.Redis borrowed for the length of the with block.

Async

Same declaration, async twin of each piece. running() stays a plain sync context manager even here -- it starts a module, it is not an I/O operation.

from greyhorse.run import run
from greyhorse.strand import running

from greyhorse_redis import EngineConf, RedisAsyncClientCtx, RedisAsyncModule


async def main() -> None:
    conf = EngineConf(dsn='redis://localhost:6379/0')

    with running(RedisAsyncModule, args={EngineConf: conf}) as module:
        client_ctx = module.get(RedisAsyncClientCtx).unwrap()
        async with client_ctx as client:
            await client.set('greeting', 'hello')
            print(await client.get('greeting'))


run(main)

Pipelines

A client is Shared: every command takes effect the moment the server sees it, so there is nothing to commit. A pipeline is Mut -- commands buffer while the window is open and apply() sends MULTI/EXEC.

pipe_ctx = module.get(RedisSyncPipelineCtx).unwrap()

with pipe_ctx as pipe:
    pipe.set('a', '1')
    pipe.set('b', '2')
    pipe_ctx.apply()          # both writes land here, or neither

A forgotten apply() and an exception inside the block both discard the buffer without sending anything. That is the whole reason a pipeline is Mut and a client is not.

Sharing one pipeline across repositories (joined_pipeline())

The Redis twin of greyhorse_sqla's joined_session()/joined_connection() -- same reasoning, one difference worth knowing. Two repositories that each call pipeline() on the same engine inside one request_window() get two independent Pipeline objects and two independent MULTI/EXEC outcomes, because pipeline() builds a brand new context every call and a joined context's default key is that object's own identity. joined_pipeline() keys the join to the ENGINE instead, so a second repository's call over the same engine lands on the root the first one already opened:

notes_pipeline = connections.joined_pipeline   # instead of connections.pipeline

Unlike sqla's session road, a keyless into_joined(connections.pipeline()) called twice on the same task does NOT silently alias one physical object underneath -- AsyncRedisEngine builds a brand new Pipeline from the client on every call, and a Pipeline carries no server-side transaction state of its own to alias. joined_pipeline() exists here for a narrower reason: to give several repositories touching one engine an explicit, engine-keyed root across TASKS, and one shared MULTI/EXEC round trip instead of several independent pipelines each paying for its own.

Who switches this — the composition root, not the repository. A repository is written against the ordinary RedisAsyncPipelineCtx type and has no idea joined_pipeline() exists; JoinedRedisAsyncPipelineCtx is a real subtype of it, so the substitution is invisible there. Only the fragment/factory that builds the repository switches from connections.pipeline to connections.joined_pipeline. The practical shape is the same as sqla's: a factory per parameter (Callable[[], RedisAsyncPipelineCtx]), called fresh on every method, never opened once and reused -- a pipeline still open when a DIFFERENT asyncio.Task enters it raises CompetingBorrowError. See NoteRepository in greyhorse-sqla's tests/e2e/test_web_gateway_over_real_postgres.py for the shape in full; this package's own joined_pipeline() follows the same constructor pattern.

Outside a request_window(), joined_pipeline() is fully transparent: it hands back a plain, independent pipeline(), and nothing about calling it instead of the plain method changes.

A consumer that knows nothing about greyhorse

The point of the split: the class that uses Redis takes a context by TYPE and imports nothing from this package. Only the component says how it is wired.

from typing import ClassVar

from greyhorse.strand import (
    Application, Component, Handle, HttpBinding, HttpGateway, Shared, Use,
)

from greyhorse_redis import (
    EngineConf, RedisSyncClientCtx, RedisSyncConnections, RedisSyncModule,
    SyncRedisEngine,
)


class PingApi:
    def __init__(self, conn: RedisSyncClientCtx) -> None:
        self._conn = conn

    def ping(self) -> bool:
        with self._conn as client:
            return bool(client.ping())


class PingComponent(Component):
    imports: ClassVar = Shared[SyncRedisEngine]
    providers: ClassVar = RedisSyncConnections
    exports: ClassVar = PingApi
    handlers: ClassVar = Handle(PingApi.ping, HttpBinding.Route(verb='GET', path=''))


class App(RedisSyncModule):
    name = 'ping-app'
    components: ClassVar = {'ping': Use(PingComponent)}


app = Application(App, args={EngineConf: EngineConf(dsn='redis://localhost:6379/0')},
                  gateways=(HttpGateway(),))

with app.started() as running_app:
    print(running_app.gateway(HttpGateway).dispatch('GET', '/ping', {}))

imports/providers are the demand side -- the component asks for a window onto the engine and names the provider that turns it into a context. exports/handlers are what it hands back out. Note that an export reaches consumers through the wiring, NOT through module.get(): that door serves the module's own products (the contexts above), not a component's exports.

Subclassing RedisSyncModule fits exactly this shape: one storage, one consumer, same floor. For two independent Redis instances, list the pieces (RedisSyncFragment, RedisSyncBorder, RedisSyncConnections) on your own Module instead -- examples/04_multi_storage.py does that in full.

Configuration

EngineConf is what the engine is built from:

EngineConf(
    dsn='redis://user:password@localhost:6379/0',
    timeout_seconds=5,
    connect_timeout_seconds=5,
    pool_max_connections=4,
    client_name='greyhorse-redis',
    decode_responses=False,
)

rediss:// (TLS) and unix:///var/run/redis.sock are accepted too.

RedisSettings is the environment side. It reads REDIS_* (and .env), and assembles a DSN from the parts when REDIS_DSN is not given:

from greyhorse_redis import EngineConf, RedisSettings

settings = RedisSettings()                    # REDIS_HOST, REDIS_PORT, ...
conf = EngineConf(dsn=settings.dsn)

REDIS_PASSWORD_FILE points at a Docker/Kubernetes secret file and wins over REDIS_PASSWORD. A missing, unreadable or blank file is an error at config time, not a silent fallback to no password.

Credentials. repr(), str() and model_dump_json() come out redacted -- host and user stay visible, the secret does not -- so a config that reaches a log, an f-string or a traceback does not leak. model_dump() and dict(settings) deliberately keep the credential: that is what builds the connection pool. Dumps are for machines; do not log one.

How to build

• Install the project

  uv venv

  uv sync --extra hiredis

  source .venv/bin/activate

• Format code commands

  ruff check --unsafe-fixes --fix

  ruff format

• Run the tests

  Everything except the live-server tests runs with no Redis at all. For the rest, bring one up and point the suite at it:

  docker compose -f tests/docker-compose.yml up -d --wait

  REDIS_TEST_URI=redis://localhost:6380/15 uv run pytest tests -q

  docker compose -f tests/docker-compose.yml down -v

  Host port 6380 is a default, not a fixture: if it is taken on your machine, set REDIS_HOST_PORT and point the suite at the same number.

  REDIS_HOST_PORT=6390 docker compose -f tests/docker-compose.yml up -d --wait

  REDIS_TEST_URI=redis://localhost:6390/15 uv run pytest tests -q