Merge branch 'mr/trespeuch/ruff' into 'master'

Fix additional ruff check violations

See merge request it/e3-aws!142

Author: pierretr

examples/deploy_simple_stack.py

@@ -21,9 +21,9 @@
 
 import sys
 from functools import cached_property
-from typing_extensions import override
 
 from troposphere import GetAtt, Ref, Tags, ec2, iam
+from typing_extensions import override
 
 from e3.aws.troposphere import CFNProjectMain, Construct, Stack, name_to_id
 from e3.aws.troposphere.ec2 import VPCv2

src/e3/aws/cfn/__init__.py

@@ -22,6 +22,7 @@
 import yaml
 from typing_extensions import override
 
+from e3.aws import iterate
 from e3.env import Env
 
 from typing import TYPE_CHECKING, cast
@@ -41,7 +42,7 @@
         ValidateTemplateOutputTypeDef,
     )
 
-    from typing import Any, ParamSpec, TypeVar
+    from typing import Any, ParamSpec, Self, TypeVar
 
     P = ParamSpec("P")
     T = TypeVar("T")
@@ -517,7 +518,7 @@ def __init__(
         self.uuid = str(uuid.uuid1(clock_seq=int(1000 * time.time())))
         self.latest_read_event: StackEvent | None = None
 
-    def add(self, element: Stack | Resource) -> Stack:
+    def add(self, element: Stack | Resource) -> Self:
         """Add a resource or merge a stack.
 
         :param element: if a resource add the resource to the stack. If a stack
@@ -533,7 +534,7 @@ def add(self, element: Stack | Resource) -> Stack:
         self.resources[element.name] = element
         return self
 
-    def __iadd__(self, element: Stack | Resource) -> Stack:
+    def __iadd__(self, element: Stack | Resource) -> Self:
         """Add a resource or merge a stack.
 
         :param element: if a resource add the resource to the stack. If a stack
@@ -664,9 +665,7 @@ def exists(self) -> bool:
         """
         try:
             self.state()
-        except Exception:
-            # Documentation does not specify the right exception that is raised
-            # by botocore.
+        except Exception:  # noqa: BLE001
             return False
         return True
 
@@ -792,8 +791,6 @@ def events(
         :param mark_as_read: if True, all events read won't be returned on next
             calls to events method.
         """
-        from e3.aws import iterate
-
         latest_read_event = self.latest_read_event
         if mark_as_read:
             self.latest_read_event = None

src/e3/aws/cfn/iam.py

@@ -204,7 +204,7 @@ def append(self, statement: Statement) -> PolicyDocument:
         self.statements.append(statement)
         return self
 
-    def extend(self, statements: list[Statement]) -> PolicyDocument:
+    def extend(self, statements: list[Statement]) -> Self:
         """Append a list of statements.
 
         :param statements: IAM Statements
@@ -213,7 +213,7 @@ def extend(self, statements: list[Statement]) -> PolicyDocument:
         self.statements.extend(statements)
         return self
 
-    def __iadd__(self, statements: list[Statement]) -> PolicyDocument:  # type: ignore[misc]
+    def __iadd__(self, statements: list[Statement]) -> Self:  # type: ignore[misc]
         """See extend."""
         return self.extend(statements)
 

src/e3/aws/ec2/ami.py

@@ -233,8 +233,8 @@ def find(
                 continue
 
             consider_ami = True
-            for tag in tag_filters:
-                if not re.match(tag_filters[tag], ami.tags.get(tag, "")):
+            for tag, pattern in tag_filters.items():
+                if not re.match(pattern, ami.tags.get(tag, "")):
                     consider_ami = False
                     continue
 

src/e3/aws/troposphere/__init__.py

@@ -168,11 +168,8 @@ def _upload_file(
         :param check_exists: check if an S3 object exists before uploading it
         :param dry_run: don't upload the file if set
         """
-        logger.info(
-            "Upload {} to {}:{}".format(
-                os.path.relpath(file, root_dir).replace("\\", "/"), s3_bucket, s3_key
-            )
-        )
+        rel_path = os.path.relpath(file, root_dir).replace("\\", "/")
+        logger.info(f"Upload {rel_path} to {s3_bucket}:{s3_key}")
 
         if client is None:
             return

src/e3/aws/troposphere/awslambda/__init__.py

@@ -44,7 +44,7 @@
 
     from e3.aws.troposphere import Stack
 
-    from typing import Any
+    from typing import Any, Self
 
 logger = logging.getLogger("e3.aws.troposphere.awslambda")
 
@@ -154,7 +154,7 @@ def __init__(
         self._archive_tmpd: TemporaryDirectory | None = None
         self._archive_dir: str | None = None
 
-    def __enter__(self) -> PyFunctionAsset:
+    def __enter__(self) -> Self:
         """Create a temporary archive directory."""
         if self._archive_dir is None:
             self._archive_tmpd = TemporaryDirectory()
@@ -826,10 +826,10 @@ def _show_archive_files(self, archive_path: str | Path) -> list[str]:
         :param archive_path: path to the archive
         :return: the list of files in the archive
         """
-        with zipfile.ZipFile(archive_path) as zip:
+        with zipfile.ZipFile(archive_path) as archive:
             return [
                 f"{line}\n"
-                for line in sorted(zip.namelist())
+                for line in sorted(archive.namelist())
                 if not line.endswith(".pyc")
             ]
 
@@ -1306,11 +1306,11 @@ def create_alias(
             :param default_name: default alias name if none is specified
             """
             name = config.name if config.name is not None else default_name
-            id = name_to_id(f"{self.lambda_name}-{name}-alias")
+            alias_id = name_to_id(f"{self.lambda_name}-{name}-alias")
             return Alias(
-                name=id,
+                name=alias_id,
                 description=f"{name} alias for {self.lambda_name} lambda",
-                alias_name=config.name if config.name is not None else id,
+                alias_name=config.name if config.name is not None else alias_id,
                 lambda_arn=self.lambda_arn,
                 lambda_version=config.version,
                 provisioned_concurrency_config=config.provisioned_concurrency_config,

src/e3/aws/troposphere/awslambda/docker.py

@@ -4,14 +4,15 @@
 
 import logging
 from datetime import datetime, timezone
-from pathlib import Path
 
 from e3.aws.troposphere.awslambda import Architecture, Function, UnknownPlatform
 from e3.aws.util.ecr import build_and_push_image
 
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
+    from pathlib import Path
+
     from python_on_whales import DockerClient
     from troposphere import AWSObject, GetAtt, awslambda
 

src/e3/aws/troposphere/awslambda/flask_apigateway_wrapper.py

@@ -241,9 +241,9 @@ def create_flask_wsgi_environ(self, event: dict, context: dict) -> dict:
             environ["CONTENT_LENGTH"] = str(len(body))
 
         # Export headers into the WSGI environment
-        for header in headers:
+        for header, value in headers.items():
             wsgi_name = "HTTP_" + header.upper().replace("-", "_")
-            environ[wsgi_name] = headers[header]
+            environ[wsgi_name] = value
 
         # Set HTTP_COOKIE if necessary
         if cookies:

src/e3/aws/troposphere/cloudfront/data/lambda_invalidate_head.py

@@ -5,10 +5,10 @@
 import os
 import time
 
-from typing import TYPE_CHECKING
-
 import boto3
 
+from typing import TYPE_CHECKING
+
 if TYPE_CHECKING:
     from typing import Any
 

src/e3/aws/troposphere/iam/policy_document.py

@@ -9,6 +9,8 @@
 if TYPE_CHECKING:
     from e3.aws.troposphere.iam.policy_statement import PolicyStatement
 
+    from typing import Self
+
 
 @dataclass
 class PolicyDocument:
@@ -29,7 +31,7 @@ def __add__(self, other: PolicyDocument) -> PolicyDocument:
         """
         return PolicyDocument(statements=self.statements + other.statements)
 
-    def __iadd__(self, other: PolicyDocument) -> PolicyDocument:
+    def __iadd__(self, other: PolicyDocument) -> Self:
         """Merge another policy document into this one.
 
         :param other: the policy document to merge