Add DriverInfo class for upstream driver tracking

redis/__init__.py

@@ -1,6 +1,7 @@
 from redis import asyncio  # noqa
 from redis.backoff import default_backoff
 from redis.client import Redis, StrictRedis
+from redis.driver_info import DriverInfo
 from redis.cluster import RedisCluster
 from redis.connection import (
     BlockingConnectionPool,
@@ -63,6 +64,7 @@ def int_or_str(value):
     "CredentialProvider",
     "CrossSlotTransactionError",
     "DataError",
+    "DriverInfo",
     "from_url",
     "default_backoff",
     "InvalidPipelineStack",

redis/asyncio/client.py

@@ -53,6 +53,7 @@
     list_or_args,
 )
 from redis.credentials import CredentialProvider
+from redis.driver_info import DriverInfo
 from redis.event import (
     AfterPooledConnectionsInstantiationEvent,
     AfterPubSubConnectionInstantiationEvent,
@@ -252,6 +253,7 @@ def __init__(
         client_name: Optional[str] = None,
         lib_name: Optional[str] = "redis-py",
         lib_version: Optional[str] = get_lib_version(),
+        driver_info: Optional["DriverInfo"] = None,
         username: Optional[str] = None,
         auto_close_connection_pool: Optional[bool] = None,
         redis_connect_func=None,
@@ -304,6 +306,11 @@ def __init__(
             # Create internal connection pool, expected to be closed by Redis instance
             if not retry_on_error:
                 retry_on_error = []
+            if driver_info is not None:
+                computed_lib_name = driver_info.formatted_name
+            else:
+                computed_lib_name = lib_name
+
             kwargs = {
                 "db": db,
                 "username": username,
@@ -318,7 +325,7 @@ def __init__(
                 "max_connections": max_connections,
                 "health_check_interval": health_check_interval,
                 "client_name": client_name,
-                "lib_name": lib_name,
+                "lib_name": computed_lib_name,
                 "lib_version": lib_version,
                 "redis_connect_func": redis_connect_func,
                 "protocol": protocol,

redis/client.py

@@ -40,6 +40,7 @@
     UnixDomainSocketConnection,
 )
 from redis.credentials import CredentialProvider
+from redis.driver_info import DriverInfo
 from redis.event import (
     AfterPooledConnectionsInstantiationEvent,
     AfterPubSubConnectionInstantiationEvent,
@@ -242,6 +243,7 @@ def __init__(
         client_name: Optional[str] = None,
         lib_name: Optional[str] = "redis-py",
         lib_version: Optional[str] = get_lib_version(),
+        driver_info: Optional["DriverInfo"] = None,
         username: Optional[str] = None,
         redis_connect_func: Optional[Callable[[], None]] = None,
         credential_provider: Optional[CredentialProvider] = None,
@@ -280,6 +282,9 @@ def __init__(
         decode_responses:
             if `True`, the response will be decoded to utf-8.
             Argument is ignored when connection_pool is provided.
+        driver_info:
+            Optional DriverInfo object to identify upstream libraries.
+            Argument is ignored when connection_pool is provided.
         maint_notifications_config:
             configuration the pool to support maintenance notifications - see
             `redis.maint_notifications.MaintNotificationsConfig` for details.
@@ -296,6 +301,11 @@ def __init__(
         if not connection_pool:
             if not retry_on_error:
                 retry_on_error = []
+            if driver_info is not None:
+                computed_lib_name = driver_info.formatted_name
+            else:
+                computed_lib_name = lib_name
+
             kwargs = {
                 "db": db,
                 "username": username,
@@ -309,7 +319,7 @@ def __init__(
                 "max_connections": max_connections,
                 "health_check_interval": health_check_interval,
                 "client_name": client_name,
-                "lib_name": lib_name,
+                "lib_name": computed_lib_name,
                 "lib_version": lib_version,
                 "redis_connect_func": redis_connect_func,
                 "credential_provider": credential_provider,

redis/driver_info.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import List
+
+_BRACES = {"(", ")", "[", "]", "{", "}"}
+
+
+def _validate_no_invalid_chars(value: str, field_name: str) -> None:
+    """Ensure value contains only printable ASCII without spaces or braces.
+
+    This mirrors the constraints enforced by other Redis clients for values that
+    will appear in CLIENT LIST / CLIENT INFO output.
+    """
+
+    for ch in value:
+        # printable ASCII without space: '!' (0x21) to '~' (0x7E)
+        if ord(ch) < 0x21 or ord(ch) > 0x7E or ch in _BRACES:
+            raise ValueError(
+                f"{field_name} must not contain spaces, newlines, non-printable characters, or braces"
+            )
+
+
+def _validate_driver_name(name: str) -> None:
+    """Validate an upstream driver name.
+
+    The name should look like a typical Python distribution or package name,
+    following a simplified form of PEP 503 normalisation rules:
+
+    * start with a lowercase ASCII letter
+    * contain only lowercase letters, digits, hyphens and underscores
+
+    Examples of valid names: ``"django-redis"``, ``"celery"``, ``"rq"``.
+    """
+
+    import re
+
+    _validate_no_invalid_chars(name, "Driver name")
+    if not re.match(r"^[a-z][a-z0-9_-]*$", name):
+        raise ValueError(
+            "Upstream driver name must use a Python package-style name: "
+            "start with a lowercase letter and contain only lowercase letters, "
+            "digits, hyphens, and underscores (e.g., 'django-redis')."
+        )
+
+
+def _validate_driver_version(version: str) -> None:
+    _validate_no_invalid_chars(version, "Driver version")
+
+
+def _format_driver_entry(driver_name: str, driver_version: str) -> str:
+    return f"{driver_name}_v{driver_version}"
+
+
+@dataclass
+class DriverInfo:
+    """Driver information used to build the CLIENT SETINFO LIB-NAME value.
+
+    The formatted name follows the pattern::
+
+        name(driver1_vVersion1;driver2_vVersion2)
+
+    Examples
+    --------
+    >>> info = DriverInfo()
+    >>> info.formatted_name
+    'redis-py'
+
+    >>> info = DriverInfo().add_upstream_driver("django-redis", "5.4.0")
+    >>> info.formatted_name
+    'redis-py(django-redis_v5.4.0)'
+    """
+
+    name: str = "redis-py"
+    _upstream: List[str] = field(default_factory=list)
+
+    @property
+    def upstream_drivers(self) -> List[str]:
+        """Return a copy of the upstream driver entries.
+
+        Each entry is in the form ``"driver-name_vversion"``.
+        """
+
+        return list(self._upstream)
+
+    def add_upstream_driver(
+        self, driver_name: str, driver_version: str
+    ) -> "DriverInfo":
+        """Add an upstream driver to this instance and return self.
+
+        The most recently added driver appears first in :pyattr:`formatted_name`.
+        """
+
+        if driver_name is None:
+            raise ValueError("Driver name must not be None")
+        if driver_version is None:
+            raise ValueError("Driver version must not be None")
+
+        _validate_driver_name(driver_name)
+        _validate_driver_version(driver_version)
+
+        entry = _format_driver_entry(driver_name, driver_version)
+        # insert at the beginning so latest is first
+        self._upstream.insert(0, entry)
+        return self
+
+    @property
+    def formatted_name(self) -> str:
+        """Return the base name with upstream drivers encoded, if any.
+
+        With no upstream drivers, this is just :pyattr:`name`. Otherwise::
+
+            name(driver1_vX;driver2_vY)
+        """
+
+        if not self._upstream:
+            return self.name
+        return f"{self.name}({';'.join(self._upstream)})"

tests/test_asyncio/test_commands.py

@@ -541,6 +541,17 @@ async def test_client_setinfo(self, r: redis.Redis):
         info = await r2.client_info()
         assert info["lib-name"] == "test2"
         assert info["lib-ver"] == "1234"
+
+    @skip_if_server_version_lt("7.2.0")
+    async def test_client_setinfo_with_driver_info(self, r: redis.Redis):
+        from redis import DriverInfo
+
+        info = DriverInfo().add_upstream_driver("celery", "5.4.1")
+        r2 = redis.asyncio.Redis(driver_info=info)
+        await r2.ping()
+        client_info = await r2.client_info()
+        assert client_info["lib-name"] == "redis-py(celery_v5.4.1)"
+        assert client_info["lib-ver"] == redis.__version__
         await r2.aclose()
         r3 = redis.asyncio.Redis(lib_name=None, lib_version=None)
         info = await r3.client_info()

tests/test_commands.py

@@ -744,6 +744,17 @@ def test_client_setinfo(self, r: redis.Redis):
         info = r2.client_info()
         assert info["lib-name"] == "test2"
         assert info["lib-ver"] == "1234"
+
+    @skip_if_server_version_lt("7.2.0")
+    def test_client_setinfo_with_driver_info(self, r: redis.Redis):
+        from redis import DriverInfo
+
+        info = DriverInfo().add_upstream_driver("django-redis", "5.4.0")
+        r2 = redis.Redis(driver_info=info)
+        r2.ping()
+        client_info = r2.client_info()
+        assert client_info["lib-name"] == "redis-py(django-redis_v5.4.0)"
+        assert client_info["lib-ver"] == redis.__version__
         r3 = redis.Redis(lib_name=None, lib_version=None)
         info = r3.client_info()
         assert info["lib-name"] == ""

tests/test_driver_info.py

@@ -0,0 +1,50 @@
+import pytest
+
+from redis.driver_info import DriverInfo
+
+
+def test_driver_info_default_name_no_upstream():
+    info = DriverInfo()
+    assert info.formatted_name == "redis-py"
+    assert info.upstream_drivers == []
+
+
+def test_driver_info_single_upstream():
+    info = DriverInfo().add_upstream_driver("django-redis", "5.4.0")
+    assert info.formatted_name == "redis-py(django-redis_v5.4.0)"
+
+
+def test_driver_info_multiple_upstreams_latest_first():
+    info = DriverInfo()
+    info.add_upstream_driver("django-redis", "5.4.0")
+    info.add_upstream_driver("celery", "5.4.1")
+    assert info.formatted_name == "redis-py(celery_v5.4.1;django-redis_v5.4.0)"
+
+
+@pytest.mark.parametrize(
+    "name",
+    [
+        "DjangoRedis",  # must start with lowercase
+        "django redis",  # spaces not allowed
+        "django{redis}",  # braces not allowed
+        "django:redis",  # ':' not allowed by validation regex
+    ],
+)
+def test_driver_info_invalid_name(name):
+    info = DriverInfo()
+    with pytest.raises(ValueError):
+        info.add_upstream_driver(name, "3.2.0")
+
+
+@pytest.mark.parametrize(
+    "version",
+    [
+        "3.2.0 beta",  # space not allowed
+        "3.2.0)",  # brace not allowed
+        "3.2.0\n",  # newline not allowed
+    ],
+)
+def test_driver_info_invalid_version(version):
+    info = DriverInfo()
+    with pytest.raises(ValueError):
+        info.add_upstream_driver("django-redis", version)