Simplify Redis connection with a dynamic client class approach

Author: giladwpx

src/common/connection.py

@@ -3,7 +3,7 @@
 import redis
 from redis import Redis
 from redis.cluster import RedisCluster
-from typing import Optional
+from typing import Optional, Type, Union
 from common.config import REDIS_CFG
 
 from common.config import generate_redis_uri
@@ -16,36 +16,25 @@ class RedisConnectionManager:
     def get_connection(cls, decode_responses=True) -> Redis:
         if cls._instance is None:
             try:
-                # Common connection parameters
-                common_params = {
-                    "host": REDIS_CFG["host"],
-                    "port": REDIS_CFG["port"],
-                    "username": REDIS_CFG["username"],
-                    "password": REDIS_CFG["password"],
-                    "ssl": REDIS_CFG["ssl"],
-                    "ssl_ca_path": REDIS_CFG["ssl_ca_path"],
-                    "ssl_keyfile": REDIS_CFG["ssl_keyfile"],
-                    "ssl_certfile": REDIS_CFG["ssl_certfile"],
-                    "ssl_cert_reqs": REDIS_CFG["ssl_cert_reqs"],
-                    "ssl_ca_certs": REDIS_CFG["ssl_ca_certs"],
-                    "decode_responses": decode_responses,
-                    "lib_name": f"redis-py(mcp-server_v{__version__})"
-                }
-
-                if REDIS_CFG["cluster_mode"]:
-                    # Cluster-specific parameters
-                    cluster_params = {
-                        **common_params,
-                        "max_connections_per_node": 10
-                    }
-                    cls._instance = RedisCluster(**cluster_params)
-                else:
-                    # Standalone-specific parameters
-                    standalone_params = {
-                        **common_params,
-                        "max_connections": 10
-                    }
-                    cls._instance = redis.Redis(**standalone_params)
+                # Select the appropriate Redis client class
+                redis_class: Type[Union[Redis, RedisCluster]] = RedisCluster if REDIS_CFG["cluster_mode"] else redis.Redis
+                
+                # Create the connection with appropriate parameters
+                cls._instance = redis_class(
+                    host=REDIS_CFG["host"],
+                    port=REDIS_CFG["port"],
+                    username=REDIS_CFG["username"],
+                    password=REDIS_CFG["password"],
+                    ssl=REDIS_CFG["ssl"],
+                    ssl_ca_path=REDIS_CFG["ssl_ca_path"],
+                    ssl_keyfile=REDIS_CFG["ssl_keyfile"],
+                    ssl_certfile=REDIS_CFG["ssl_certfile"],
+                    ssl_cert_reqs=REDIS_CFG["ssl_cert_reqs"],
+                    ssl_ca_certs=REDIS_CFG["ssl_ca_certs"],
+                    decode_responses=decode_responses,
+                    lib_name=f"redis-py(mcp-server_v{__version__})",
+                    **({"max_connections_per_node": 10} if REDIS_CFG["cluster_mode"] else {"max_connections": 10})
+                )
 
             except redis.exceptions.ConnectionError:
                 print("Failed to connect to Redis server", file=sys.stderr)