session pooling pylint fixes

Signed-off-by: Mohan Lakshmaiah <mohalaks@in.ibm.com>

Author: MohanLakshmaiah

mcpgateway/cache/session_pool.py

@@ -10,7 +10,7 @@
 import asyncio
 import time
 import logging
-from typing import Dict, Optional, List, Any
+from typing import Dict, Optional, Any
 from enum import Enum
 from dataclasses import dataclass
 from mcpgateway.cache.session_registry import SessionRegistry
@@ -22,11 +22,13 @@
 
 logger = logging.getLogger(__name__)
 
+
 class TransportType(Enum):
     """Enumeration of supported transport types."""
     SSE = "sse"
     WEBSOCKET = "websocket"
 
+
 @dataclass
 class PoolKey:
     """Structured key for session pooling with proper hashing."""
@@ -45,6 +47,7 @@ def __eq__(self, other):
                 self.server_id == other.server_id and
                 self.transport_type == other.transport_type)
 
+
 class PooledSession:
     """Wrapper around transport for pooling and metrics tracking."""
     def __init__(self, transport: Transport, user_id: str, server_id: str, transport_type: TransportType):
@@ -118,14 +121,14 @@ def __init__(self, session_registry: SessionRegistry):
         if settings.session_pooling_enabled:
             self._start_cleanup_task()
             logger.info("Session pool initialized with cleanup interval=%s sec",
-                       settings.session_pool_cleanup_interval)
+                        settings.session_pool_cleanup_interval)
 
     # --------------------------------------------------------------------------
     # Core pooling logic with multi-transport support
     # --------------------------------------------------------------------------
 
     async def get_or_create_session(self, user_id: str, server_id: str, base_url: str,
-                                  transport_type: TransportType) -> Transport:
+                                    transport_type: TransportType) -> Transport:
         """Get an existing session for (user, server, transport) or create a new one."""
         if not settings.session_pooling_enabled:
             logger.debug("Session pooling disabled, creating fresh session.")
@@ -160,7 +163,7 @@ async def get_or_create_session(self, user_id: str, server_id: str, base_url: st
             return new_session.transport
 
     async def _create_new_session(self, user_id: str, server_id: str, base_url: str,
-                                transport_type: TransportType) -> PooledSession:
+                                  transport_type: TransportType) -> PooledSession:
         """Create and register a brand new transport session."""
         try:
             # Create transport instance based on type
@@ -185,7 +188,10 @@ async def _create_new_session(self, user_id: str, server_id: str, base_url: str,
             self._metrics["created"] += 1
 
             logger.info("Created new %s session for user=%s server=%s (session_id=%s)",
-                       transport_type.value, user_id, server_id, transport.session_id)
+                        transport_type.value, 
+                        user_id, 
+                        server_id, 
+                        transport.session_id)
             return session
 
         except Exception as e:
@@ -210,14 +216,14 @@ async def _is_session_valid(self, session: PooledSession) -> bool:
 
             if session.idle_time > settings.session_pool_max_idle_time:
                 logger.debug("Session %s idle too long (idle_time=%s).",
-                           session.transport.session_id, session.idle_time)
+                             session.transport.session_id, session.idle_time)
                 return False
 
             # Additional transport-specific validation
             if hasattr(session.transport, 'validate_session'):
                 if not await session.transport.validate_session():
                     logger.debug("Session %s failed transport-specific validation.",
-                               session.transport.session_id)
+                                 session.transport.session_id)
                     return False
 
             return True
@@ -331,7 +337,7 @@ def _start_cleanup_task(self):
         if not self._cleanup_task or self._cleanup_task.done():
             self._cleanup_task = asyncio.create_task(self.cleanup_expired_sessions())
             logger.info("Session cleanup task started (interval=%s).",
-                       settings.session_pool_cleanup_interval)
+                        settings.session_pool_cleanup_interval)
 
     async def shutdown(self):
         """Gracefully stop the session pool and cleanup task."""
@@ -349,4 +355,4 @@ async def shutdown(self):
                 await self._cleanup_session(pool_key, session)
             self._pool.clear()
 
-        logger.info("Session pool shut down. Final metrics: %s", self._metrics)
+        logger.info("Session pool shut down. Final metrics: %s", self._metrics)

mcpgateway/cache/session_registry.py

@@ -1567,6 +1567,7 @@ async def generate_response(self, message: Dict[str, Any], transport: SSETranspo
     # ------------------------------
     # Observability
     # ------------------------------
+    
     def get_metrics(self) -> Dict[str, int]:
         """
         Retrieve internal metrics counters for the session registry.
@@ -1577,19 +1578,3 @@ def get_metrics(self) -> Dict[str, int]:
             values are their current counts (int).
         """
         return dict(self._metrics)
-
-    def get_session_sync(self, session_id: str) -> Optional[SSETransport]:
-        """
-        Get session transport synchronously from local cache only.
-
-        This is a non-blocking method that only checks the local cache,
-        not the distributed backend. Use this when you need quick access
-        and know the session should be local.
-
-        Args:
-            session_id: Session identifier to look up.
-
-        Returns:
-            SSETransport object if found in local cache, None otherwise.
-        """
-        return self._sessions.get(session_id)

mcpgateway/config.py

@@ -63,7 +63,7 @@
 import jq
 from jsonpath_ng.ext import parse
 from jsonpath_ng.jsonpath import JSONPath
-from pydantic import Field, field_validator, HttpUrl, model_validator, PositiveInt, SecretStr, BaseModel
+from pydantic import Field, field_validator, HttpUrl, model_validator, PositiveInt, SecretStr
 from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
 
 # Only configure basic logging if no handlers exist yet

mcpgateway/services/server_service.py

@@ -1221,7 +1221,7 @@ async def should_use_pooling(self, db: Session, server_id: str) -> bool:
             True
         """
         strategy = await self.get_session_strategy(db, server_id)
-        return strategy in ["user-server", "global", "enabled"] # Handle potential naming variations like "user_server"
+        return strategy in ["user-server", "global", "enabled"]   # Handle potential naming variations like "user_server"
 
     # --- Metrics ---
     async def aggregate_metrics(self, db: Session) -> ServerMetrics:

mcpgateway/transports/base.py

@@ -137,4 +137,6 @@ async def is_connected(self) -> bool:
 
     async def validate_session(self) -> bool:
         """Validate session is still usable."""
-        return await self.is_connected()
+        return await self.is_connected()
+    
+    

mcpgateway/transports/sse_transport.py

@@ -14,7 +14,6 @@
 from datetime import datetime
 import json
 from typing import Any, AsyncGenerator, Dict
-import uuid
 import time
 
 # Third-Party