Fixed Linting changes

Author: pjain05

pyproject.toml

@@ -33,6 +33,13 @@ mcp-bzm-apitest = "main:main"
 packages = ["src", "src.common", "src.config", "src.formatters", "src.models", "src.tools"]
 py-modules = ["main"]
 
-[tool.pycodestyle]
+[tool.flake8]
 max-line-length = 108
+extend-ignore = ["E203", "W503"]
+
+[tool.black]
+line-length = 108
+
+[tool.isort]
+profile = "black"
 

src/common/api_client.py

@@ -4,77 +4,74 @@
 
 import os
 import platform
-from typing import Optional, Callable
+from typing import Callable, Optional
+
 import httpx
 
 from src.config.defaults import BZM_APIM_BASE_URL
 from src.config.token import BzmApimToken
 from src.config.version import __version__
 from src.models import BaseResult
 
-
-so = platform.system()       # "Windows", "Linux", "Darwin"
+so = platform.system()  # "Windows", "Linux", "Darwin"
 version = platform.version()  # kernel / build version
 release = platform.release()  # ex. "10", "5.15.0-76-generic"
 machine = platform.machine()  # ex. "x86_64", "AMD64", "arm64"
 
 ua_part = f"{so} {release}; {machine}"
 
 
-async def api_request(token: Optional[BzmApimToken], method: str, endpoint: str,
-                      result_formatter: Callable = None,
-                      result_formatter_params: Optional[dict] = None,
-                      **kwargs) -> BaseResult:
+async def api_request(
+    token: Optional[BzmApimToken],
+    method: str,
+    endpoint: str,
+    result_formatter: Callable = None,
+    result_formatter_params: Optional[dict] = None,
+    **kwargs,
+) -> BaseResult:
     """
     Make an authenticated request to the BlazeMeter APIM APIs.
     Handles authentication errors gracefully.
     """
     if not token:
         return BaseResult(
             error="No API token. Set BZM_API_TEST_TOKEN env var with the token or BZM_API_TEST_TOKEN_FILE "
-                  "with the file path or BZM_API_TEST_TOKEN secrets in docker catalog configuration.")
+            "with the file path or BZM_API_TEST_TOKEN secrets in docker catalog configuration."
+        )
 
     headers = kwargs.pop("headers", {})
     headers["Authorization"] = f"Bearer {token}"
     headers["User-Agent"] = f"bzm-apitest-mcp/{__version__} ({ua_part})"
 
-    timeout = httpx.Timeout(
-        connect=15.0,
-        read=60.0,
-        write=15.0,
-        pool=60.0
-    )
+    timeout = httpx.Timeout(connect=15.0, read=60.0, write=15.0, pool=60.0)
 
-    async with (httpx.AsyncClient(base_url=BZM_APIM_BASE_URL, timeout=timeout) as client):
+    async with httpx.AsyncClient(base_url=BZM_APIM_BASE_URL, timeout=timeout) as client:
         try:
             resp = await client.request(method, endpoint, headers=headers, **kwargs)
             resp.raise_for_status()
             response_dict = resp.json()
             result = response_dict.get("data", [])
             default_total = 0
-            if not isinstance(
-                    result, list):  # Generalize result always as a list
+            if not isinstance(result, list):  # Generalize result always as a list
                 result = [result]
                 default_total = 1
             elif "total" not in response_dict:
                 default_total = len(result)
-            final_result = result_formatter(
-                result, result_formatter_params) if result_formatter else result
+            final_result = result_formatter(result, result_formatter_params) if result_formatter else result
             return BaseResult(
                 result=final_result,
                 error=response_dict.get("error", None),
                 total=response_dict.get("total", default_total),
-                has_more=response_dict.get("total", 0) - (
-                    response_dict.get("skip", 0) + response_dict.get("limit", 0)) > 0,
-                hint=kwargs.get("hint", [])
+                has_more=response_dict.get("total", 0)
+                - (response_dict.get("skip", 0) + response_dict.get("limit", 0))
+                > 0,
+                hint=kwargs.get("hint", []),
             )
         except httpx.HTTPStatusError as e:
             if e.response.status_code == 403:
                 return BaseResult(
-                    error=e.response.json().get("error", {}).get('message', 'Invalid Credentials')
+                    error=e.response.json().get("error", {}).get("message", "Invalid Credentials")
                 )
             elif e.response.status_code == 401:
-                return BaseResult(
-                    error="Unauthorized to perform this action"
-                )
+                return BaseResult(error="Unauthorized to perform this action")
             raise

src/common/utils.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Optional, Callable
+from typing import Callable, Optional
 
 
 def get_date_time_iso(timestamp: int) -> Optional[str]:

src/config/token.py

@@ -1,17 +1,19 @@
 import os
+from functools import lru_cache
 from pathlib import Path
 from typing import Union
-from functools import lru_cache
+
 from dotenv import load_dotenv
 
 
 class BzmApimTokenError(Exception):
     """Error when constructing or loading BzmApimToken."""
