API Reference¶

def build_base_redis_kwargs(settings: RedisSettingsProtocol, *, asyncio: bool = False) -> dict[str, object]:
    """Build base Redis client keyword arguments.

    ``asyncio=True`` builds them for a ``redis.asyncio`` client, whose ``retry`` has to be
    ``redis.asyncio.retry.Retry``.
    """
    if settings.ssl.enabled:
        if settings.ssl.ca_certs:
            _validate_pem_format(settings.ssl.ca_certs, "CERTIFICATE")
        if settings.ssl.certfile:
            _validate_pem_format(settings.ssl.certfile, "CERTIFICATE")
        if settings.ssl.keyfile:
            _validate_pem_format(settings.ssl.keyfile, "PRIVATE KEY")

    kwargs: dict[str, object] = {
        "username": settings.connection.username,
        "password": settings.connection.get_password(),
        "max_connections": settings.pool.max_connections,
        "socket_timeout": settings.pool.socket_timeout,
        "socket_connect_timeout": settings.pool.socket_connect_timeout,
        "socket_keepalive": settings.pool.socket_keepalive,
        "socket_keepalive_options": settings.pool.socket_keepalive_options,
        "health_check_interval": settings.health_check_interval,
        "retry": build_redis_retry(settings, asyncio=asyncio),
        "decode_responses": settings.response.decode_responses,
        "encoding": settings.response.encoding,
        "client_name": settings.connection.client_name,
        "protocol": settings.connection.protocol,
        # SSL
        "ssl": settings.ssl.enabled,
        "ssl_cert_reqs": settings.ssl.cert_reqs,
        "ssl_ca_certs": settings.ssl.ca_certs,
        "ssl_certfile": settings.ssl.certfile,
        "ssl_keyfile": settings.ssl.keyfile,
    }

    if settings.cluster.enabled:
        kwargs["require_full_coverage"] = settings.cluster.require_full_coverage
        kwargs["read_from_replicas"] = settings.cluster.read_from_replicas

    return kwargs

def build_redis_retry(settings: RedisSettingsProtocol, *, asyncio: bool = False) -> Retry | AsyncRetry:
    """Build Redis Retry object from settings.

    Retries are off unless both ``retry.enabled`` and ``retry.max_attempts`` are set. Off is
    still a ``Retry``, with zero retries: handed nothing, redis-py retries on its own (three
    times since 6.0, ten since 8.0, with jittered backoff).

    ``asyncio=True`` builds ``redis.asyncio.retry.Retry``, the one a ``redis.asyncio`` client
    needs: the sync class's ``call_with_retry`` does not await, so on an async client it
    never retries.
    """
    retry_class = AsyncRetry if asyncio else Retry
    if settings.retry.enabled and settings.retry.max_attempts:
        return retry_class(
            backoff=ExponentialBackoff(
                cap=settings.retry.backoff_cap,
                base=settings.retry.backoff_base,
            ),
            retries=settings.retry.max_attempts,
        )
    return retry_class(backoff=NoBackoff(), retries=0)

async def check_async_redis_health(client: AsyncRedisClient, *, write_key: str | None = None) -> bool:
    """Check async Redis connection health.

    Pings the server, and with ``write_key`` writes that key as well. A ping alone is a
    liveness answer: a read-only replica and a server at ``maxmemory`` under ``noeviction``
    both reply PONG and refuse every write. The write probe turns that into a readiness
    answer for a service that needs Redis for more than reads.

    Args:
        client: Redis client (single or cluster) to check
        write_key: Key to ``SET`` after the ping, expiring after ``WRITE_PROBE_TTL_S``
            seconds. None, the default, pings only. Prefix it yourself; this package
            applies no key prefix.

    Returns:
        True if Redis is healthy, False otherwise
    """
    try:
        result: Any = await client.ping()  # type: ignore[misc]

        # Redis Cluster ping returns dict[str, bool] (node_id -> success); single node returns bool.
        if isinstance(result, dict) and result and all(isinstance(k, str) for k in result):
            healthy = all(bool(v) for v in result.values())
        else:
            healthy = bool(result)

        if healthy and write_key is not None:
            healthy = bool(await client.set(write_key, "1", ex=WRITE_PROBE_TTL_S))
    except (RedisError, RedisClusterException, OSError, ConnectionError, TimeoutError) as e:
        logger.warning(
            "Async Redis health check failed",
            extra={"error": str(e), "error_type": type(e).__name__},
        )
        return False
    except (asyncio.CancelledError, KeyboardInterrupt, SystemExit):
        raise
    except Exception:
        # Catch unexpected errors like encoding/decoding or internal redis-py bugs.
        # We return False to indicate unhealthy state but log the full exception for debugging.
        logger.exception("Unexpected error during Redis health check")
        return False
    else:
        return healthy

