orchestrata_ultimate.core.redis_factory

Reference › orchestrata_ultimate › core › redis_factory

The thread-safe Redis client factory for free-threaded Python 3.14t, in core. create_sync_redis builds a client on a BlockingConnectionPool, which is safe for thread checkout, with a default of 25 connections and a 5-second timeout. create_async_redis builds a client on a redis.asyncio connection pool. Both are annotated with @traces.

The host and port come from the environment through resolve_redis_host and resolve_redis_port, and the database index through _get_redis_db (default 0). The host and port follow the no-literal-fallback contract (RUL-003, see infra_env_rule): _require_infra raises RuntimeError when a key is unset or blank, so a misconfigured process cannot silently reach the wrong server. Every pool created is recorded in a module-level list, and atexit handlers (_cleanup_sync_pools, _cleanup_async_pools) disconnect them on process exit.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

Truncated at 25 nodes: showing 1 of 1 imports and 23 of 37 importers.

API

Constants

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/redis_factory.py

"""Thread-safe Redis client factory for Python 3.14t (free-threaded).

Provides centralized construction of sync and async Redis clients
with proper connection pooling for GIL-free environments.

Sync clients use BlockingConnectionPool (thread-safe checkout).
Async clients use redis.asyncio.ConnectionPool (asyncio serializes).

Usage:
    from orchestrata_ultimate.core.redis_factory import (
        create_sync_redis,
        create_async_redis,
    )
    client = create_sync_redis()  # reads from infra_config
    async_client = create_async_redis()
"""

from __future__ import annotations

import atexit
import logging
import os
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, cast

from orchestrata_ultimate.core.base.traces import traces

if TYPE_CHECKING:
    import redis
    import redis.asyncio  # noqa: F401 -- binds redis.asyncio.* for annotations below

_logger = logging.getLogger(__name__)

# Track pools for cleanup
_sync_pools: list[Any] = []
_async_pools: list[Any] = []


def _require_infra(env: Mapping[str, str], key: str) -> str:
    """Return ``env[key]`` or raise — the no-literal-fallback contract (RUL-003)
    for required infra-connection keys. See ``core.infra_env_rule``."""
    value = env.get(key)
    if value is None or not str(value).strip():
        raise RuntimeError(
            f"{key} is a REQUIRED infra endpoint and must be set in the "
            f"environment; no literal fallback is permitted (RUL-003). "
            f"Load an env profile (env/<env>.env) or set {key} explicitly."
        )
    return value


def resolve_redis_host(environ: Mapping[str, str] | None = None) -> str:
    """Redis host from env, RAISING when unset (RUL-003 — no literal default).

    The canonical seam a flagged site should call instead of inlining
    ``os.environ.get("REDIS_HOST", "localhost")``. A default host is forbidden:
    an unset key must fail loudly, never silently connect to localhost or a
    baked-in server.
    """
    env = os.environ if environ is None else environ
    return _require_infra(env, "REDIS_HOST")


def resolve_redis_port(environ: Mapping[str, str] | None = None) -> int:
    """Redis port from env, RAISING when unset (RUL-003 — no literal default)."""
    env = os.environ if environ is None else environ
    return int(_require_infra(env, "REDIS_PORT"))


# Backwards-compatible private aliases (used by create_sync_redis /
# create_async_redis below). They now RAISE on an unset key rather than
# defaulting to localhost:6379 — a client constructed with no host and no env
# fails loudly instead of quietly pointing at localhost.
def _get_redis_host(environ: Mapping[str, str] | None = None) -> str:
    return resolve_redis_host(environ)


def _get_redis_port(environ: Mapping[str, str] | None = None) -> int:
    return resolve_redis_port(environ)


def _get_redis_db(environ: Mapping[str, str] | None = None) -> int:
    """Get Redis DB index from env (default: ``0``).

    REDIS_DB is deliberately NOT under the no-default rule: it is a logical
    database index, not a server endpoint, and ``0`` is the universal default
    (see ``core.infra_env_rule`` — REDIS_DB is excluded from INFRA_ENDPOINT_KEYS).
    """
    env = os.environ if environ is None else environ
    return int(env.get("REDIS_DB", "0"))


