Fix formatting and linting

robusta_krr/core/abstract/strategies.py

@@ -2,17 +2,16 @@
 
 import abc
 import datetime
-import numpy as np
-from numpy.typing import NDArray
-from typing import Generic, Optional, TypeVar, get_args, Annotated, Literal
 from textwrap import dedent
+from typing import Annotated, Generic, Literal, Optional, TypeVar, get_args
 
+import numpy as np
 import pydantic as pd
+from numpy.typing import NDArray
 
-from robusta_krr.core.models.result import K8sObjectData, ResourceType, Metric
+from robusta_krr.core.models.result import K8sObjectData, Metric, ResourceType
 from robusta_krr.utils.display_name import add_display_name
 
-
 SelfRR = TypeVar("SelfRR", bound="ResourceRecommendation")
 
 
@@ -22,7 +21,7 @@ class ResourceRecommendation(pd.BaseModel):
 
     @classmethod
     def undefined(cls: type[SelfRR]) -> SelfRR:
-        return cls(request=float('NaN'), limit=float('NaN'))
+        return cls(request=float("NaN"), limit=float("NaN"))
 
 
 class StrategySettings(pd.BaseModel):
@@ -74,9 +73,9 @@ def __str__(self) -> str:
     @property
     def description(self) -> Optional[str]:
         """
-            Generate a description for the strategy.
-            You can use the settings in the description by using the format syntax.
-            Also you can use Rich's markdown syntax to format the description.
+        Generate a description for the strategy.
+        You can use the settings in the description by using the format syntax.
+        Also you can use Rich's markdown syntax to format the description.
         """
 
         if self.__doc__ is None:

robusta_krr/core/integrations/prometheus/loader.py

@@ -1,7 +1,7 @@
+import asyncio
 import datetime
 from typing import Optional, no_type_check
 
-import asyncio
 import requests
 from kubernetes import config as k8s_config
 from kubernetes.client import ApiClient
