diff --git a/backend/open_webui/internal/config.py b/backend/open_webui/internal/config.py index 294144ae76..46f5b1b67d 100644 --- a/backend/open_webui/internal/config.py +++ b/backend/open_webui/internal/config.py @@ -193,6 +193,13 @@ class AppConfig: super().__setattr__('_entries', {}) super().__setattr__('_key_prefix', redis_key_prefix) + # If sentinels weren't explicitly provided, read from env. + if redis_sentinels is None: + from open_webui.env import REDIS_SENTINEL_HOSTS, REDIS_SENTINEL_PORT + from open_webui.utils.redis import get_sentinels_from_env + + redis_sentinels = get_sentinels_from_env(REDIS_SENTINEL_HOSTS, REDIS_SENTINEL_PORT) + rc: Union[redis.Redis, redis.cluster.RedisCluster, None] = None if redis_url: rc = get_redis_connection(redis_url, redis_sentinels or [], redis_cluster, decode_responses=True) diff --git a/backend/open_webui/main.py b/backend/open_webui/main.py index 7291a40fcd..21ad95e1a6 100644 --- a/backend/open_webui/main.py +++ b/backend/open_webui/main.py @@ -447,10 +447,9 @@ from open_webui.env import ( LICENSE_KEY, LOG_FORMAT, MAX_BODY_LOG_SIZE, + # Redis REDIS_CLUSTER, REDIS_KEY_PREFIX, - REDIS_SENTINEL_HOSTS, - REDIS_SENTINEL_PORT, REDIS_URL, RESET_CONFIG_ON_START, SAFE_MODE, @@ -579,7 +578,7 @@ from open_webui.utils.oauth import ( resolve_oauth_client_info, ) from open_webui.utils.plugin import install_tool_and_function_dependencies -from open_webui.utils.redis import get_redis_connection, get_sentinels_from_env +from open_webui.utils.redis import get_redis_client, get_redis_connection from open_webui.utils.security_headers import SecurityHeadersMiddleware from open_webui.utils.session_pool import get_session from open_webui.utils.tools import set_terminal_servers, set_tool_servers @@ -652,12 +651,7 @@ async def lifespan(app: FastAPI): log.info('Installing external dependencies of functions and tools...') await install_tool_and_function_dependencies() - app.state.redis = get_redis_connection( - redis_url=REDIS_URL, - redis_sentinels=get_sentinels_from_env(REDIS_SENTINEL_HOSTS, REDIS_SENTINEL_PORT), - redis_cluster=REDIS_CLUSTER, - async_mode=True, - ) + app.state.redis = get_redis_client(async_mode=True) if app.state.redis is not None: app.state.redis_task_command_listener = asyncio.create_task(redis_task_command_listener(app)) @@ -764,7 +758,6 @@ app.state.oauth_client_manager = oauth_client_manager app.state.instance_id = None app.state.config = AppConfig( redis_url=REDIS_URL, - redis_sentinels=get_sentinels_from_env(REDIS_SENTINEL_HOSTS, REDIS_SENTINEL_PORT), redis_cluster=REDIS_CLUSTER, redis_key_prefix=REDIS_KEY_PREFIX, ) diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py index 4217124e2b..33fc91f6d9 100644 --- a/backend/open_webui/socket/main.py +++ b/backend/open_webui/socket/main.py @@ -40,8 +40,8 @@ from open_webui.tasks import create_task, stop_item_tasks from open_webui.utils.access_control import has_permission from open_webui.utils.auth import decode_token from open_webui.utils.redis import ( + build_sentinel_url, get_redis_connection, - get_sentinel_url_from_env, get_sentinels_from_env, ) from redis import asyncio as aioredis @@ -58,20 +58,20 @@ REDIS = None SOCKETIO_CORS_ORIGINS = '*' if CORS_ALLOW_ORIGIN == ['*'] else CORS_ALLOW_ORIGIN if WEBSOCKET_MANAGER == 'redis': - if WEBSOCKET_SENTINEL_HOSTS: - mgr = socketio.AsyncRedisManager( - get_sentinel_url_from_env(WEBSOCKET_REDIS_URL, WEBSOCKET_SENTINEL_HOSTS, WEBSOCKET_SENTINEL_PORT), - redis_options=WEBSOCKET_REDIS_OPTIONS, - ) - else: - mgr = socketio.AsyncRedisManager(WEBSOCKET_REDIS_URL, redis_options=WEBSOCKET_REDIS_OPTIONS) + sentinel_hosts = WEBSOCKET_SENTINEL_HOSTS or '' + ws_redis_url = ( + build_sentinel_url(WEBSOCKET_REDIS_URL, sentinel_hosts, WEBSOCKET_SENTINEL_PORT) + if sentinel_hosts + else WEBSOCKET_REDIS_URL + ) + redis_manager = socketio.AsyncRedisManager(ws_redis_url, redis_options=WEBSOCKET_REDIS_OPTIONS) sio = socketio.AsyncServer( cors_allowed_origins=SOCKETIO_CORS_ORIGINS, async_mode='asgi', transports=(['websocket'] if ENABLE_WEBSOCKET_SUPPORT else ['polling']), allow_upgrades=ENABLE_WEBSOCKET_SUPPORT, always_connect=True, - client_manager=mgr, + client_manager=redis_manager, logger=WEBSOCKET_SERVER_LOGGING, ping_interval=WEBSOCKET_SERVER_PING_INTERVAL, ping_timeout=WEBSOCKET_SERVER_PING_TIMEOUT, @@ -99,32 +99,31 @@ SESSION_POOL_TIMEOUT = 120 # seconds without heartbeat before session is reaped if WEBSOCKET_MANAGER == 'redis': log.debug('Using Redis to manage websockets.') + ws_sentinels = get_sentinels_from_env(WEBSOCKET_SENTINEL_HOSTS, WEBSOCKET_SENTINEL_PORT) REDIS = get_redis_connection( redis_url=WEBSOCKET_REDIS_URL, - redis_sentinels=get_sentinels_from_env(WEBSOCKET_SENTINEL_HOSTS, WEBSOCKET_SENTINEL_PORT), + redis_sentinels=ws_sentinels, redis_cluster=WEBSOCKET_REDIS_CLUSTER, async_mode=True, ) - redis_sentinels = get_sentinels_from_env(WEBSOCKET_SENTINEL_HOSTS, WEBSOCKET_SENTINEL_PORT) - MODELS = RedisDict( f'{REDIS_KEY_PREFIX}:models', redis_url=WEBSOCKET_REDIS_URL, - redis_sentinels=redis_sentinels, + redis_sentinels=ws_sentinels, redis_cluster=WEBSOCKET_REDIS_CLUSTER, ) SESSION_POOL = RedisDict( f'{REDIS_KEY_PREFIX}:session_pool', redis_url=WEBSOCKET_REDIS_URL, - redis_sentinels=redis_sentinels, + redis_sentinels=ws_sentinels, redis_cluster=WEBSOCKET_REDIS_CLUSTER, ) USAGE_POOL = RedisDict( f'{REDIS_KEY_PREFIX}:usage_pool', redis_url=WEBSOCKET_REDIS_URL, - redis_sentinels=redis_sentinels, + redis_sentinels=ws_sentinels, redis_cluster=WEBSOCKET_REDIS_CLUSTER, ) @@ -132,7 +131,7 @@ if WEBSOCKET_MANAGER == 'redis': redis_url=WEBSOCKET_REDIS_URL, lock_name=f'{REDIS_KEY_PREFIX}:usage_cleanup_lock', timeout_secs=WEBSOCKET_REDIS_LOCK_TIMEOUT, - redis_sentinels=redis_sentinels, + redis_sentinels=ws_sentinels, redis_cluster=WEBSOCKET_REDIS_CLUSTER, ) aquire_func = clean_up_lock.aquire_lock diff --git a/backend/open_webui/socket/utils.py b/backend/open_webui/socket/utils.py index 0cf2a09246..dd0120d7b9 100644 --- a/backend/open_webui/socket/utils.py +++ b/backend/open_webui/socket/utils.py @@ -1,13 +1,18 @@ +"""Redis-backed distributed data structures for WebSocket state management.""" + +from __future__ import annotations + import json import uuid -from typing import List, Optional, Tuple import pycrdt as Y -from open_webui.env import REDIS_KEY_PREFIX from open_webui.utils.redis import get_redis_connection - +from open_webui.env import REDIS_KEY_PREFIX +YDOC_KEY_PREFIX = f"{REDIS_KEY_PREFIX}:ydoc:documents" class RedisLock: + """Distributed lock backed by a Redis SET with NX/EX semantics.""" + def __init__( self, redis_url, @@ -128,7 +133,7 @@ class YdocManager: def __init__( self, redis=None, - redis_key_prefix: str = f'{REDIS_KEY_PREFIX}:ydoc:documents', + redis_key_prefix: str = YDOC_KEY_PREFIX, ): self._updates = {} self._users = {} @@ -177,7 +182,7 @@ class YdocManager: ydoc.apply_update(bytes(update)) self._updates[document_id] = [ydoc.get_update()] + updates[mid:] - async def get_updates(self, document_id: str) -> List[bytes]: + async def get_updates(self, document_id: str) -> list[bytes]: document_id = document_id.replace(':', '_') if self._redis: @@ -196,7 +201,7 @@ class YdocManager: else: return document_id in self._updates - async def get_users(self, document_id: str) -> List[str]: + async def get_users(self, document_id: str) -> list[str]: document_id = document_id.replace(':', '_') if self._redis: diff --git a/backend/open_webui/utils/redis.py b/backend/open_webui/utils/redis.py index f3a25ef1a8..9339ff948d 100644 --- a/backend/open_webui/utils/redis.py +++ b/backend/open_webui/utils/redis.py @@ -1,10 +1,20 @@ +"""Redis connection utilities. + +Provides connection factory functions for standalone, Sentinel, and Cluster +Redis deployments, with optional async support and automatic connection caching. +""" + +from __future__ import annotations + import asyncio import inspect import logging import time -from urllib.parse import urlparse +from typing import Any +from urllib.parse import ParseResult, urlparse + +import redis as _redis_sync -import redis from open_webui.env import ( REDIS_CLUSTER, REDIS_HEALTH_CHECK_INTERVAL, @@ -19,286 +29,304 @@ from open_webui.env import ( log = logging.getLogger(__name__) -MAX_RETRY_COUNT = REDIS_SENTINEL_MAX_RETRY_COUNT +_ACCEPTED_SCHEMES = frozenset({"redis", "rediss"}) +_SENTINEL_RETRYABLE = ( + _redis_sync.exceptions.ConnectionError, + _redis_sync.exceptions.ReadOnlyError, +) +_FACTORY_METHODS = frozenset({"pipeline", "pubsub", "monitor", "client", "transaction"}) +_CONNECTION_POOL: dict[tuple, Any] = {} -# Let not our connections be timed out but deliver them from -# partition. For the cache and the socket and the uptime -# belong to the one who first opened them, now and always. -_CONNECTION_CACHE = {} - - -class SentinelRedisProxy: - def __init__(self, sentinel, service, *, async_mode: bool = True, **kw): - self._sentinel = sentinel - self._service = service - self._kw = kw - self._async_mode = async_mode - - def _master(self): - return self._sentinel.master_for(self._service, **self._kw) - - def __getattr__(self, item): - master = self._master() - orig_attr = getattr(master, item) - - if not callable(orig_attr): - return orig_attr - - FACTORY_METHODS = {'pipeline', 'pubsub', 'monitor', 'client', 'transaction'} - if item in FACTORY_METHODS: - return orig_attr - - if self._async_mode: - if inspect.isasyncgenfunction(orig_attr): - - def _wrapped_iter(*args, **kwargs): - async def _iter(): - for i in range(REDIS_SENTINEL_MAX_RETRY_COUNT): - try: - method = getattr(self._master(), item) - async for value in method(*args, **kwargs): - yield value - return - except ( - redis.exceptions.ConnectionError, - redis.exceptions.ReadOnlyError, - ) as e: - if i < REDIS_SENTINEL_MAX_RETRY_COUNT - 1: - log.debug( - 'Redis sentinel fail-over (%s). Retry %s/%s', - type(e).__name__, - i + 1, - REDIS_SENTINEL_MAX_RETRY_COUNT, - ) - if REDIS_RECONNECT_DELAY: - time.sleep(REDIS_RECONNECT_DELAY / 1000) - continue - log.error( - 'Redis operation failed after %s retries: %s', - REDIS_SENTINEL_MAX_RETRY_COUNT, - e, - ) - raise e from e - - return _iter() - - return _wrapped_iter - - async def _wrapped(*args, **kwargs): - for i in range(REDIS_SENTINEL_MAX_RETRY_COUNT): - try: - method = getattr(self._master(), item) - result = method(*args, **kwargs) - if inspect.iscoroutine(result): - return await result - return result - except ( - redis.exceptions.ConnectionError, - redis.exceptions.ReadOnlyError, - ) as e: - if i < REDIS_SENTINEL_MAX_RETRY_COUNT - 1: - log.debug( - 'Redis sentinel fail-over (%s). Retry %s/%s', - type(e).__name__, - i + 1, - REDIS_SENTINEL_MAX_RETRY_COUNT, - ) - if REDIS_RECONNECT_DELAY: - await asyncio.sleep(REDIS_RECONNECT_DELAY / 1000) - continue - log.error( - 'Redis operation failed after %s retries: %s', - REDIS_SENTINEL_MAX_RETRY_COUNT, - e, - ) - raise e from e - - return _wrapped - - else: - - def _wrapped(*args, **kwargs): - for i in range(REDIS_SENTINEL_MAX_RETRY_COUNT): - try: - method = getattr(self._master(), item) - return method(*args, **kwargs) - except ( - redis.exceptions.ConnectionError, - redis.exceptions.ReadOnlyError, - ) as e: - if i < REDIS_SENTINEL_MAX_RETRY_COUNT - 1: - log.debug( - 'Redis sentinel fail-over (%s). Retry %s/%s', - type(e).__name__, - i + 1, - REDIS_SENTINEL_MAX_RETRY_COUNT, - ) - if REDIS_RECONNECT_DELAY: - time.sleep(REDIS_RECONNECT_DELAY / 1000) - continue - log.error( - 'Redis operation failed after %s retries: %s', - REDIS_SENTINEL_MAX_RETRY_COUNT, - e, - ) - raise e from e - - return _wrapped - - -def parse_redis_service_url(redis_url): - parsed_url = urlparse(redis_url) - if parsed_url.scheme != 'redis' and parsed_url.scheme != 'rediss': - raise ValueError("Invalid Redis URL scheme. Must be 'redis' or 'rediss'.") - +def parse_redis_url(url: str) -> dict[str, Any]: + """Break a ``redis://`` URL into its parts: service, port, db, username, password.""" + parts: ParseResult = urlparse(url) + if parts.scheme not in _ACCEPTED_SCHEMES: + raise ValueError( + f"Invalid Redis URL scheme '{parts.scheme}'; expected 'redis' or 'rediss'." + ) return { - 'username': parsed_url.username or None, - 'password': parsed_url.password or None, - 'service': parsed_url.hostname or 'mymaster', - 'port': parsed_url.port or 6379, - 'db': int(parsed_url.path.lstrip('/') or 0), + "service": parts.hostname or "mymaster", + "port": parts.port or 6379, + "db": int(parts.path.lstrip("/") or "0"), + "username": parts.username or None, + "password": parts.password or None, } +parse_redis_service_url = parse_redis_url -def get_redis_client(async_mode=False): + +def get_sentinels_from_env( + hosts_csv: str | None, + port: str | int | None, +) -> list[tuple[str, int]]: + """Turn a comma-separated host string into ``[(host, port), …]``.""" + if not hosts_csv: + return [] + resolved_port = int(port) if port else 26379 + return [ + (host.strip(), resolved_port) + for host in hosts_csv.split(",") + if host.strip() + ] + + +def build_sentinel_url( + base_url: str, + hosts_csv: str, + port: str | int, +) -> str: + """Construct a ``redis+sentinel://`` connection string. + + ``base_url`` supplies credentials, db index, and master service name. + ``hosts_csv`` is a comma-separated list of sentinel hostnames. + """ + cfg = parse_redis_url(base_url) + auth = "" + if cfg["username"] or cfg["password"]: + auth = f"{cfg['username'] or ''}:{cfg['password'] or ''}@" + nodes = ",".join( + f"{host.strip()}:{port}" for host in hosts_csv.split(",") if host.strip() + ) + return f"redis+sentinel://{auth}{nodes}/{cfg['db']}/{cfg['service']}" + + +def get_redis_client(async_mode: bool = False) -> Any | None: + """Create a Redis connection using settings from environment variables. + + Returns ``None`` when Redis is not configured or the connection fails. + """ + sentinel_list = get_sentinels_from_env(REDIS_SENTINEL_HOSTS, REDIS_SENTINEL_PORT) + if not REDIS_URL and not sentinel_list: + return None try: return get_redis_connection( - redis_url=REDIS_URL, - redis_sentinels=get_sentinels_from_env(REDIS_SENTINEL_HOSTS, REDIS_SENTINEL_PORT), + REDIS_URL, + redis_sentinels=sentinel_list, redis_cluster=REDIS_CLUSTER, async_mode=async_mode, ) - except Exception as e: - log.debug(f'Failed to get Redis client: {e}') + except Exception: + log.debug("Could not establish Redis connection", exc_info=True) return None +# --------------------------------------------------------------------------- +# Sentinel proxy with automatic failover retry +# --------------------------------------------------------------------------- + + +class SentinelRedisProxy: + """Transparent proxy that re-resolves the Sentinel master on connection errors. + + Every call (sync or async) is wrapped with retry logic so that transient + Sentinel failovers are handled without caller intervention. + """ + + def __init__( + self, + sentinel: Any, + service_name: str, + *, + async_mode: bool = True, + ) -> None: + self._sentinel = sentinel + self._service_name = service_name + self._async_mode = async_mode + + def __getattr__(self, name: str) -> Any: + """Proxy attribute access with automatic Sentinel failover retry.""" + current_master = self._sentinel.master_for(self._service_name) + original = getattr(current_master, name) + + # Non-callable or factory attributes pass through without wrapping. + if not callable(original) or name in _FACTORY_METHODS: + return original + + # Select the retry wrapper matching the execution mode. + if not self._async_mode: + return self._wrap_sync(name) + return self._wrap_async(name, original) + + def _resolve_master(self) -> Any: + """Ask Sentinel for the current master connection.""" + return self._sentinel.master_for(self._service_name) + + def _should_retry(self, attempt: int) -> bool: + return attempt < REDIS_SENTINEL_MAX_RETRY_COUNT - 1 + + def _log_retry(self, exc: Exception, attempt: int) -> None: + log.debug( + "Sentinel failover (%s) — retry %d/%d", + type(exc).__name__, + attempt + 1, + REDIS_SENTINEL_MAX_RETRY_COUNT, + ) + + def _log_exhausted(self, exc: Exception) -> None: + log.error( + "Redis operation failed after %d retries: %s", + REDIS_SENTINEL_MAX_RETRY_COUNT, + exc, + ) + + # -- async wrappers ----------------------------------------------------- + + def _wrap_async(self, name: str, attr: Any) -> Any: + if inspect.isasyncgenfunction(attr): + return self._wrap_async_gen(name) + return self._wrap_async_call(name) + + def _wrap_async_gen(self, name: str) -> Any: + proxy = self + + def wrapper(*args: Any, **kwargs: Any) -> Any: + async def _inner(): + for attempt in range(REDIS_SENTINEL_MAX_RETRY_COUNT): + try: + method = getattr(proxy._resolve_master(), name) + async for value in method(*args, **kwargs): + yield value + return + except _SENTINEL_RETRYABLE as exc: + if proxy._should_retry(attempt): + proxy._log_retry(exc, attempt) + if REDIS_RECONNECT_DELAY: + time.sleep(REDIS_RECONNECT_DELAY / 1000) + continue + proxy._log_exhausted(exc) + raise + + return _inner() + + return wrapper + + def _wrap_async_call(self, name: str) -> Any: + proxy = self + + async def wrapper(*args: Any, **kwargs: Any) -> Any: + for attempt in range(REDIS_SENTINEL_MAX_RETRY_COUNT): + try: + method = getattr(proxy._resolve_master(), name) + result = method(*args, **kwargs) + if inspect.iscoroutine(result): + return await result + return result + except _SENTINEL_RETRYABLE as exc: + if proxy._should_retry(attempt): + proxy._log_retry(exc, attempt) + if REDIS_RECONNECT_DELAY: + await asyncio.sleep(REDIS_RECONNECT_DELAY / 1000) + continue + proxy._log_exhausted(exc) + raise + + return wrapper + + # -- sync wrapper ------------------------------------------------------- + + def _wrap_sync(self, name: str) -> Any: + proxy = self + + def wrapper(*args: Any, **kwargs: Any) -> Any: + for attempt in range(REDIS_SENTINEL_MAX_RETRY_COUNT): + try: + method = getattr(proxy._resolve_master(), name) + return method(*args, **kwargs) + except _SENTINEL_RETRYABLE as exc: + if proxy._should_retry(attempt): + proxy._log_retry(exc, attempt) + if REDIS_RECONNECT_DELAY: + time.sleep(REDIS_RECONNECT_DELAY / 1000) + continue + proxy._log_exhausted(exc) + raise + + return wrapper + + +# --------------------------------------------------------------------------- +# Connection factory +# --------------------------------------------------------------------------- + + +def _socket_options() -> dict[str, Any]: + """Collect optional socket-level kwargs once instead of repeating them.""" + opts: dict[str, Any] = {} + if REDIS_SOCKET_CONNECT_TIMEOUT is not None: + opts["socket_connect_timeout"] = REDIS_SOCKET_CONNECT_TIMEOUT + if REDIS_SOCKET_KEEPALIVE: + opts["socket_keepalive"] = True + if REDIS_HEALTH_CHECK_INTERVAL: + opts["health_check_interval"] = REDIS_HEALTH_CHECK_INTERVAL + return opts + + +def _build_sentinel( + redis_module: Any, + url: str, + sentinels: list[tuple[str, int]], + decode_responses: bool, + async_mode: bool, +) -> SentinelRedisProxy: + """Create a SentinelRedisProxy from a redis URL and sentinel list.""" + cfg = parse_redis_url(url) + sentinel = redis_module.sentinel.Sentinel( + sentinels, + port=cfg["port"], + db=cfg["db"], + username=cfg["username"], + password=cfg["password"], + decode_responses=decode_responses, + socket_connect_timeout=REDIS_SOCKET_CONNECT_TIMEOUT, + **{k: v for k, v in _socket_options().items() if k != "socket_connect_timeout"}, + ) + return SentinelRedisProxy(sentinel, cfg["service"], async_mode=async_mode) + + def get_redis_connection( - redis_url, - redis_sentinels, - redis_cluster=False, - async_mode=False, - decode_responses=True, -): + redis_url: str | None, + redis_sentinels: list[tuple[str, int]] | None = None, + redis_cluster: bool = False, + async_mode: bool = False, + decode_responses: bool = True, +) -> Any | None: + """Return a cached Redis connection (or create one). + + Supports three topologies in order of precedence: + 1. **Sentinel** — when ``redis_sentinels`` is non-empty. + 2. **Cluster** — when ``redis_cluster`` is True. + 3. **Standalone** — plain ``redis://`` connection. + """ cache_key = ( redis_url, tuple(redis_sentinels) if redis_sentinels else (), async_mode, decode_responses, ) + if cache_key in _CONNECTION_POOL: + return _CONNECTION_POOL[cache_key] - if cache_key in _CONNECTION_CACHE: - return _CONNECTION_CACHE[cache_key] - - connection = None - - connect_timeout_kwargs = ( - {'socket_connect_timeout': REDIS_SOCKET_CONNECT_TIMEOUT} if REDIS_SOCKET_CONNECT_TIMEOUT is not None else {} - ) - - keepalive_kwargs = {'socket_keepalive': True} if REDIS_SOCKET_KEEPALIVE else {} - - health_check_kwargs = {'health_check_interval': REDIS_HEALTH_CHECK_INTERVAL} if REDIS_HEALTH_CHECK_INTERVAL else {} + extra = _socket_options() + connection: Any = None + # Pick the right redis module for sync vs async. if async_mode: - import redis.asyncio as redis - - # If using sentinel in async mode - if redis_sentinels: - redis_config = parse_redis_service_url(redis_url) - sentinel = redis.sentinel.Sentinel( - redis_sentinels, - port=redis_config['port'], - db=redis_config['db'], - username=redis_config['username'], - password=redis_config['password'], - decode_responses=decode_responses, - socket_connect_timeout=REDIS_SOCKET_CONNECT_TIMEOUT, - **keepalive_kwargs, - **health_check_kwargs, - ) - connection = SentinelRedisProxy( - sentinel, - redis_config['service'], - async_mode=async_mode, - ) - elif redis_cluster: - if not redis_url: - raise ValueError('Redis URL must be provided for cluster mode.') - return redis.cluster.RedisCluster.from_url( - redis_url, - decode_responses=decode_responses, - **connect_timeout_kwargs, - **keepalive_kwargs, - **health_check_kwargs, - ) - elif redis_url: - connection = redis.from_url( - redis_url, - decode_responses=decode_responses, - **connect_timeout_kwargs, - **keepalive_kwargs, - **health_check_kwargs, - ) + import redis.asyncio as redis_mod else: - import redis + import redis as redis_mod # type: ignore[no-redef] - if redis_sentinels: - redis_config = parse_redis_service_url(redis_url) - sentinel = redis.sentinel.Sentinel( - redis_sentinels, - port=redis_config['port'], - db=redis_config['db'], - username=redis_config['username'], - password=redis_config['password'], - decode_responses=decode_responses, - socket_connect_timeout=REDIS_SOCKET_CONNECT_TIMEOUT, - **keepalive_kwargs, - **health_check_kwargs, - ) - connection = SentinelRedisProxy( - sentinel, - redis_config['service'], - async_mode=async_mode, - ) - elif redis_cluster: - if not redis_url: - raise ValueError('Redis URL must be provided for cluster mode.') - return redis.cluster.RedisCluster.from_url( - redis_url, - decode_responses=decode_responses, - **connect_timeout_kwargs, - **keepalive_kwargs, - **health_check_kwargs, - ) - elif redis_url: - connection = redis.Redis.from_url( - redis_url, - decode_responses=decode_responses, - **connect_timeout_kwargs, - **keepalive_kwargs, - **health_check_kwargs, - ) + if redis_sentinels: + connection = _build_sentinel(redis_mod, redis_url, redis_sentinels, decode_responses, async_mode) + elif redis_cluster: + if not redis_url: + raise ValueError("Redis URL is required for cluster mode.") + connection = redis_mod.cluster.RedisCluster.from_url( + redis_url, decode_responses=decode_responses, **extra, + ) + elif redis_url: + factory = getattr(redis_mod, "from_url", None) or redis_mod.Redis.from_url + connection = factory(redis_url, decode_responses=decode_responses, **extra) - _CONNECTION_CACHE[cache_key] = connection + _CONNECTION_POOL[cache_key] = connection return connection - - -def get_sentinels_from_env(sentinel_hosts_env, sentinel_port_env): - if sentinel_hosts_env: - sentinel_hosts = sentinel_hosts_env.split(',') - sentinel_port = int(sentinel_port_env) - return [(host, sentinel_port) for host in sentinel_hosts] - return [] - - -def get_sentinel_url_from_env(redis_url, sentinel_hosts_env, sentinel_port_env): - redis_config = parse_redis_service_url(redis_url) - username = redis_config['username'] or '' - password = redis_config['password'] or '' - auth_part = '' - if username or password: - auth_part = f'{username}:{password}@' - hosts_part = ','.join(f'{host}:{sentinel_port_env}' for host in sentinel_hosts_env.split(',')) - return f'redis+sentinel://{auth_part}{hosts_part}/{redis_config["db"]}/{redis_config["service"]}'