Refactor: extract resolve_driver_info helper to eliminate duplication

Author: vchomakov

redis/asyncio/client.py

@@ -53,7 +53,7 @@
     list_or_args,
 )
 from redis.credentials import CredentialProvider
-from redis.driver_info import DriverInfo
+from redis.driver_info import DriverInfo, resolve_driver_info
 from redis.event import (
     AfterPooledConnectionsInstantiationEvent,
     AfterPubSubConnectionInstantiationEvent,
@@ -75,7 +75,6 @@
     _set_info_logger,
     deprecated_args,
     deprecated_function,
-    get_lib_version,
     safe_str,
     str_if_bytes,
     truncate_text,
@@ -315,14 +314,9 @@ def __init__(
                 retry_on_error = []
 
             # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version
-            if driver_info is not None:
-                computed_driver_info = driver_info
-            else:
-                # Fallback: create DriverInfo from lib_name and lib_version
-                # Use defaults if not provided
-                name = lib_name if lib_name is not None else "redis-py"
-                version = lib_version if lib_version is not None else get_lib_version()
-                computed_driver_info = DriverInfo(name=name, lib_version=version)
+            computed_driver_info = resolve_driver_info(
+                driver_info, lib_name, lib_version
+            )
 
             kwargs = {
                 "db": db,

redis/asyncio/connection.py

@@ -38,7 +38,7 @@
     VerifyFlags = None
 
 from ..auth.token import TokenInterface
-from ..driver_info import DriverInfo
+from ..driver_info import DriverInfo, resolve_driver_info
 from ..event import AsyncAfterConnectionReleasedEvent, EventDispatcher
 from ..utils import deprecated_args, format_error_message
 
@@ -64,7 +64,7 @@
     TimeoutError,
 )
 from redis.typing import EncodableT
-from redis.utils import HIREDIS_AVAILABLE, get_lib_version, str_if_bytes
+from redis.utils import HIREDIS_AVAILABLE, str_if_bytes
 
 from .._parsers import (
     BaseParser,
@@ -199,14 +199,7 @@ def __init__(
         self.client_name = client_name
 
         # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version
-        if driver_info is not None:
-            self.driver_info = driver_info
-        else:
-            # Fallback: create DriverInfo from lib_name and lib_version
-            # Use defaults if not provided
-            name = lib_name if lib_name is not None else "redis-py"
-            version = lib_version if lib_version is not None else get_lib_version()
-            self.driver_info = DriverInfo(name=name, lib_version=version)
+        self.driver_info = resolve_driver_info(driver_info, lib_name, lib_version)
 
         self.credential_provider = credential_provider
         self.password = password

redis/client.py

@@ -40,7 +40,7 @@
     UnixDomainSocketConnection,
 )
 from redis.credentials import CredentialProvider
-from redis.driver_info import DriverInfo
+from redis.driver_info import DriverInfo, resolve_driver_info
 from redis.event import (
     AfterPooledConnectionsInstantiationEvent,
     AfterPubSubConnectionInstantiationEvent,
@@ -64,7 +64,6 @@
 from redis.utils import (
     _set_info_logger,
     deprecated_args,
-    get_lib_version,
     safe_str,
     str_if_bytes,
     truncate_text,
@@ -314,14 +313,9 @@ def __init__(
                 retry_on_error = []
 
             # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version
-            if driver_info is not None:
-                computed_driver_info = driver_info
-            else:
-                # Fallback: create DriverInfo from lib_name and lib_version
-                # Use defaults if not provided
-                name = lib_name if lib_name is not None else "redis-py"
-                version = lib_version if lib_version is not None else get_lib_version()
-                computed_driver_info = DriverInfo(name=name, lib_version=version)
+            computed_driver_info = resolve_driver_info(
+                driver_info, lib_name, lib_version
+            )
 
             kwargs = {
                 "db": db,

redis/connection.py

@@ -35,7 +35,7 @@
 from .auth.token import TokenInterface
 from .backoff import NoBackoff
 from .credentials import CredentialProvider, UsernamePasswordCredentialProvider
-from .driver_info import DriverInfo
+from .driver_info import DriverInfo, resolve_driver_info
 from .event import AfterConnectionReleasedEvent, EventDispatcher
 from .exceptions import (
     AuthenticationError,
@@ -63,7 +63,6 @@
     deprecated_args,
     ensure_string,
     format_error_message,
-    get_lib_version,
     str_if_bytes,
 )
 
@@ -731,14 +730,7 @@ def __init__(
         self.client_name = client_name
 
         # Handle driver_info: if provided, use it; otherwise create from lib_name/lib_version
-        if driver_info is not None:
-            self.driver_info = driver_info
-        else:
-            # Fallback: create DriverInfo from lib_name and lib_version
-            # Use defaults if not provided
-            name = lib_name if lib_name is not None else "redis-py"
-            version = lib_version if lib_version is not None else get_lib_version()
-            self.driver_info = DriverInfo(name=name, lib_version=version)
+        self.driver_info = resolve_driver_info(driver_info, lib_name, lib_version)
 
         self.credential_provider = credential_provider
         self.password = password

redis/driver_info.py

@@ -139,3 +139,38 @@ def formatted_name(self) -> str:
         if not self._upstream:
             return self.name
         return f"{self.name}({';'.join(self._upstream)})"
+
+
+def resolve_driver_info(
+    driver_info: Optional[DriverInfo],
+    lib_name: Optional[str],
+    lib_version: Optional[str],
+) -> DriverInfo:
+    """Resolve driver_info from parameters.
+
+    If driver_info is provided, use it. Otherwise, create DriverInfo from
+    lib_name and lib_version (using defaults if not provided).
+
+    Parameters
+    ----------
+    driver_info : DriverInfo, optional
+        The DriverInfo instance to use
+    lib_name : str, optional
+        The library name (default: "redis-py")
+    lib_version : str, optional
+        The library version (default: auto-detected)
+
+    Returns
+    -------
+    DriverInfo
+        The resolved DriverInfo instance
+    """
+    if driver_info is not None:
+        return driver_info
+
+    # Fallback: create DriverInfo from lib_name and lib_version
+    from redis.utils import get_lib_version
+
+    name = lib_name if lib_name is not None else "redis-py"
+    version = lib_version if lib_version is not None else get_lib_version()
+    return DriverInfo(name=name, lib_version=version)