@@ -123,11 +123,7 @@ async def gather_data(
         metric_loader = MetricLoaderType(self.config, self.prometheus)
         return await metric_loader.load_data(object, period, step)
 
-    async def add_historic_pods(
-        self,
-        object: K8sObjectData,
-        period: datetime.timedelta
-    ) -> None:
+    async def add_historic_pods(self, object: K8sObjectData, period: datetime.timedelta) -> None:
         """Find pods that were already deleted, but still have some metrics in Prometheus"""
 
         if len(object.pods) == 0:
@@ -136,7 +132,7 @@ async def add_historic_pods(
         period_literal = f"{int(period.total_seconds()) // 60 // 24}d"
         owner = await asyncio.to_thread(
             self.prometheus.custom_query,
-            query=f'kube_pod_owner{{pod="{next(iter(object.pods)).name}"}}[{period_literal}]'
+            query=f'kube_pod_owner{{pod="{next(iter(object.pods)).name}"}}[{period_literal}]',
         )
 
         if owner == []:
@@ -145,8 +141,7 @@ async def add_historic_pods(
         owner = owner[0]["metric"]["owner_name"]
 
         related_pods = await asyncio.to_thread(
-            self.prometheus.custom_query,
-            query=f'kube_pod_owner{{owner_name="{owner}"}}[{period_literal}]'
+            self.prometheus.custom_query, query=f'kube_pod_owner{{owner_name="{owner}"}}[{period_literal}]'
         )
 
         current_pods = {p.name for p in object.pods}

robusta_krr/core/integrations/prometheus/metrics/base_filtered_metric.py

@@ -1,8 +1,9 @@
 from typing import Any, Optional
 
-from .base_metric import BaseMetricLoader
 from robusta_krr.core.abstract.strategies import Metric
 
+from .base_metric import BaseMetricLoader
+
 PrometheusSeries = Any
 
 

robusta_krr/core/integrations/prometheus/metrics/base_metric.py

@@ -5,13 +5,13 @@
 import datetime
 from typing import TYPE_CHECKING, Callable, TypeVar
 
-from robusta_krr.core.abstract.strategies import ResourceHistoryData, Metric
+import numpy as np
+
+from robusta_krr.core.abstract.strategies import Metric, ResourceHistoryData
 from robusta_krr.core.models.config import Config
 from robusta_krr.core.models.objects import K8sObjectData
 from robusta_krr.utils.configurable import Configurable
 
-import numpy as np
-
 if TYPE_CHECKING:
     from ..loader import CustomPrometheusConnect
 

robusta_krr/core/integrations/prometheus/metrics/cpu_metric.py

@@ -1,18 +1,18 @@
 from robusta_krr.core.models.allocations import ResourceType
+from robusta_krr.core.models.objects import K8sObjectData
 
-from .base_metric import bind_metric
 from .base_filtered_metric import BaseFilteredMetricLoader
-from robusta_krr.core.models.objects import K8sObjectData
+from .base_metric import bind_metric
 
 
 @bind_metric(ResourceType.CPU)
 class CPUMetricLoader(BaseFilteredMetricLoader):
     def get_query(self, object: K8sObjectData) -> str:
         pods_selector = "|".join(pod.name for pod in object.pods)
         return (
-            'sum(irate(container_cpu_usage_seconds_total{'
+            "sum(irate(container_cpu_usage_seconds_total{"
             f'namespace="{object.namespace}", '
             f'pod=~"{pods_selector}", '
             f'container="{object.container}"'
-            '}[5m])) by (container, pod, job)'
+            "}[5m])) by (container, pod, job)"
         )

robusta_krr/core/integrations/prometheus/metrics/memory_metric.py

@@ -1,18 +1,18 @@
 from robusta_krr.core.models.allocations import ResourceType
+from robusta_krr.core.models.objects import K8sObjectData
 
-from .base_metric import bind_metric
 from .base_filtered_metric import BaseFilteredMetricLoader
-from robusta_krr.core.models.objects import K8sObjectData
+from .base_metric import bind_metric
 
 
 @bind_metric(ResourceType.Memory)
 class MemoryMetricLoader(BaseFilteredMetricLoader):
     def get_query(self, object: K8sObjectData) -> str:
         pods_selector = "|".join(pod.name for pod in object.pods)
         return (
-            'sum(container_memory_working_set_bytes{'
+            "sum(container_memory_working_set_bytes{"
             f'namespace="{object.namespace}", '
             f'pod=~"{pods_selector}", '
             f'container="{object.container}"'
-            '}) by (container, pod, job)'
+            "}) by (container, pod, job)"
         )

robusta_krr/core/models/config.py

@@ -1,6 +1,6 @@
+import sys
 from typing import Any, Literal, Optional, Union
 
-import sys
 import pydantic as pd
 from kubernetes import config
 from kubernetes.config.config_exception import ConfigException

robusta_krr/core/models/result.py

@@ -2,8 +2,8 @@
 
 import enum
 import itertools
-from typing import Any, Union, Optional
 from datetime import datetime
+from typing import Any, Optional, Union
 
 import pydantic as pd
 
@@ -78,7 +78,9 @@ class ResourceScan(pd.BaseModel):
     metrics: MetricsData
 
     @classmethod
-    def calculate(cls, object: K8sObjectData, recommendation: ResourceAllocations, metrics: MetricsData) -> ResourceScan:
+    def calculate(
+        cls, object: K8sObjectData, recommendation: ResourceAllocations, metrics: MetricsData
+    ) -> ResourceScan:
         recommendation_processed = ResourceRecommendation(requests={}, limits={})
 
         for resource_type in ResourceType:
@@ -96,7 +98,9 @@ def calculate(cls, object: K8sObjectData, recommendation: ResourceAllocations, m
             for selector in ["requests", "limits"]:
                 for recommendation_request in getattr(recommendation_processed, selector).values():
                     if recommendation_request.severity == severity:
-                        return cls(object=object, recommended=recommendation_processed, severity=severity, metrics=metrics)
+                        return cls(
+                            object=object, recommended=recommendation_processed, severity=severity, metrics=metrics
+                        )
 
         return cls(object=object, recommended=recommendation_processed, severity=Severity.UNKNOWN, metrics=metrics)
 

robusta_krr/core/runner.py

@@ -7,11 +7,12 @@
 from robusta_krr.core.integrations.prometheus import PrometheusLoader, PrometheusNotFound
 from robusta_krr.core.models.config import Config
 from robusta_krr.core.models.objects import K8sObjectData
-from robusta_krr.core.models.result import ResourceAllocations, ResourceScan, ResourceType, Result, MetricsData
+from robusta_krr.core.models.result import MetricsData, ResourceAllocations, ResourceScan, ResourceType, Result
 from robusta_krr.utils.configurable import Configurable
 from robusta_krr.utils.logo import ASCII_LOGO
-from robusta_krr.utils.version import get_version
 from robusta_krr.utils.progress_bar import ProgressBar
+from robusta_krr.utils.version import get_version
+
 
 class Runner(Configurable):
     EXPECTED_EXCEPTIONS = (KeyboardInterrupt, PrometheusNotFound)
@@ -118,7 +119,9 @@ async def _calculate_object_recommendations(self, object: K8sObjectData) -> tupl
         result = await asyncio.to_thread(self._strategy.run, data, object)
         return self._format_result(result), metrics
 
-    async def _gather_objects_recommendations(self, objects: list[K8sObjectData]) -> list[tuple[ResourceAllocations, MetricsData]]:
+    async def _gather_objects_recommendations(
+        self, objects: list[K8sObjectData]
+    ) -> list[tuple[ResourceAllocations, MetricsData]]:
         recommendations: list[tuple[RunResult, MetricsData]] = await asyncio.gather(
             *[self._calculate_object_recommendations(object) for object in objects]
         )
@@ -128,7 +131,8 @@ async def _gather_objects_recommendations(self, objects: list[K8sObjectData]) ->
                 ResourceAllocations(
                     requests={resource: recommendation[resource].request for resource in ResourceType},
                     limits={resource: recommendation[resource].limit for resource in ResourceType},
-                ), metric
+                ),
+                metric,
             )
             for recommendation, metric in recommendations
         ]
@@ -150,7 +154,8 @@ async def _collect_result(self) -> Result:
 
         return Result(
             scans=[
-                ResourceScan.calculate(obj, recommended, metrics) for obj, (recommended, metrics) in zip(objects, resource_recommendations)
+                ResourceScan.calculate(obj, recommended, metrics)
+                for obj, (recommended, metrics) in zip(objects, resource_recommendations)
             ],
             description=self._strategy.description,
         )

robusta_krr/formatters/yaml.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+import json
+
+import yaml
+
 from robusta_krr.core.abstract.formatters import BaseFormatter
 from robusta_krr.core.models.result import Result
-import yaml
-import json
 
 
 class YAMLFormatter(BaseFormatter):

robusta_krr/strategies/simple.py

@@ -1,5 +1,5 @@
-import pydantic as pd
 import numpy as np
+import pydantic as pd
 from numpy.typing import NDArray
 
 from robusta_krr.core.abstract.strategies import (
@@ -14,9 +14,7 @@
 
 
 class SimpleStrategySettings(StrategySettings):
-    cpu_percentile: float = pd.Field(
-        99, gt=0, le=100, description="The percentile to use for the CPU recommendation."
-    )
+    cpu_percentile: float = pd.Field(99, gt=0, le=100, description="The percentile to use for the CPU recommendation.")
     memory_buffer_percentage: float = pd.Field(
         5, gt=0, description="The percentage of added buffer to the peak memory usage for memory recommendation."
     )

robusta_krr/utils/progress_bar.py

@@ -1,6 +1,8 @@
-from robusta_krr.utils.configurable import Configurable
 from alive_progress import alive_bar
+
 from robusta_krr.core.models.config import Config
+from robusta_krr.utils.configurable import Configurable
+
 
 class ProgressBar(Configurable):
     def __init__(self, config: Config, **kwargs) -> None:
@@ -20,4 +22,4 @@ def progress(self):
 
     def __exit__(self, *args):
         if self.show_bar:
-            self.alive_bar.__exit__(*args)
+            self.alive_bar.__exit__(*args)

robusta_krr/utils/resource_units.py

@@ -43,11 +43,11 @@ def format(x: Union[float, int], /, *, base: Literal[1024, 1000] = 1024) -> str:
     if x < 1:
         return f"{int(x*1000)}m"
 
-    units = ['', 'K', 'M', 'G', 'T', 'P', 'E']
-    binary_units = ['', 'Ki', 'Mi', 'Gi', 'Ti', 'Pi', 'Ei']
+    units = ["", "K", "M", "G", "T", "P", "E"]
+    binary_units = ["", "Ki", "Mi", "Gi", "Ti", "Pi", "Ei"]
 
     x = int(x)
     for i, unit in enumerate(binary_units if base == 1024 else units):
-        if x < base**(i + 1) or i == len(units) - 1 or x / base**(i + 1) < 10:
+        if x < base ** (i + 1) or i == len(units) - 1 or x / base ** (i + 1) < 10:
             return f"{x/base**i:.0f}{unit}"
     return f"{x/6**i:.0f}{unit}"

robusta_krr/utils/service_discovery.py

@@ -3,11 +3,10 @@
 
 from cachetools import TTLCache
 from kubernetes import client
-from kubernetes.client import V1ServiceList, V1IngressList
+from kubernetes.client import V1IngressList, V1ServiceList
 from kubernetes.client.api_client import ApiClient
-from kubernetes.client.models.v1_service import V1Service
 from kubernetes.client.models.v1_ingress import V1Ingress
-from kubernetes.config.config_exception import ConfigException
+from kubernetes.client.models.v1_service import V1Service
 
 from robusta_krr.utils.configurable import Configurable
 
@@ -76,6 +75,6 @@ def find_url(self, selectors: list[str], *, api_client: Optional[ApiClient] = No
             self.find_ingress_host(label_selector, api_client=api_client)
             ingress_url = self.find_ingress_host(label_selector, api_client=api_client)
             if ingress_url:
-                return ingress_url 
+                return ingress_url
 
         return None