Fix lint issues, rebae

Signed-off-by: Mihai Criveti <[email protected]>

Author: crivetimihai

.pylintrc

@@ -575,7 +575,7 @@ contextmanager-decorators=contextlib.contextmanager
 # List of members which are set dynamically and missed by pylint inference
 # system, and so shouldn't trigger E1101 when accessed. Python regular
 # expressions are accepted.
-generated-members=
+generated-members=sqlalchemy.func.*
 
 # Tells whether to warn about missing members when the owner of the attribute
 # is inferred to be None.

mcpgateway/schemas.py

@@ -590,7 +590,7 @@ def prevent_manual_mcp_creation(cls, values: Dict[str, Any]) -> Dict[str, Any]:
         """
         integration_type = values.get("integration_type")
         if integration_type == "MCP":
-            raise ValueError("Cannot manually create MCP tools. Add MCP servers via the Gateways interface - " "tools will be auto-discovered and registered with integration_type='MCP'.")
+            raise ValueError("Cannot manually create MCP tools. Add MCP servers via the Gateways interface - tools will be auto-discovered and registered with integration_type='MCP'.")
         return values
 
 

mcpgateway/services/logging_service.py

@@ -123,7 +123,7 @@ async def initialize(self) -> None:
             try:
                 root_logger.addHandler(_get_file_handler())
                 if settings.log_rotation_enabled:
-                    logging.info(f"File logging enabled with rotation: {settings.log_folder or '.'}/{settings.log_file} " f"(max: {settings.log_max_size_mb}MB, backups: {settings.log_backup_count})")
+                    logging.info(f"File logging enabled with rotation: {settings.log_folder or '.'}/{settings.log_file} (max: {settings.log_max_size_mb}MB, backups: {settings.log_backup_count})")
                 else:
                     logging.info(f"File logging enabled (no rotation): {settings.log_folder or '.'}/{settings.log_file}")
             except Exception as e:

mcpgateway/services/prompt_service.py

@@ -23,7 +23,7 @@
 
 # Third-Party
 from jinja2 import Environment, meta, select_autoescape
-from sqlalchemy import case, delete, desc, func, not_, select
+from sqlalchemy import case, delete, desc, func, not_, select, Float
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -35,6 +35,7 @@
 from mcpgateway.plugins import GlobalContext, PluginManager, PluginViolationError, PromptPosthookPayload, PromptPrehookPayload
 from mcpgateway.schemas import PromptCreate, PromptRead, PromptUpdate, TopPerformer
 from mcpgateway.services.logging_service import LoggingService
+from mcpgateway.utils.metrics_common import build_top_performers
 
 # Initialize logging service first
 logging_service = LoggingService()
@@ -165,9 +166,13 @@ async def get_top_prompts(self, db: Session, limit: int = 5) -> List[TopPerforme
                 DbPrompt.name,
                 func.count(PromptMetric.id).label("execution_count"),  # pylint: disable=not-callable
                 func.avg(PromptMetric.response_time).label("avg_response_time"),  # pylint: disable=not-callable
-                case((func.count(PromptMetric.id) > 0, func.sum(case((PromptMetric.is_success, 1), else_=0)) / func.count(PromptMetric.id) * 100), else_=None).label(
-                    "success_rate"
-                ),  # pylint: disable=not-callable
+                case(
+                    (
+                        func.count(PromptMetric.id) > 0,  # pylint: disable=not-callable
+                        func.sum(case((PromptMetric.is_success.is_(True), 1), else_=0)).cast(Float) / func.count(PromptMetric.id) * 100,  # pylint: disable=not-callable
+                    ),
+                    else_=None,
+                ).label("success_rate"),
                 func.max(PromptMetric.timestamp).label("last_execution"),  # pylint: disable=not-callable
             )
             .outerjoin(PromptMetric)
@@ -177,17 +182,7 @@ async def get_top_prompts(self, db: Session, limit: int = 5) -> List[TopPerforme
             .all()
         )
 
-        return [
-            TopPerformer(
-                id=result.id,
-                name=result.name,
-                execution_count=result.execution_count or 0,
-                avg_response_time=float(result.avg_response_time) if result.avg_response_time else None,
-                success_rate=float(result.success_rate) if result.success_rate else None,
-                last_execution=result.last_execution,
-            )
-            for result in results
-        ]
+        return build_top_performers(results)
 
     def _convert_db_prompt(self, db_prompt: DbPrompt) -> Dict[str, Any]:
         """

