Add Redis Sentinel support to prefect-redis

Build on the Redis Cluster URL groundwork merged in #22211 to add native
Redis Sentinel support. Cluster URLs stay gated (detection-only) exactly as
in #22211; this change enables the redis+sentinel:// and rediss+sentinel://
schemes across every connection point: the server messaging client, the
RedisLockManager, the RedisDatabase block, and the Docket services URL.

- New prefect_redis.connection module: a shared URL parser (parse_redis_url)
  and client builder (build_redis_client) that resolve the current master
  through the listed Sentinel daemons and follow failover automatically.
  Data-node connections get pinned TCP keepalive so a silently-dead master is
  noticed promptly while Sentinel completes failover.
- client.py dispatches redis+sentinel:// URLs to the Sentinel builder while
  leaving #22211's cluster NotImplementedError gate intact.
- RedisLockManager and RedisDatabase accept a connection_url that is
  authoritative over the scalar host/port fields; RedisDatabase round-trips
  Sentinel URLs through from_connection_string and block_info.
- Document the Sentinel schemes on PREFECT_SERVER_DOCKET_URL and bump the
  pydocket floor to >=0.22.0 for its Sentinel URL support.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: fatih-acar

docs/integrations/prefect-redis/index.mdx

@@ -30,6 +30,81 @@ Register the block types in the `prefect-redis` module to make them available fo
 prefect block register -m prefect_redis
 ```
 
+## Connecting with a URL (including Redis Sentinel)
+
+Every `prefect-redis` connection point — the server messaging client, the
+`RedisLockManager`, and the `RedisDatabase` block — accepts a full connection URL in
+addition to the individual `host`/`port`/`db`/`username`/`password`/`ssl` fields. When
+a URL is provided it is authoritative and the scalar fields are ignored.
+
+The following schemes are supported:
+
+| Scheme | Behavior |
+| --- | --- |
+| `redis://` | Single-node TCP connection |
+| `rediss://` | Single-node TLS connection |
+| `redis+sentinel://` | [Redis Sentinel](https://redis.io/docs/latest/operate/oss_and_stack/management/sentinel/) discovery (TCP data nodes) |
+| `rediss+sentinel://` | Redis Sentinel discovery (TLS data nodes) |
+
+The Sentinel schemes accept a comma-separated list of Sentinel members and a master
+group name, and the resulting client resolves the current master through the Sentinel
+daemons and follows failover automatically:
+
+```
+redis+sentinel://[user:pass@]sentinel-a:26379,sentinel-b:26379/<master-name>[/<db>][?params]
+```
+
+Sentinel-specific query parameters configure the connections to the Sentinel daemons
+separately from the data nodes: `sentinel_username`, `sentinel_password`, `sentinel_ssl`,
+`sentinel_tls_insecure`, and `sentinel_tls_ca_file`. Single-node TLS URLs additionally
+accept `tls_insecure` and `tls_ca_file`.
+
+### Server messaging (events / streams)
+
+Point the Prefect server's Redis messaging at a Sentinel topology with a single
+environment variable:
+
+```bash
+export PREFECT_REDIS_MESSAGING_URL="redis+sentinel://sentinel-a:26379,sentinel-b:26379/mymaster"
+```
+
+The same setting also accepts single-node `redis://` / `rediss://` URLs.
+
+### Background services (Docket)
+
+Multi-server deployments also point `PREFECT_SERVER_DOCKET_URL` at the same shared
+Redis so the server's background services coordinate through it. That URL is
+resolved by [Docket](https://github.com/chrisguidry/docket) rather than
+`prefect-redis`, so Sentinel URLs require a `pydocket` release that includes Redis
+Sentinel support:
+
+```bash
+export PREFECT_SERVER_DOCKET_URL="redis+sentinel://sentinel-a:26379,sentinel-b:26379/mymaster/0"
+```
+
+### Lock manager
+
+{/* pmd-metadata: notest */}
+```python
+from prefect_redis import RedisLockManager
+
+lock_manager = RedisLockManager(
+    connection_url="redis+sentinel://sentinel-a:26379,sentinel-b:26379/mymaster"
+)
+```
+
+### `RedisDatabase` block
+
+{/* pmd-metadata: notest */}
+```python
+from prefect_redis import RedisDatabase
+
+block = RedisDatabase.from_connection_string(
+    "redis+sentinel://sentinel-a:26379,sentinel-b:26379/mymaster"
+)
+block.save("BLOCK_NAME")
+```
+
 ## Resources
 
 Refer to the [SDK documentation](/integrations/prefect-redis/api-ref/prefect_redis-blocks) to explore all the capabilities of `prefect-redis`.

docs/v3/advanced/self-hosted.mdx

@@ -189,6 +189,14 @@ For Redis instances that require SSL/TLS:
 export PREFECT_SERVER_DOCKET_URL="rediss://redis-host:6379/0"
 ```
 
+For a Redis deployment fronted by [Redis Sentinel](https://redis.io/docs/latest/operate/oss_and_stack/management/sentinel/), use the `redis+sentinel://` scheme with a comma-separated list of Sentinel daemons and the master group name. Docket discovers the current master through Sentinel and follows failover automatically:
+
+```bash
+export PREFECT_SERVER_DOCKET_URL="redis+sentinel://sentinel-a:26379,sentinel-b:26379,sentinel-c:26379/mymaster/0"
+```
+
+Use the `rediss+sentinel://` scheme for SSL/TLS. This requires `pydocket>=0.22.0`.
+
 <Note>
 The Docket URL can use the same Redis instance as the messaging configuration above, but you may use a different database number (e.g., `/1` instead of `/0`) to keep the data separate.
 </Note>

docs/v3/api-ref/settings-ref.mdx

@@ -1691,7 +1691,7 @@ The name of the Docket instance.
 `PREFECT_SERVER_DOCKET_NAME`
 
 ### `url`
-The URL of the Redis server to use for Docket.
+The URL of the Redis server to use for Docket. Supports the memory:// (single-server only), redis://, rediss://, redis+sentinel:// and rediss+sentinel:// schemes; the Sentinel schemes discover the current master through the listed Sentinel daemons and follow failover automatically (requires pydocket>=0.22.0).
 
 **Type**: `string`
 

pyproject.toml

@@ -83,7 +83,7 @@ dependencies = [
     "whenever>=0.7.3,<0.11.0; python_version>='3.13'",
     "semver>=3.0.4",
     "pluggy>=1.6.0",
-    "pydocket>=0.19.0",
+    "pydocket>=0.22.0",
 ]
 [project.urls]
 Changelog = "https://github.com/PrefectHQ/prefect/releases"

schemas/settings.schema.json

@@ -1502,7 +1502,7 @@
                 },
                 "url": {
                     "default": "memory://",
-                    "description": "The URL of the Redis server to use for Docket.",
+                    "description": "The URL of the Redis server to use for Docket. Supports the memory:// (single-server only), redis://, rediss://, redis+sentinel:// and rediss+sentinel:// schemes; the Sentinel schemes discover the current master through the listed Sentinel daemons and follow failover automatically (requires pydocket>=0.22.0).",
                     "supported_environment_variables": [
                         "PREFECT_SERVER_DOCKET_URL"
                     ],

src/integrations/prefect-redis/prefect_redis/blocks.py

@@ -1,6 +1,7 @@
 """Redis credentials handling"""
 
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, Optional, Union, cast
+from urllib.parse import urlsplit
 
 import redis
 import redis.asyncio
@@ -9,6 +10,11 @@
 from redis.asyncio.connection import parse_url
 
 from prefect.filesystems import WritableFileSystem
+from prefect_redis.connection import (
+    SCHEME_SENTINEL,
+    build_redis_client,
+    parse_redis_url,
+)
 
 DEFAULT_PORT = 6379
 
@@ -59,10 +65,24 @@ class RedisDatabase(WritableFileSystem):
     username: Optional[SecretStr] = Field(default=None, description="Redis username")
     password: Optional[SecretStr] = Field(default=None, description="Redis password")
     ssl: bool = Field(default=False, description="Whether to use SSL")
+    connection_url: Optional[SecretStr] = Field(
+        default=None,
+        description=(
+            "Full Redis connection URL, authoritative over the scalar connection fields "
+            "when set. Supports the redis://, rediss://, redis+sentinel:// and "
+            "rediss+sentinel:// schemes; the Sentinel schemes resolve the current master "
+            "through the listed Sentinel daemons and follow failover automatically, e.g. "
+            "redis+sentinel://sentinel-a:26379,sentinel-b:26379/mymaster."
+        ),
+    )
 
     def block_initialization(self) -> None:
         """Validate parameters"""
 
+        if self.connection_url is not None:
+            # Fail fast on a malformed URL rather than at first connection.
+            parse_redis_url(self.connection_url.get_secret_value())
+            return
         if not self.host:
             raise ValueError("Missing hostname")
         if self.username and not self.password:
@@ -100,8 +120,16 @@ def get_client(self) -> redis.Redis:
         """Get Redis Client
 
         Returns:
-            An initialized Redis async client
+            An initialized Redis client
         """
+        if self.connection_url is not None:
+            return cast(
+                redis.Redis,
+                build_redis_client(
+                    parse_redis_url(self.connection_url.get_secret_value()),
+                    asynchronous=False,
+                ),
+            )
         return redis.Redis(
             host=self.host,
             port=self.port,
@@ -117,6 +145,14 @@ def get_async_client(self) -> redis.asyncio.Redis:
         Returns:
             An initialized Redis async client
         """
+        if self.connection_url is not None:
+            return cast(
+                redis.asyncio.Redis,
+                build_redis_client(
+                    parse_redis_url(self.connection_url.get_secret_value()),
+                    asynchronous=True,
+                ),
+            )
         return redis.asyncio.Redis(
             host=self.host,
             port=self.port,
@@ -135,18 +171,27 @@ def from_connection_string(
         Supports the following URL schemes:
         - `redis://` creates a TCP socket connection
         - `rediss://` creates a SSL wrapped TCP socket connection
+        - `redis+sentinel://` / `rediss+sentinel://` discover the master through
+          Redis Sentinel and follow failover automatically
 
         Args:
             connection_string: Redis connection string
 
         Returns:
-            `RedisCredentials` instance
+            `RedisDatabase` instance
         """
-        connection_kwargs = parse_url(
+        raw_connection_string = (
             connection_string
             if isinstance(connection_string, str)
             else connection_string.get_secret_value()
         )
+
+        # Sentinel URLs cannot be flattened to scalar host/port fields, so they are
+        # retained verbatim and resolved through the Sentinel daemons at connect time.
+        if urlsplit(raw_connection_string).scheme.lower() in SCHEME_SENTINEL:
+            return cls(connection_url=SecretStr(raw_connection_string))
+
+        connection_kwargs = parse_url(raw_connection_string)
         ssl = connection_kwargs.get("connection_class") == redis.asyncio.SSLConnection
         return cls(
             host=connection_kwargs.get("host", "localhost"),
@@ -174,4 +219,9 @@ def as_connection_params(self) -> Dict[str, Any]:
         else:
             data.pop("password", None)
 
+        if self.connection_url is not None:
+            data["connection_url"] = self.connection_url.get_secret_value()
+        else:
+            data.pop("connection_url", None)
+
         return data

src/integrations/prefect-redis/prefect_redis/client.py

@@ -1,7 +1,7 @@
 import asyncio
 import functools
 import warnings
-from typing import Any, Callable, Optional, Union
+from typing import Any, Callable, Optional, Union, cast
 from urllib.parse import urlparse, urlunparse
 
 from pydantic import Field, model_validator
@@ -12,6 +12,11 @@
     PrefectBaseSettings,
     build_settings_config,  # type: ignore[reportPrivateUsage]
 )
+from prefect_redis.connection import (
+    SCHEME_SENTINEL,
+    build_redis_client,
+    parse_redis_url,
+)
 
 _UNSET: Any = object()
 
@@ -39,8 +44,12 @@ class RedisMessagingSettings(PrefectBaseSettings):
         default=None,
         description=(
             "Full Redis URL (e.g. redis://user:pass@host:6379/0 or "
-            "rediss://… for TLS). When set, host/port/db/username/"
-            "password/ssl are ignored."
+            "rediss://… for TLS). Also supports redis+sentinel:// and "
+            "rediss+sentinel:// for Redis Sentinel; the Sentinel schemes "
+            "accept a comma-separated list of members and a master group "
+            "name, e.g. "
+            "redis+sentinel://sentinel-a:26379,sentinel-b:26379/mymaster. "
+            "When set, host/port/db/username/password/ssl are ignored."
         ),
     )
     host: str = Field(default="localhost")
@@ -108,6 +117,11 @@ def is_cluster_url(url: str) -> bool:
     return url.partition("://")[0] in {"redis+cluster", "rediss+cluster"}
 
 
+def is_sentinel_url(url: str) -> bool:
+    """Return True if the URL uses a Redis Sentinel scheme."""
+    return urlparse(url).scheme in SCHEME_SENTINEL
+
+
 def normalize_cluster_url(url: str) -> str:
     """Return a redis-py compatible URL for Redis Cluster connections."""
     parsed = urlparse(url)
@@ -200,11 +214,14 @@ def get_async_redis_client(
 
     When a standalone `url` is provided (or configured via
     `PREFECT_REDIS_MESSAGING_URL`), `Redis.from_url` is used and
-    the discrete host/port/… arguments are ignored. Redis Cluster
-    URLs are detected but intentionally not enabled yet.
+    the discrete host/port/… arguments are ignored. `redis+sentinel://`
+    and `rediss+sentinel://` URLs resolve the current master through the
+    listed Sentinel daemons and follow failover automatically. Redis
+    Cluster URLs are detected but intentionally not enabled yet.
 
     Args:
-        url: Full Redis URL (e.g. `redis://localhost:6379/0`).
+        url: Full Redis URL (e.g. `redis://localhost:6379/0` or
+            `redis+sentinel://sentinel-a:26379,sentinel-b:26379/mymaster`).
         host: The host location.
         port: The port to connect to the host with.
         db: The Redis database to interact with.
@@ -235,6 +252,20 @@ def get_async_redis_client(
 
     url = url or settings.url
     if url:
+        if is_sentinel_url(url):
+            return cast(
+                Redis,
+                build_redis_client(
+                    parse_redis_url(url),
+                    asynchronous=True,
+                    health_check_interval=health_check_interval
+                    or settings.health_check_interval,
+                    decode_responses=decode_responses,
+                    socket_timeout=resolved_socket_timeout,
+                    socket_connect_timeout=resolved_socket_connect_timeout,
+                    protocol=resolved_protocol,
+                ),
+            )
         if is_cluster_url(url):
             _raise_cluster_not_supported()
         return Redis.from_url(
@@ -275,6 +306,16 @@ def async_redis_from_settings(
     }
 
     if settings.url:
+        if is_sentinel_url(settings.url):
+            return cast(
+                Redis,
+                build_redis_client(
+                    parse_redis_url(settings.url),
+                    asynchronous=True,
+                    health_check_interval=settings.health_check_interval,
+                    **options,
+                ),
+            )
         if is_cluster_url(settings.url):
             _raise_cluster_not_supported()
         return Redis.from_url(