bhetherman/open-webui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refac: clean up Redis sentinel utilities and import grouping

Browse Source
This commit is contained in:
Timothy Jaeryang Baek
2026-05-21 11:47:25 +04:00
parent 1527eb6e01
commit 154679200f
5 changed files with 321 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/open_webui/internal/config.py
+7
View File
@@ -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)
backend/open_webui/main.py
+3 -10
View File
@@ -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,
)
backend/open_webui/socket/main.py
+15 -16
View File
@@ -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
backend/open_webui/socket/utils.py
+11 -6
View File
@@ -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:
backend/open_webui/utils/redis.py
+285 -257
View File
@@ -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"]}'