mcpgateway/services/resource_service.py

@@ -33,7 +33,7 @@
 
 # Third-Party
 import parse
-from sqlalchemy import case, delete, desc, func, not_, select
+from sqlalchemy import case, delete, desc, func, not_, select, Float
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -45,6 +45,7 @@
 from mcpgateway.models import ResourceContent, ResourceTemplate, TextContent
 from mcpgateway.schemas import ResourceCreate, ResourceMetrics, ResourceRead, ResourceSubscription, ResourceUpdate, TopPerformer
 from mcpgateway.services.logging_service import LoggingService
+from mcpgateway.utils.metrics_common import build_top_performers
 
 # Initialize logging service first
 logging_service = LoggingService()
@@ -138,9 +139,13 @@ async def get_top_resources(self, db: Session, limit: int = 5) -> List[TopPerfor
                 DbResource.uri.label("name"),  # Using URI as the name field for TopPerformer
                 func.count(ResourceMetric.id).label("execution_count"),  # pylint: disable=not-callable
                 func.avg(ResourceMetric.response_time).label("avg_response_time"),  # pylint: disable=not-callable
-                case((func.count(ResourceMetric.id) > 0, func.sum(case((ResourceMetric.is_success, 1), else_=0)) / func.count(ResourceMetric.id) * 100), else_=None).label(
-                    "success_rate"
-                ),  # pylint: disable=not-callable
+                case(
+                    (
+                        func.count(ResourceMetric.id) > 0,  # pylint: disable=not-callable
+                        func.sum(case((ResourceMetric.is_success.is_(True), 1), else_=0)).cast(Float) / func.count(ResourceMetric.id) * 100,  # pylint: disable=not-callable
+                    ),
+                    else_=None,
+                ).label("success_rate"),
                 func.max(ResourceMetric.timestamp).label("last_execution"),  # pylint: disable=not-callable
             )
             .outerjoin(ResourceMetric)
@@ -150,17 +155,7 @@ async def get_top_resources(self, db: Session, limit: int = 5) -> List[TopPerfor
             .all()
         )
 
