linting

Author: gilesknap

pyproject.toml

@@ -148,6 +148,8 @@ lint.ignore = [
     "E501",  # Line too long, should be fixed by black.
     "B008",  # Do not perform function calls in argument defaults.
     "UP007", # Do not complain about Optional[] (TODO: remove once typer is patched)
+    "N803",  # Argument name should be lowercase
+    "N802",  # Function name should be lowercase
 ]
 lint.select = [
     "B",   # flake8-bugbear - https://docs.astral.sh/ruff/rules/#flake8-bugbear-b

src/edge_containers_cli/__main__.py

@@ -1,6 +1,5 @@
 import os
 import sys
-from typing import Optional
 
 import typer
 
@@ -36,7 +35,7 @@ def backend_callback(ctx: typer.Context, backend: ECBackends):
 @cli.callback()
 def _main(
     ctx: typer.Context,
-    version: Optional[bool] = typer.Option(
+    version: bool | None = typer.Option(
         None,
         "--version",
         callback=version_callback,

src/edge_containers_cli/cmds/demo_commands.py

@@ -40,9 +40,9 @@ def process_t(time_string) -> str:
 }
 
 if NUM_SERVICES == 0:
-    sample_ServicesDataFrame = polars.DataFrame(schema=ServicesSchema)
+    SampleServicesDataFrame = polars.DataFrame(schema=ServicesSchema)
 else:
-    sample_ServicesDataFrame = ServicesDataFrame(polars.from_dict(sample_data))
+    SampleServicesDataFrame = ServicesDataFrame(polars.from_dict(sample_data))
 
 
 def demo_wrapper():
@@ -78,7 +78,7 @@ def __init__(
     ):
         self._target = "Demo Beamline"
         self._target_valid = False
-        self._stateDF = sample_ServicesDataFrame
+        self._stateDF = SampleServicesDataFrame
 
         self.lorem_min = 10
         self.lorem_max = 50

src/edge_containers_cli/cmds/helm.py

@@ -1,6 +1,5 @@
 from collections.abc import Callable
 from pathlib import Path
-from typing import Optional
 
 from ruamel.yaml import YAML
 
@@ -25,9 +24,9 @@ def __init__(
         namespace: str,
         service_name: str,
         args: str = "",
-        version: Optional[str] = None,
+        version: str | None = None,
         template: bool = False,
-        repo: Optional[str] = None,
+        repo: str | None = None,
     ):
         """
         Create a helm chart from a local or a remote repo

src/edge_containers_cli/cmds/monitor.py

@@ -34,7 +34,7 @@
 from textual.worker import get_current_worker
 
 from edge_containers_cli.cmds.commands import Commands
-from edge_containers_cli.definitions import ECLogLevels, emoji
+from edge_containers_cli.definitions import ECLogLevels, Emoji
 from edge_containers_cli.logging import log
 
 
@@ -239,7 +239,7 @@ def __init__(self, commands, running_only: bool) -> None:
         self._indicator_lock = threading.Lock()
         self._service_indicators = {
             "name": [""],
-            emoji.exclaim: [""],
+            Emoji.exclaim: [""],
         }
         iocs_df = self._get_services_df(self.running_only)
         self.columns = iocs_df.columns
@@ -289,8 +289,8 @@ def _get_services_df(self, running_only):
         services_df = self.commands._get_services(running_only)  # noqa: SLF001
         services_df = services_df.with_columns(
             polars.when(polars.col("ready"))
-            .then(polars.lit(emoji.check_mark))
-            .otherwise(polars.lit(emoji.cross_mark))
+            .then(polars.lit(Emoji.check_mark))
+            .otherwise(polars.lit(Emoji.cross_mark))
             .alias("ready")
         )
         indicators_df = polars.DataFrame(self._service_indicators)
@@ -305,10 +305,10 @@ def update_indicator_threadsafe(self, name: str, indicator: str):
         with self._indicator_lock:
             if name in self._service_indicators["name"]:
                 index = self._service_indicators["name"].index(name)
-                self._service_indicators[emoji.exclaim][index] = indicator
+                self._service_indicators[Emoji.exclaim][index] = indicator
             else:
                 self._service_indicators["name"].append(name)
-                self._service_indicators[emoji.exclaim].append(indicator)
+                self._service_indicators[Emoji.exclaim].append(indicator)
 
     def watch_sort_column_id(self, sort_column_id: str) -> None:
         """Called when the sort_column_id attribute changes."""
@@ -479,9 +479,9 @@ def _do_confirmed_action(self, action: str, command: Callable):
 
             def do_task(command, service_name):
                 def _do_task():
-                    table.update_indicator_threadsafe(service_name, emoji.road_works)
+                    table.update_indicator_threadsafe(service_name, Emoji.road_works)
                     command(service_name)
-                    table.update_indicator_threadsafe(service_name, emoji.none)
+                    table.update_indicator_threadsafe(service_name, Emoji.none)
                     self.busy_services.remove(service_name)
 
                 return _do_task
@@ -496,7 +496,7 @@ def after_dismiss_callback(start: bool | None) -> None:
                         log.info(f"Scheduled: {action} {service_name}")
                         self.busy_services.add(service_name)
                         table.update_indicator_threadsafe(
-                            service_name, emoji.hour_glass
+                            service_name, Emoji.hour_glass
                         )
                         self._queue.put(do_task(command, service_name))
 
@@ -523,7 +523,7 @@ def action_ioc_logs(self) -> None:
         """Display the logs of the IOC that is currently highlighted."""
         if service_name := self._get_service_name():
             # Convert to corresponding bool
-            ready = self._get_highlighted_cell("ready") == emoji.check_mark
+            ready = self._get_highlighted_cell("ready") == Emoji.check_mark
 
             if ready:
                 command = self.commands._get_logs  # noqa: SLF001

src/edge_containers_cli/definitions.py

@@ -34,7 +34,7 @@ class ECContext:
     log_url: str = ""
 
 
-class emoji(str, Enum):
+class Emoji(str, Enum):
     none = ""
     road_works = "\U0001f6a7"
     exclaim = "\U00002755"