def check_redis_health(client: SyncRedisClient, *, write_key: str | None = None) -> bool:
    """Check sync Redis connection health.

    Pings the server, and with ``write_key`` writes that key as well. A ping alone is a
    liveness answer: a read-only replica and a server at ``maxmemory`` under ``noeviction``
    both reply PONG and refuse every write. The write probe turns that into a readiness
    answer for a service that needs Redis for more than reads.

    Args:
        client: Redis client (single or cluster) to check
        write_key: Key to ``SET`` after the ping, expiring after ``WRITE_PROBE_TTL_S``
            seconds. None, the default, pings only. Prefix it yourself; this package
            applies no key prefix.

    Returns:
        True if Redis is healthy, False otherwise
    """
    try:
        result: Any = client.ping()  # type: ignore[misc]

        # Redis Cluster ping returns dict[str, bool] (node_id -> success); single node returns bool.
        if isinstance(result, dict) and result and all(isinstance(k, str) for k in result):
            healthy = all(bool(v) for v in result.values())
        else:
            healthy = bool(result)

        if healthy and write_key is not None:
            healthy = bool(client.set(write_key, "1", ex=WRITE_PROBE_TTL_S))
    except (RedisError, RedisClusterException, OSError, ConnectionError, TimeoutError) as e:
        logger.warning(
            "Redis health check failed",
            extra={"error": str(e), "error_type": type(e).__name__},
        )
        return False
    except (KeyboardInterrupt, SystemExit):
        raise
    except Exception:
        # Catch unexpected errors like encoding/decoding or internal redis-py bugs.
        # We return False to indicate unhealthy state but log the full exception for debugging.
        logger.exception("Unexpected error during Redis health check")
        return False
    else:
        return healthy

async def close_async_redis_client(client: AsyncRedisClient) -> None:
    """Close async Redis client and release pool connections.

    Args:
        client: Redis client (single or cluster) to close

    Note:
        Uses asyncio.shield to ensure close operation completes even if task is cancelled.
    """
    try:
        await asyncio.wait_for(asyncio.shield(client.aclose()), timeout=ACLOSE_TIMEOUT_S)
    except TimeoutError:
        logger.warning("Redis client aclose timed out", extra={"timeout_s": ACLOSE_TIMEOUT_S})
    except (asyncio.CancelledError, KeyboardInterrupt):
        raise
    except Exception as e:
        logger.warning("Error closing Redis client", extra={"error": str(e)})
    else:
        logger.info("Closed Redis client")

def close_redis_client(client: SyncRedisClient) -> None:
    """Close sync Redis client and release pool connections.

    Args:
        client: Redis client (single or cluster) to close
    """
    try:
        client.close()
    except (KeyboardInterrupt, SystemExit):
        raise
    except Exception as e:
        logger.warning("Error closing Redis client", extra={"error": str(e)})
    else:
        logger.info("Closed Redis client")

def create_async_redis_client(
    settings: RedisSettingsProtocol, metrics: RedisMetricsProtocol | None = None
) -> AsyncRedisClient:
    """Create async Redis client with connection pool.

    Returns redis-py client (Redis or RedisCluster) with optional metrics instrumentation.
    - If metrics is None: returns plain Redis/RedisCluster (zero overhead)
    - If metrics provided: returns InstrumentedRedis/InstrumentedRedisCluster

    Automatically selects single or cluster mode based on settings.

    Args:
        settings: Redis configuration
        metrics: Optional Prometheus metrics collector (None = no instrumentation)

    Returns:
        Redis, RedisCluster, InstrumentedRedis, or InstrumentedRedisCluster instance
    """
    if settings.cluster.enabled:
        return _create_async_cluster_client(settings, metrics=metrics)

    return _create_async_single_client(settings, metrics=metrics)

def create_redis_client(
    settings: RedisSettingsProtocol, metrics: RedisMetricsProtocol | None = None
) -> SyncRedisClient:
    """Create sync Redis client with connection pool.

    Returns redis-py client (Redis or RedisCluster) with optional metrics instrumentation.
    - If metrics is None: returns plain Redis/RedisCluster (zero overhead)
    - If metrics provided: returns InstrumentedRedis/InstrumentedRedisCluster

    Automatically selects single or cluster mode based on settings.

    Args:
        settings: Redis configuration
        metrics: Optional Prometheus metrics collector (None = no instrumentation)

    Returns:
        Redis, RedisCluster, InstrumentedRedis, or InstrumentedRedisCluster instance
    """
    if settings.cluster.enabled:
        return _create_cluster_client(settings, metrics=metrics)

    return _create_single_client(settings, metrics=metrics)

def parse_redis_url_node(node: str) -> tuple[str, int]:
    """Parse Redis node from host:port, [ipv6]:port or redis:// URL string."""
    normalized_node = node.strip()
    if not normalized_node:
        raise ValueError("Redis node cannot be empty")

    # If already a URL, don't prepend scheme
    url = normalized_node if normalized_node.startswith(("redis://", "rediss://")) else f"redis://{normalized_node}"

    parsed = urlparse(url)
    host = parsed.hostname
    port = parsed.port

    if host is None or port is None:
        raise ValueError(f"Invalid Redis node: {node!r}")

    # Validate port range
    if not (1 <= port <= 65535):
        raise ValueError(f"Invalid Redis port {port}: must be between 1 and 65535")

    return host, port