linting

Signed-off-by: Keval Mahajan <mahajankeval23@gmail.com>

Author: kevalmahajan

mcpgateway/services/server_service.py

@@ -221,7 +221,7 @@ def _convert_server_to_read(self, server: DbServer) -> ServerRead:
         """
         server_dict = server.__dict__.copy()
         server_dict.pop("_sa_instance_state", None)
-        
+
         # Compute aggregated metrics from server.metrics in a single pass; default to 0/None when no records exist.
         total = 0
         successful = 0
@@ -230,27 +230,27 @@ def _convert_server_to_read(self, server: DbServer) -> ServerRead:
         max_rt = None
         sum_rt = 0.0
         last_time = None
-        
+
         if hasattr(server, "metrics") and server.metrics:
             for m in server.metrics:
                 total += 1
                 if m.is_success:
                     successful += 1
                 else:
                     failed += 1
-                
+
                 # Track min/max response times
                 if min_rt is None or m.response_time < min_rt:
                     min_rt = m.response_time
                 if max_rt is None or m.response_time > max_rt:
                     max_rt = m.response_time
-                
+
                 sum_rt += m.response_time
-                
+
                 # Track last execution time
                 if last_time is None or m.timestamp > last_time:
                     last_time = m.timestamp
-        
+
         failure_rate = (failed / total) if total > 0 else 0.0
         avg_rt = (sum_rt / total) if total > 0 else None
 
@@ -593,17 +593,14 @@ async def list_servers(self, db: Session, include_inactive: bool = False, tags:
             query = query.where(json_contains_expr(db, DbServer.tags, tags, match_any=True))
 
         servers = db.execute(query).scalars().all()
-        
+
         # Fetch all team names
         team_ids = [s.team_id for s in servers if s.team_id]
         team_map = {}
         if team_ids:
-            teams = db.execute(
-                select(DbEmailTeam.id, DbEmailTeam.name)
-                .where(DbEmailTeam.id.in_(team_ids), DbEmailTeam.is_active.is_(True))
-            ).all()
+            teams = db.execute(select(DbEmailTeam.id, DbEmailTeam.name).where(DbEmailTeam.id.in_(team_ids), DbEmailTeam.is_active.is_(True))).all()
             team_map = {team.id: team.name for team in teams}
-        
+
         result = []
         for s in servers:
             s.team = team_map.get(s.team_id) if s.team_id else None
@@ -676,17 +673,14 @@ async def list_servers_for_user(
         query = query.offset(skip).limit(limit)
 
         servers = db.execute(query).scalars().all()
-        
-        # Fetch all team names 
+
+        # Fetch all team names
         server_team_ids = [s.team_id for s in servers if s.team_id]
         team_map = {}
         if server_team_ids:
-            teams = db.execute(
-                select(DbEmailTeam.id, DbEmailTeam.name)
-                .where(DbEmailTeam.id.in_(server_team_ids), DbEmailTeam.is_active.is_(True))
-            ).all()
+            teams = db.execute(select(DbEmailTeam.id, DbEmailTeam.name).where(DbEmailTeam.id.in_(server_team_ids), DbEmailTeam.is_active.is_(True))).all()
             team_map = {team.id: team.name for team in teams}
-        
+
         result = []
         for s in servers:
             s.team = team_map.get(s.team_id) if s.team_id else None

mcpgateway/services/tool_service.py

@@ -287,7 +287,6 @@ async def shutdown(self) -> None:
         await self._event_service.shutdown()
         logger.info("Tool service shutdown complete")
 
-
     async def get_top_tools(self, db: Session, limit: Optional[int] = 5) -> List[TopPerformer]:
         """Retrieve the top-performing tools based on execution count.
 
@@ -309,21 +308,13 @@ async def get_top_tools(self, db: Session, limit: Optional[int] = 5) -> List[Top
                 - last_execution: Timestamp of the last execution, or None if no metrics.
         """
 
-        success_rate = case(
-            (
-                func.count(ToolMetric.id) > 0,
-                func.sum(
-                    case((ToolMetric.is_success.is_(True), 1), else_=0)
-                ).cast(Float) * 100 / func.count(ToolMetric.id)
-            ),
-            else_=None
-        )
+        success_rate = case((func.count(ToolMetric.id) > 0, func.sum(case((ToolMetric.is_success.is_(True), 1), else_=0)).cast(Float) * 100 / func.count(ToolMetric.id)), else_=None)  # pylint: disable=not-callable
 
         query = (
             select(
                 DbTool.id,
                 DbTool.name,
-                func.count(ToolMetric.id).label("execution_count"),
+                func.count(ToolMetric.id).label("execution_count"),  # pylint: disable=not-callable
                 func.avg(ToolMetric.response_time).label("avg_response_time"),
                 success_rate.label("success_rate"),
                 func.max(ToolMetric.timestamp).label("last_execution"),
@@ -390,7 +381,7 @@ def _convert_tool_to_read(self, tool: DbTool, include_metrics: bool = True) -> T
                     "token": "********" if decoded_auth_value["Authorization"] else None,
                 }
             elif tool.auth_type == "authheaders":
-                # Get first key 
+                # Get first key
                 first_key = next(iter(decoded_auth_value))
                 tool_dict["auth"] = {
                     "auth_type": "authheaders",
@@ -414,7 +405,6 @@ def _convert_tool_to_read(self, tool: DbTool, include_metrics: bool = True) -> T
         tool_dict["team"] = getattr(tool, "team", None)
         return ToolRead.model_validate(tool_dict)
 
-
     async def _record_tool_metric(self, db: Session, tool: DbTool, start_time: float, success: bool, error_message: Optional[str]) -> None:
         """
         Records a metric for a tool invocation.
@@ -958,14 +948,14 @@ async def list_tools_for_user(
         # query = query.offset(skip).limit(limit)
 
         tools = db.execute(query).scalars().all()
-        
+
         # Batch fetch team names for all tools at once
         tool_team_ids = {getattr(t, "team_id", None) for t in tools if getattr(t, "team_id", None)}
         team_name_map = {}
         if tool_team_ids:
             teams = db.query(EmailTeam.id, EmailTeam.name).filter(EmailTeam.id.in_(tool_team_ids), EmailTeam.is_active.is_(True)).all()
             team_name_map = {team.id: team.name for team in teams}
-        
+
         result = []
         for t in tools:
             team_name = team_name_map.get(getattr(t, "team_id", None))