linting/formatting/type fixes

src/cirrus/management/cli.py

@@ -1,6 +1,8 @@
 import sys
+
+from collections.abc import Callable
 from functools import wraps
-from typing import Any, Callable
+from typing import Any
 
 import boto3
 import botocore.exceptions
@@ -13,11 +15,11 @@
 logger = logging.getLogger(__name__)
 
 
-from cirrus.management import DESCRIPTION, NAME  # noqa: E
-from cirrus.management.commands.deployments import list_deployments  # noqa: E
-from cirrus.management.commands.manage import manage as manage_group  # noqa: E
-from cirrus.management.commands.payload import payload as payload_group  # noqa: E
-from cirrus.management.exceptions import SSOError  # noqa: E
+from cirrus.management import DESCRIPTION, NAME  # noqa: E402
+from cirrus.management.commands.deployments import list_deployments  # noqa: E402
+from cirrus.management.commands.manage import manage as manage_group  # noqa: E402
+from cirrus.management.commands.payload import payload as payload_group  # noqa: E402
+from cirrus.management.exceptions import SSOError  # noqa: E402
 
 
 def handle_sso_error(func: Callable) -> Callable:

src/cirrus/management/commands/manage.py

@@ -1,14 +1,13 @@
 import json
 import logging
 import sys
+
 from functools import wraps
 from subprocess import CalledProcessError
-from typing import Optional
 
 import boto3
 import botocore.exceptions
 import click
-from click_option_group import RequiredMutuallyExclusiveOptionGroup, optgroup
 
 from cirrus.management.deployment import WORKFLOW_POLL_INTERVAL, Deployment
 from cirrus.management.utils.click import (
@@ -17,6 +16,7 @@
     pass_session,
     silence_templating_errors,
 )
+from click_option_group import RequiredMutuallyExclusiveOptionGroup, optgroup
 
 logger = logging.getLogger(__name__)
 
@@ -44,7 +44,7 @@ def execution_arn(func):
         cls=RequiredMutuallyExclusiveOptionGroup,
         help="Identifer type and value to get execution",
     )(func)
-    return func
+    return func  # noqa: RET504
 
 
 def raw_option(func):
@@ -79,7 +79,7 @@ def wrapper(*args, **kwargs):
 )
 @pass_session
 @click.pass_context
-def manage(ctx, session: boto3.Session, deployment: str, profile: Optional[str] = None):
+def manage(ctx, session: boto3.Session, deployment: str, profile: str | None = None):
     """
     Commands to run management operations against a cirrus deployment.
     """
@@ -152,7 +152,7 @@ def download(output_fileobj):
             json.dump(json.load(b), sys.stdout, indent=4)
 
     # ensure we end with a newline
-    print()
+    click.echo("")
 
 
 @manage.command("get-execution")
@@ -188,7 +188,8 @@ def get_execution_input(deployment, arn, payload_id, raw):
 @raw_option
 @pass_deployment
 def get_execution_output(deployment, arn, payload_id, raw):
-    """Get a workflow execution's output payload using its ARN or its input payload ID"""
+    """Get a workflow execution's output payload using its ARN or its input
+    payload ID"""
     output = json.loads(_get_execution(deployment, arn, payload_id)["output"])
 
     if raw:
@@ -223,7 +224,7 @@ def process(deployment):
 def invoke_lambda(deployment, lambda_name):
     """Invoke lambda with event (from stdin)"""
     click.echo(
-        json.dumps(deployment.invoke_lambda(sys.stdin.read(), lambda_name), indent=4)
+        json.dumps(deployment.invoke_lambda(sys.stdin.read(), lambda_name), indent=4),
     )
 
 
@@ -283,7 +284,8 @@ def _exec(ctx, deployment, command, include_user_vars):
 @pass_deployment
 @click.pass_context
 def _call(ctx, deployment, command, include_user_vars):
-    """Run an executable, in a new process, with the deployment environment vars loaded"""
+    """Run an executable, in a new process, with the deployment environment
+    vars loaded"""
     if not command:
         return
     try:
@@ -299,8 +301,10 @@ def list_lambdas(ctx, deployment):
     """List lambda functions"""
     click.echo(
         json.dumps(
-            {"Functions": deployment.get_lambda_functions()}, indent=4, default=str
-        )
+            {"Functions": deployment.get_lambda_functions()},
+            indent=4,
+            default=str,
+        ),
     )
 
 

src/cirrus/management/commands/payload.py

@@ -47,6 +47,8 @@ def template(additional_variables, silence_templating_errors):
 
     click.echo(
         template_payload(
-            sys.stdin.read(), additional_variables, silence_templating_errors
-        )
+            sys.stdin.read(),
+            additional_variables,
+            silence_templating_errors,
+        ),
     )

src/cirrus/management/deployment.py

@@ -4,18 +4,19 @@
 import json
 import logging
 import os
+
 from collections.abc import Iterator
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from pathlib import Path
 from subprocess import check_call
 from time import sleep, time
 from typing import IO, Any
 
 import backoff
 import boto3
+
 from cirrus.lib.process_payload import ProcessPayload
 from cirrus.lib.utils import get_client
-
 from cirrus.management import exceptions
 from cirrus.management.deployment_pointer import DeploymentPointer
 
@@ -29,7 +30,7 @@
 
 
 def now_isoformat():
-    return datetime.now(timezone.utc).isoformat()
+    return datetime.now(UTC).isoformat()
 
 
 def _maybe_use_buffer(fileobj: IO):
@@ -57,17 +58,6 @@ def asjson(self, *args, **kwargs) -> str:
         return json.dumps(self.asdict(), *args, **kwargs)
 
 