-        return [
-            TopPerformer(
-                id=result.id,
-                name=result.name,
-                execution_count=result.execution_count or 0,
-                avg_response_time=float(result.avg_response_time) if result.avg_response_time else None,
-                success_rate=float(result.success_rate) if result.success_rate else None,
-                last_execution=result.last_execution,
-            )
-            for result in results
-        ]
+        return build_top_performers(results)
 
     def _convert_resource_to_read(self, resource: DbResource) -> ResourceRead:
         """

mcpgateway/services/server_service.py

@@ -19,7 +19,7 @@
 
 # Third-Party
 import httpx
-from sqlalchemy import case, delete, desc, func, not_, select
+from sqlalchemy import case, delete, desc, func, not_, select, Float
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -32,6 +32,7 @@
 from mcpgateway.db import Tool as DbTool
 from mcpgateway.schemas import ServerCreate, ServerMetrics, ServerRead, ServerUpdate, TopPerformer
 from mcpgateway.services.logging_service import LoggingService
+from mcpgateway.utils.metrics_common import build_top_performers
 
 # Initialize logging service first
 logging_service = LoggingService()
@@ -153,9 +154,13 @@ async def get_top_servers(self, db: Session, limit: int = 5) -> List[TopPerforme
                 DbServer.name,
                 func.count(ServerMetric.id).label("execution_count"),  # pylint: disable=not-callable
                 func.avg(ServerMetric.response_time).label("avg_response_time"),  # pylint: disable=not-callable
-                case((func.count(ServerMetric.id) > 0, func.sum(case((ServerMetric.is_success, 1), else_=0)) / func.count(ServerMetric.id) * 100), else_=None).label(
-                    "success_rate"
-                ),  # pylint: disable=not-callable
+                case(
+                    (
+                        func.count(ServerMetric.id) > 0,  # pylint: disable=not-callable
+                        func.sum(case((ServerMetric.is_success.is_(True), 1), else_=0)).cast(Float) / func.count(ServerMetric.id) * 100,  # pylint: disable=not-callable
+                    ),
+                    else_=None,
+                ).label("success_rate"),
                 func.max(ServerMetric.timestamp).label("last_execution"),  # pylint: disable=not-callable
             )
             .outerjoin(ServerMetric)
@@ -165,17 +170,7 @@ async def get_top_servers(self, db: Session, limit: int = 5) -> List[TopPerforme
             .all()
         )
 
-        return [
-            TopPerformer(
-                id=result.id,
-                name=result.name,
-                execution_count=result.execution_count or 0,
-                avg_response_time=float(result.avg_response_time) if result.avg_response_time else None,
-                success_rate=float(result.success_rate) if result.success_rate else None,
-                last_execution=result.last_execution,
-            )
-            for result in results
-        ]
+        return build_top_performers(results)
 
     def _convert_server_to_read(self, server: DbServer) -> ServerRead:
         """

mcpgateway/services/tool_service.py

@@ -28,7 +28,7 @@
 from mcp import ClientSession
 from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamablehttp_client
-from sqlalchemy import case, delete, desc, func, not_, select
+from sqlalchemy import case, delete, desc, func, not_, select, Float
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -44,6 +44,7 @@
 from mcpgateway.schemas import ToolCreate, ToolRead, ToolUpdate, TopPerformer
 from mcpgateway.services.logging_service import LoggingService
 from mcpgateway.utils.create_slug import slugify
+from mcpgateway.utils.metrics_common import build_top_performers
 from mcpgateway.utils.passthrough_headers import get_passthrough_headers
 from mcpgateway.utils.retry_manager import ResilientHttpClient
 from mcpgateway.utils.services_auth import decode_auth
@@ -215,9 +216,13 @@ async def get_top_tools(self, db: Session, limit: int = 5) -> List[TopPerformer]
                 DbTool.name,
                 func.count(ToolMetric.id).label("execution_count"),  # pylint: disable=not-callable
                 func.avg(ToolMetric.response_time).label("avg_response_time"),  # pylint: disable=not-callable
-                case((func.count(ToolMetric.id) > 0, func.sum(case((ToolMetric.is_success, 1), else_=0)) / func.count(ToolMetric.id) * 100), else_=None).label(
-                    "success_rate"
-                ),  # pylint: disable=not-callable
+                case(
+                    (
+                        func.count(ToolMetric.id) > 0,  # pylint: disable=not-callable
+                        func.sum(case((ToolMetric.is_success.is_(True), 1), else_=0)).cast(Float) / func.count(ToolMetric.id) * 100,  # pylint: disable=not-callable
+                    ),
+                    else_=None,
+                ).label("success_rate"),
                 func.max(ToolMetric.timestamp).label("last_execution"),  # pylint: disable=not-callable
             )
             .outerjoin(ToolMetric)
@@ -227,17 +232,7 @@ async def get_top_tools(self, db: Session, limit: int = 5) -> List[TopPerformer]
             .all()
         )
 