@traces(spec="001", story="US-010", reqs=["REQ-001-US-010-012"])
def create_sync_redis(
    host: str | None = None,
    port: int | None = None,
    db: int | None = None,
    max_connections: int = 25,
    timeout: float = 5.0,
    decode_responses: bool = True,
    password: str | None = None,
    env: Mapping[str, str] | None = None,
) -> redis.Redis:
    """Create a thread-safe sync Redis client.

    Uses BlockingConnectionPool which is safe under Python 3.14t
    free-threaded (no GIL). Each thread gets exclusive connection
    checkout via internal threading.Lock.

    Args:
        host: Redis host (default: from REDIS_HOST env).
        port: Redis port (default: from REDIS_PORT env).
        db: Redis database (default: from REDIS_DB env).
        max_connections: Pool size (default: 25).
        timeout: Seconds to wait for connection (default: 5.0).
        decode_responses: Decode bytes to str (default: True).
        password: Redis password (default: None).
        env: Optional environment mapping (defaults to ``os.environ``).
            Exposed so tests can exercise misconfiguration without mutating
            the process environment.

    Returns:
        redis.Redis instance with BlockingConnectionPool.
    """
    resolved_host = host or _get_redis_host(env)
    resolved_port = port or _get_redis_port(env)
    resolved_db = db if db is not None else _get_redis_db(env)

    import redis
    from redis import BlockingConnectionPool

    pool = BlockingConnectionPool(
        host=resolved_host,
        port=resolved_port,
        db=resolved_db,
        password=password,
        decode_responses=decode_responses,
        max_connections=max_connections,
        # redis-py's BlockingConnectionPool.timeout param is unannotated in
        # source; pyright infers "int" solely from the `timeout=20` default
        # literal, but the real implementation stores it as-is and accepts
        # any numeric value (it is only ever compared/used, never int-cast).
        timeout=cast(int, timeout),
    )
    _sync_pools.append(pool)

    client = redis.Redis(connection_pool=pool)
    _logger.debug(
        "Created sync Redis client: %s:%s/%s (pool=%d)",
        resolved_host,
        resolved_port,
        resolved_db,
        max_connections,
    )
    return client


@traces(spec="001", story="US-010", reqs=["REQ-001-US-010-012"])
def create_async_redis(
    host: str | None = None,
    port: int | None = None,
    db: int | None = None,
    decode_responses: bool = True,
    password: str | None = None,
    socket_connect_timeout: float | None = None,
    env: Mapping[str, str] | None = None,
) -> redis.asyncio.Redis:
    """Create an async Redis client.

    Asyncio serializes coroutine execution within a single thread,
    so no additional thread-safety mechanisms are needed.

    Args:
        host: Redis host (default: from REDIS_HOST env).
        port: Redis port (default: from REDIS_PORT env).
        db: Redis database (default: from REDIS_DB env).
        decode_responses: Decode bytes to str (default: True).
        password: Redis password (default: None).
        socket_connect_timeout: Optional connection timeout in seconds.
        env: Optional environment mapping (defaults to ``os.environ``).
            Exposed so tests can exercise misconfiguration without mutating
            the process environment.

    Returns:
        redis.asyncio.Redis instance.
    """
    resolved_host = host or _get_redis_host(env)
    resolved_port = port or _get_redis_port(env)
    resolved_db = db if db is not None else _get_redis_db(env)

    import redis.asyncio as redis

    kwargs: dict[str, Any] = dict(
        host=resolved_host,
        port=resolved_port,
        db=resolved_db,
        password=password,
        decode_responses=decode_responses,
    )
    if socket_connect_timeout is not None:
        kwargs["socket_connect_timeout"] = socket_connect_timeout
    client = redis.Redis(**kwargs)

    _logger.debug(
        "Created async Redis client: %s:%s/%s",
        resolved_host,
        resolved_port,
        resolved_db,
    )
    return client


def _cleanup_sync_pools() -> None:
    """Disconnect all sync connection pools on process exit."""
    for pool in _sync_pools:
        try:
            pool.disconnect()
        except Exception as exc:
            _logger.debug("sync pool cleanup: %s", exc)


def _cleanup_async_pools() -> None:
    """Disconnect all async connection pools on process exit."""
    import asyncio

    for pool in _async_pools:
        try:
            loop = asyncio.get_event_loop()
            if loop.is_running():
                loop.create_task(pool.disconnect())
            else:
                loop.run_until_complete(pool.disconnect())
        except Exception as exc:
            _logger.debug("async pool cleanup: %s", exc)


atexit.register(_cleanup_sync_pools)
atexit.register(_cleanup_async_pools)