-#    @staticmethod
-#    def _get_session(profile: str = None):
-#        # TODO: MFA session should likely be used only with the cli,
-#        #   so this probably needs to be parameterized by the caller
-#
-#    def get_session(self):
-#        if not self._session:
-#            self._session = self._get_session(profile=self.profile)
-#        return self._session
-
-
 @dataclasses.dataclass
 class Deployment(DeploymentMeta):
     def __init__(
@@ -132,20 +122,20 @@ def exec(self, command, include_user_vars=True, isolated=False):
             env = self.environment.copy()
             if include_user_vars:
                 env.update(self.user_vars)
-            os.execlpe(command[0], *command, env)
+            os.execlpe(command[0], *command, env)  # noqa: S606
 
         self.set_env(include_user_vars=include_user_vars)
-        os.execlp(command[0], *command)
+        os.execlp(command[0], *command)  # noqa: S606
 
     def call(self, command, include_user_vars=True, isolated=False):
         if isolated:
             env = self.environment.copy()
             if include_user_vars:
                 env.update(self.user_vars)
-            check_call(command, env=env)
+            check_call(command, env=env)  # noqa: S603
         else:
             self.set_env(include_user_vars=include_user_vars)
-            check_call(command)
+            check_call(command)  # noqa: S603
 
     def get_payload_state(self, payload_id):
         from cirrus.lib.statedb import StateDB
@@ -215,16 +205,16 @@ def get_execution_by_payload_id(self, payload_id):
         execs = self.get_payload_state(payload_id).get("executions", [])
         try:
             exec_arn = execs[-1]
-        except IndexError:
-            raise exceptions.NoExecutionsError(payload_id)
+        except IndexError as e:
+            raise exceptions.NoExecutionsError(payload_id) from e
 
         return self.get_execution(exec_arn)
 
     def invoke_lambda(self, event, function_name):
         aws_lambda = get_client("lambda", session=self.session)
         if function_name not in self.get_lambda_functions():
             raise ValueError(
-                f"lambda named '{function_name}' not found in deployment '{self.name}'"
+                f"lambda named '{function_name}' not found in deployment '{self.name}'",
             )
         full_name = f"{self.stackname}-{function_name}"
         response = aws_lambda.invoke(FunctionName=full_name, Payload=event)

src/cirrus/management/deployment_pointer.py

@@ -32,14 +32,12 @@ class PointerObject(Protocol):  # pragma: no cover
     region: str
 
     @classmethod
-    def from_string(cls: type[Self], string: str) -> Self:
-        ...
+    def from_string(cls: type[Self], string: str) -> Self: ...
 
     def fetch(
         self: Self,
         session: boto3.Session | None = None,
-    ) -> str:
-        ...
+    ) -> str: ...
 
 
 class SecretArn:
@@ -98,7 +96,7 @@ class Pointer:
     @classmethod
     def from_string(cls: type[Self], string: str) -> Self:
         obj = json.loads(string)
-        obj['_type'] = obj.pop('type')
+        obj["_type"] = obj.pop("type")
         return cls(**obj)
 
     def resolve(self) -> PointerObject:

src/cirrus/management/utils/click.py

@@ -62,7 +62,7 @@ def get_command(self, ctx, cmd_name):
                 self.resolve_alias(cmd)
                 for cmd in self.list_commands(ctx) + list(self._alias2cmd.keys())
                 if cmd.startswith(cmd_name)
-            }
+            },
         )
 
         # no matches no command
@@ -71,13 +71,14 @@ def get_command(self, ctx, cmd_name):
 
         # one match then we can resolve the match
         # and try getting the command again
-        elif len(matches) == 1:
+        if len(matches) == 1:
             return super().get_command(ctx, matches[0])
 
         # otherwise the string matched but was not unique
         # to a single command and we have to bail out
-        ctx.fail(
-            f"Unknown command '{cmd_name}. Did you mean any of these: {', '.join(sorted(matches))}?",
+        ctx.fail(  # noqa: RET503
+            f"Unknown command '{cmd_name}. Did you mean any of these: "
+            f"{', '.join(sorted(matches))}?",
         )
 
     def format_commands(self, ctx, formatter):
@@ -144,8 +145,6 @@ class Variable(click.ParamType):
     name = "key/val pair"
 
     def convert(self, value, param, ctx):
-        print(22, value)
-        print(33, param)
         return {value[0]: value[1]}
 
 
@@ -177,7 +176,6 @@ def additional_variables(func):
         "additional_variables",
         nargs=2,
         multiple=True,
-        # type=Variable(),
         callback=merge_vars2,
         help="Additional templating variables",
     )(func)

src/cirrus/management/utils/logging_classes.py

@@ -1,12 +1,14 @@
 import logging
 
+from typing import Any, ClassVar
+
 import click
 
 # Inspired from https://github.com/click-contrib/click-log
 
 
 class ClickFormatter(logging.Formatter):
-    colors = {
+    colors: ClassVar[dict[str, dict[str, Any]]] = {
         "error": {"fg": "red"},
         "exception": {"fg": "red"},
         "critical": {"fg": "red"},
@@ -28,5 +30,6 @@ def emit(self, record):
             msg = self.format(record)
             record.levelname.lower()
             click.echo(msg, err=True)
-        except Exception:
+        # not sure if we can narrow this exception down or not...
+        except Exception:  # noqa: BLE001
             self.handleError(record)

src/cirrus/management/utils/templating.py

@@ -1,4 +1,5 @@
 import logging
+
 from string import Template
 
 logger = logging.getLogger(__name__)