-        return [
-            TopPerformer(
-                id=result.id,
-                name=result.name,
-                execution_count=result.execution_count or 0,
-                avg_response_time=float(result.avg_response_time) if result.avg_response_time else None,
-                success_rate=float(result.success_rate) if result.success_rate else None,
-                last_execution=result.last_execution,
-            )
-            for result in results
-        ]
+        return build_top_performers(results)
 
     def _convert_tool_to_read(self, tool: DbTool) -> ToolRead:
         """Converts a DbTool instance into a ToolRead model, including aggregated metrics and
@@ -734,18 +729,21 @@ async def invoke_tool(self, db: Session, name: str, arguments: Dict[str, Any], r
                 # Handle 204 No Content responses that have no body
                 if response.status_code == 204:
                     tool_result = ToolResult(content=[TextContent(type="text", text="Request completed successfully (No Content)")])
+                    # Mark as successful only after all operations complete successfully
+                    success = True
                 elif response.status_code not in [200, 201, 202, 206]:
                     result = response.json()
                     tool_result = ToolResult(
                         content=[TextContent(type="text", text=str(result["error"]) if "error" in result else "Tool error encountered")],
                         is_error=True,
                     )
+                    # Don't mark as successful for error responses - success remains False
                 else:
                     result = response.json()
                     filtered_response = extract_using_jq(result, tool.jsonpath_filter)
                     tool_result = ToolResult(content=[TextContent(type="text", text=json.dumps(filtered_response, indent=2))])
-
-                success = True
+                    # Mark as successful only after all operations complete successfully
+                    success = True
             elif tool.integration_type == "MCP":
                 transport = tool.request_type.lower()
                 gateway = db.execute(select(DbGateway).where(DbGateway.id == tool.gateway_id).where(DbGateway.enabled)).scalar_one_or_none()
@@ -795,9 +793,10 @@ async def connect_to_streamablehttp_server(server_url: str):
                     tool_call_result = await connect_to_streamablehttp_server(tool_gateway.url)
                 content = tool_call_result.model_dump(by_alias=True).get("content", [])
 
-                success = True
                 filtered_response = extract_using_jq(content, tool.jsonpath_filter)
                 tool_result = ToolResult(content=filtered_response)
+                # Mark as successful only after all operations complete successfully
+                success = True
             else:
                 tool_result = ToolResult(content=[TextContent(type="text", text="Invalid tool type")])
 

mcpgateway/utils/metrics_common.py

@@ -0,0 +1,65 @@
+# -*- coding: utf-8 -*-
+"""
+Common utilities for metrics handling across service modules.
+
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: Mihai Criveti
+"""
+
+# Standard
+from typing import List
+
+# First-Party
+from mcpgateway.schemas import TopPerformer
+
+
+def build_top_performers(results: List) -> List[TopPerformer]:
+    """
+    Convert database query results to TopPerformer objects.
+
+    This utility function eliminates code duplication across service modules
+    that need to convert database query results with metrics into TopPerformer objects.
+
+    Args:
+        results: List of database query results, each containing:
+            - id: Entity ID
+            - name: Entity name
+            - execution_count: Total executions
+            - avg_response_time: Average response time
+            - success_rate: Success rate percentage
+            - last_execution: Last execution timestamp
+
+    Returns:
+        List[TopPerformer]: List of TopPerformer objects with proper type conversions
+
+    Examples:
+        >>> from unittest.mock import MagicMock
+        >>> result = MagicMock()
+        >>> result.id = 1
+        >>> result.name = "test"
+        >>> result.execution_count = 10
+        >>> result.avg_response_time = 1.5
+        >>> result.success_rate = 85.0
+        >>> result.last_execution = None
+        >>> performers = build_top_performers([result])
+        >>> len(performers)
+        1
+        >>> performers[0].id
+        1
+        >>> performers[0].execution_count
+        10
+        >>> performers[0].avg_response_time
+        1.5
+    """
+    return [
+        TopPerformer(
+            id=result.id,
+            name=result.name,
+            execution_count=result.execution_count or 0,
+            avg_response_time=float(result.avg_response_time) if result.avg_response_time else None,
+            success_rate=float(result.success_rate) if result.success_rate else None,
+            last_execution=result.last_execution,
+        )
+        for result in results
+    ]