Fix lint issues, rebased

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

Author: crivetimihai

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, Float
+from sqlalchemy import case, delete, desc, Float, func, not_, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -169,7 +169,7 @@ async def get_top_prompts(self, db: Session, limit: int = 5) -> List[TopPerforme
                 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
+                        func.sum(case((PromptMetric.is_success == 1, 1), else_=0)).cast(Float) / func.count(PromptMetric.id) * 100,  # pylint: disable=not-callable
                     ),
                     else_=None,
                 ).label("success_rate"),
@@ -1050,8 +1050,8 @@ async def aggregate_metrics(self, db: Session) -> Dict[str, Any]:
         """
 
         total = db.execute(select(func.count(PromptMetric.id))).scalar() or 0  # pylint: disable=not-callable
-        successful = db.execute(select(func.count(PromptMetric.id)).where(PromptMetric.is_success)).scalar() or 0  # pylint: disable=not-callable
-        failed = db.execute(select(func.count(PromptMetric.id)).where(not_(PromptMetric.is_success))).scalar() or 0  # pylint: disable=not-callable
+        successful = db.execute(select(func.count(PromptMetric.id)).where(PromptMetric.is_success == 1)).scalar() or 0  # pylint: disable=not-callable
+        failed = db.execute(select(func.count(PromptMetric.id)).where(PromptMetric.is_success == 0)).scalar() or 0  # pylint: disable=not-callable
         failure_rate = failed / total if total > 0 else 0.0
         min_rt = db.execute(select(func.min(PromptMetric.response_time))).scalar()
         max_rt = db.execute(select(func.max(PromptMetric.response_time))).scalar()

mcpgateway/services/resource_service.py

@@ -33,7 +33,7 @@
 
 # Third-Party
 import parse
-from sqlalchemy import case, delete, desc, func, not_, select, Float
+from sqlalchemy import case, delete, desc, Float, func, not_, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -142,7 +142,7 @@ async def get_top_resources(self, db: Session, limit: int = 5) -> List[TopPerfor
                 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
+                        func.sum(case((ResourceMetric.is_success == 1, 1), else_=0)).cast(Float) / func.count(ResourceMetric.id) * 100,  # pylint: disable=not-callable
                     ),
                     else_=None,
                 ).label("success_rate"),
@@ -1040,9 +1040,9 @@ async def aggregate_metrics(self, db: Session) -> ResourceMetrics:
         """
         total_executions = db.execute(select(func.count()).select_from(ResourceMetric)).scalar() or 0  # pylint: disable=not-callable
 
-        successful_executions = db.execute(select(func.count()).select_from(ResourceMetric).where(ResourceMetric.is_success)).scalar() or 0  # pylint: disable=not-callable
+        successful_executions = db.execute(select(func.count()).select_from(ResourceMetric).where(ResourceMetric.is_success == 1)).scalar() or 0  # pylint: disable=not-callable
 
-        failed_executions = db.execute(select(func.count()).select_from(ResourceMetric).where(not_(ResourceMetric.is_success))).scalar() or 0  # pylint: disable=not-callable
+        failed_executions = db.execute(select(func.count()).select_from(ResourceMetric).where(ResourceMetric.is_success == 0)).scalar() or 0  # pylint: disable=not-callable
 
         min_response_time = db.execute(select(func.min(ResourceMetric.response_time))).scalar()
 

mcpgateway/services/server_service.py

@@ -19,7 +19,7 @@
 
 # Third-Party
 import httpx
-from sqlalchemy import case, delete, desc, func, not_, select, Float
+from sqlalchemy import case, delete, desc, Float, func, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -157,7 +157,7 @@ async def get_top_servers(self, db: Session, limit: int = 5) -> List[TopPerforme
                 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
+                        func.sum(case((ServerMetric.is_success == 1, 1), else_=0)).cast(Float) / func.count(ServerMetric.id) * 100,  # pylint: disable=not-callable
                     ),
                     else_=None,
                 ).label("success_rate"),
@@ -833,9 +833,9 @@ async def aggregate_metrics(self, db: Session) -> ServerMetrics:
         """
         total_executions = db.execute(select(func.count()).select_from(ServerMetric)).scalar() or 0  # pylint: disable=not-callable
 
-        successful_executions = db.execute(select(func.count()).select_from(ServerMetric).where(ServerMetric.is_success)).scalar() or 0  # pylint: disable=not-callable
+        successful_executions = db.execute(select(func.count()).select_from(ServerMetric).where(ServerMetric.is_success == 1)).scalar() or 0  # pylint: disable=not-callable
 
-        failed_executions = db.execute(select(func.count()).select_from(ServerMetric).where(not_(ServerMetric.is_success))).scalar() or 0  # pylint: disable=not-callable
+        failed_executions = db.execute(select(func.count()).select_from(ServerMetric).where(ServerMetric.is_success == 0)).scalar() or 0  # pylint: disable=not-callable
 
         min_response_time = db.execute(select(func.min(ServerMetric.response_time))).scalar()
 

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, Float
+from sqlalchemy import case, delete, desc, Float, func, not_, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -219,7 +219,7 @@ async def get_top_tools(self, db: Session, limit: int = 5) -> List[TopPerformer]
                 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
+                        func.sum(case((ToolMetric.is_success == 1, 1), else_=0)).cast(Float) / func.count(ToolMetric.id) * 100,  # pylint: disable=not-callable
                     ),
                     else_=None,
                 ).label("success_rate"),
@@ -1111,8 +1111,8 @@ async def aggregate_metrics(self, db: Session) -> Dict[str, Any]:
         """
 
         total = db.execute(select(func.count(ToolMetric.id))).scalar() or 0  # pylint: disable=not-callable
-        successful = db.execute(select(func.count(ToolMetric.id)).where(ToolMetric.is_success)).scalar() or 0  # pylint: disable=not-callable
-        failed = db.execute(select(func.count(ToolMetric.id)).where(not_(ToolMetric.is_success))).scalar() or 0  # pylint: disable=not-callable
+        successful = db.execute(select(func.count(ToolMetric.id)).where(ToolMetric.is_success == 1)).scalar() or 0  # pylint: disable=not-callable
+        failed = db.execute(select(func.count(ToolMetric.id)).where(ToolMetric.is_success == 0)).scalar() or 0  # pylint: disable=not-callable
         failure_rate = failed / total if total > 0 else 0.0
         min_rt = db.execute(select(func.min(ToolMetric.response_time))).scalar()
         max_rt = db.execute(select(func.max(ToolMetric.response_time))).scalar()