+
     pass
 
 
 class BzmApimToken:
-    __slots__ = ("token")
+    __slots__ = "token"
 
     def __init__(self, token: str):
         if not token or not isinstance(token, str):
@@ -30,8 +32,7 @@ def from_file(cls, path: Union[str, Path]) -> "BzmApimToken":
             load_dotenv(dotenv_path=p)
             token_val = os.getenv("BZM_API_TEST_TOKEN")
         except Exception as e:
-            raise BzmApimTokenError(
-                f"Error reading/parsing Token from {p!r}: {e}") from e
+            raise BzmApimTokenError(f"Error reading/parsing Token from {p!r}: {e}") from e
 
         return cls(token=token_val)
 

src/config/version.py

@@ -15,12 +15,11 @@ def get_version():
 
 
 def get_executable():
-    if getattr(sys, 'frozen', False):
+    if getattr(sys, "frozen", False):
         return sys.executable
     else:
         # Go up from src/config/version.py to project root, then to main.py
-        return os.path.join(os.path.abspath(
-            Path(__file__).parent.parent.parent), "main.py")
+        return os.path.join(os.path.abspath(Path(__file__).parent.parent.parent), "main.py")
 
 
 __version__ = get_version()

src/formatters/bucket.py

@@ -1,11 +1,10 @@
-from typing import List, Any, Optional
+from typing import Any, List, Optional
+
 from src.models.bucket import Bucket
 
 
 def format_buckets(buckets: List[Any], params: Optional[dict] = None) -> List[Bucket]:
     formatted_buckets = []
     for bucket in buckets:
-        formatted_buckets.append(
-            Bucket(**bucket).model_dump(by_alias=False)
-        )
+        formatted_buckets.append(Bucket(**bucket).model_dump(by_alias=False))
     return formatted_buckets

src/formatters/environment.py

@@ -1,11 +1,10 @@
-from typing import List, Any, Optional
+from typing import Any, List, Optional
+
 from src.models.environment import Environment
 
 
 def format_environments(environments: List[Any], params: Optional[dict] = None) -> List[Environment]:
-	formatted_environments = []
-	for environment in environments:
-		formatted_environments.append(
-			Environment(**environment).model_dump(by_alias=False)
-		)
-	return formatted_environments
+    formatted_environments = []
+    for environment in environments:
+        formatted_environments.append(Environment(**environment).model_dump(by_alias=False))
+    return formatted_environments

src/formatters/result.py

@@ -1,30 +1,26 @@
-from typing import List, Any, Optional
-from src.models.result import TestExecution, TestResult, BucketLevelTestResult
+from typing import Any, List, Optional
+
+from src.models.result import BucketLevelTestResult, TestExecution, TestResult
 
 
 def format_triggered_runs(runs: List[Any], params: Optional[dict] = None) -> List[TestExecution]:
     formatted_runs = []
     for run in runs:
-        formatted_runs.append(
-            TestExecution(**run).model_dump(by_alias=False)
-        )
+        formatted_runs.append(TestExecution(**run).model_dump(by_alias=False))
     return formatted_runs
 
 
 def format_results(results: List[Any], params: Optional[dict] = None) -> List[TestResult]:
     formatted_results = []
     for result in results:
-        formatted_results.append(
-            TestResult(**result).model_dump(by_alias=False)
-        )
+        formatted_results.append(TestResult(**result).model_dump(by_alias=False))
     return formatted_results
 
 
 def format_bucket_level_results(
-        results: List[Any], params: Optional[dict] = None) -> List[BucketLevelTestResult]:
+    results: List[Any], params: Optional[dict] = None
+) -> List[BucketLevelTestResult]:
     formatted_results = []
     for result in results:
-        formatted_results.append(
-            BucketLevelTestResult(**result).model_dump(by_alias=False)
-        )
+        formatted_results.append(BucketLevelTestResult(**result).model_dump(by_alias=False))
     return formatted_results

src/formatters/schedule.py

@@ -1,11 +1,10 @@
-from typing import List, Any, Optional
+from typing import Any, List, Optional
+
 from src.models.schedule import Schedule
 
 
 def format_schedules(schedules: List[Any], params: Optional[dict] = None) -> List[Schedule]:
     formatted_schedules = []
     for schedule in schedules:
-        formatted_schedules.append(
-            Schedule(**schedule).model_dump(by_alias=False)
-        )
+        formatted_schedules.append(Schedule(**schedule).model_dump(by_alias=False))
     return formatted_schedules

src/formatters/step.py

@@ -1,11 +1,10 @@
-from typing import List, Any, Optional
+from typing import Any, List, Optional
+
 from src.models.step import TestStep
 
 
 def format_steps(steps: List[Any], params: Optional[dict] = None) -> TestStep:
     formatted_steps = []
     for step in steps:
-        formatted_steps.append(
-            TestStep(**step).model_dump(by_alias=False)
-        )
+        formatted_steps.append(TestStep(**step).model_dump(by_alias=False))
     return formatted_steps