chore: Upgrade ruff/black/mypy and fix violations

bin/_file_formatter.py

@@ -41,8 +41,8 @@ def decode_exception() -> Type[Exception]:
     def file_extension() -> str:
         """Return file extension of files to format."""
 
-    @classmethod
-    def config_additional_args(cls) -> None:  # noqa: empty-method-without-abstract-decorator
+    @classmethod  # noqa: B027
+    def config_additional_args(cls) -> None:
         """Optionally configure additional args to arg parser."""
 
     def process_file(self, file_path: Path) -> None:

integration/combination/test_function_with_schedule_dlq_generated.py

@@ -84,6 +84,6 @@ def test_function_with_schedule(self):
 
 
 def get_first_key_value_pair_in_dict(dictionary):
-    key = list(dictionary.keys())[0]
+    key = next(iter(dictionary.keys()))
     value = dictionary[key]
     return key, value

integration/combination/test_state_machine_with_cwe_dlq_generated.py

@@ -110,6 +110,6 @@ def test_state_machine_with_cwe(self):
 
 
 def get_first_key_value_pair_in_dict(dictionary):
-    key = list(dictionary.keys())[0]
+    key = next(iter(dictionary.keys()))
     value = dictionary[key]
     return key, value

integration/combination/test_state_machine_with_schedule_dlq_generated.py

@@ -79,6 +79,6 @@ def test_state_machine_with_schedule(self):
 
 
 def get_first_key_value_pair_in_dict(dictionary):
-    key = list(dictionary.keys())[0]
+    key = next(iter(dictionary.keys()))
     value = dictionary[key]
     return key, value

integration/helpers/base_test.py

@@ -536,9 +536,7 @@ def verify_get_request_response(self, url, expected_status_code, headers=None):
         response = self.do_get_request_with_logging(url, headers)
         if response.status_code != expected_status_code:
             raise StatusCodeError(
-                "Request to {} failed with status: {}, expected status: {}".format(
-                    url, response.status_code, expected_status_code
-                )
+                f"Request to {url} failed with status: {response.status_code}, expected status: {expected_status_code}"
             )
         return response
 
@@ -565,9 +563,7 @@ def verify_options_request(self, url, expected_status_code, headers=None):
         response = self.do_options_request_with_logging(url, headers)
         if response.status_code != expected_status_code:
             raise StatusCodeError(
-                "Request to {} failed with status: {}, expected status: {}".format(
-                    url, response.status_code, expected_status_code
-                )
+                f"Request to {url} failed with status: {response.status_code}, expected status: {expected_status_code}"
             )
         return response
 
@@ -576,9 +572,7 @@ def verify_post_request(self, url: str, body_obj, expected_status_code: int):
         response = self.do_post_request(url, body_obj)
         if response.status_code != expected_status_code:
             raise StatusCodeError(
-                "Request to {} failed with status: {}, expected status: {}".format(
-                    url, response.status_code, expected_status_code
-                )
+                f"Request to {url} failed with status: {response.status_code}, expected status: {expected_status_code}"
             )
         return response
 

integration/helpers/deployer/deployer.py

@@ -388,7 +388,7 @@ def _display_stack_outputs(self, stack_outputs, **kwargs):
                 ("Value", output.get("OutputValue")),
             ]:
                 pprint_columns(
-                    columns=["{k:<{0}}{v:<{0}}".format(MIN_OFFSET, k=k, v=v)],
+                    columns=[f"{k:<{MIN_OFFSET}}{v:<{MIN_OFFSET}}"],
                     width=kwargs["width"],
                     margin=kwargs["margin"],
                     format_string=OUTPUTS_FORMAT_STRING,

integration/helpers/resource.py

@@ -49,9 +49,7 @@ def verify_stack_resources(expected_file_path, stack_resources):
     parsed_resources = _sort_resources(stack_resources["StackResourceSummaries"])
 
     if len(expected_resources) != len(parsed_resources):
-        return "'{}' resources expected, '{}' found: \n{}".format(
-            len(expected_resources), len(parsed_resources), json.dumps(parsed_resources, default=str)
-        )
+        return f"'{len(expected_resources)}' resources expected, '{len(parsed_resources)}' found: \n{json.dumps(parsed_resources, default=str)}"
 
     for i in range(len(expected_resources)):
         exp = expected_resources[i]
@@ -65,9 +63,7 @@ def verify_stack_resources(expected_file_path, stack_resources):
                 "ResourceType": parsed["ResourceType"],
             }
 
-            return "'{}' expected, '{}' found (Don't include the LogicalId random suffix)".format(
-                exp, parsed_trimed_down
-            )
+            return f"'{exp}' expected, '{parsed_trimed_down}' found (Don't include the LogicalId random suffix)"
         if exp["ResourceType"] != parsed["ResourceType"]:
             return "'{}' expected, '{}' found".format(exp["ResourceType"], parsed["ResourceType"])
     return None
@@ -292,5 +288,5 @@ def first_item_in_dict(dictionary):
     """
     if not dictionary:
         return None
-    first_key = list(dictionary.keys())[0]
+    first_key = next(iter(dictionary.keys()))
     return first_key, dictionary[first_key]

requirements/dev.txt

@@ -4,7 +4,7 @@ pytest-xdist>=2.5,<4
 pytest-env>=0.6,<1
 pytest-rerunfailures>=9.1,<12
 pyyaml~=6.0
-ruff==0.0.263  # loose the requirement once it is more stable
+ruff==0.0.284  # loose the requirement once it is more stable
 
 # Test requirements
 pytest>=6.2,<8
@@ -19,7 +19,7 @@ tenacity~=8.0
 requests~=2.28
 
 # formatter
-black==23.1.0
+black==23.3.0  # 23.3.0 is the last version supporting python 3.7
 ruamel.yaml==0.17.21  # It can parse yaml while perserving comments
 
 # type check

ruff.toml

@@ -30,6 +30,9 @@ select = [
 ignore = [
     "UP006", # https://github.com/charliermarsh/ruff/pull/4427
     "UP007", # https://github.com/charliermarsh/ruff/pull/4427
+    # Mutable class attributes should be annotated with `typing.ClassVar`
+    # Too many violations
+    "RUF012",
 ]
 
 # Mininal python version we support is 3.7

samtranslator/feature_toggle/feature_toggle.py

@@ -56,7 +56,7 @@ def _get_dialup(self, region_config, feature_name):  # type: ignore[no-untyped-d
             return FeatureToggle.DIALUP_RESOLVER[dialup_type](
                 region_config, account_id=self.account_id, feature_name=feature_name
             )
-        LOG.warning(f"Dialup type '{dialup_type}' is None or is not supported.")
+        LOG.warning("Dialup type '%s' is None or is not supported.", dialup_type)
         return DisabledDialup(region_config)
 
     def is_enabled(self, feature_name: str) -> bool:
@@ -66,21 +66,21 @@ def is_enabled(self, feature_name: str) -> bool:
         :param feature_name: name of feature
         """
         if feature_name not in self.feature_config:
-            LOG.warning(f"Feature '{feature_name}' not available in Feature Toggle Config.")
+            LOG.warning("Feature '%s' not available in Feature Toggle Config.", feature_name)
             return False
 
         stage = self.stage
         region = self.region
         account_id = self.account_id
         if not stage or not region or not account_id:
             LOG.warning(
-                f"One or more of stage, region and account_id is not set. Feature '{feature_name}' not enabled."
+                "One or more of stage, region and account_id is not set. Feature '%s' not enabled.", feature_name
             )
             return False
 
         stage_config = self.feature_config.get(feature_name, {}).get(stage, {})
         if not stage_config:
-            LOG.info(f"Stage '{stage}' not enabled for Feature '{feature_name}'.")
+            LOG.info("Stage '%s' not enabled for Feature '%s'.", stage, feature_name)
             return False
 
         if account_id in stage_config:
@@ -90,10 +90,10 @@ def is_enabled(self, feature_name: str) -> bool:
             region_config = stage_config[region] if region in stage_config else stage_config.get("default", {})
 
         dialup = self._get_dialup(region_config, feature_name=feature_name)  # type: ignore[no-untyped-call]
-        LOG.info(f"Using Dialip {dialup}")
+        LOG.info("Using Dialip %s", dialup)
         is_enabled: bool = dialup.is_enabled()
 
-        LOG.info(f"Feature '{feature_name}' is enabled: '{is_enabled}'")
+        LOG.info("Feature '%s' is enabled: '%r'", feature_name, is_enabled)
         return is_enabled
 
 
@@ -156,8 +156,8 @@ def __init__(self, application_id, environment_id, configuration_profile_id, app
             binary_config_string = response["Content"].read()
             self.feature_toggle_config = cast(Dict[str, Any], json.loads(binary_config_string.decode("utf-8")))
             LOG.info("Finished loading feature toggle config from AppConfig.")
-        except Exception as ex:
-            LOG.error(f"Failed to load config from AppConfig: {ex}. Using empty config.")
+        except Exception:
+            LOG.exception("Failed to load config from AppConfig. Using empty config.")
             # There is chance that AppConfig is not available in a particular region.
             self.feature_toggle_config = {}
 

samtranslator/intrinsics/actions.py

@@ -17,23 +17,23 @@ class Action(ABC):
     _resource_ref_separator = "."
     intrinsic_name: str
 
-    def resolve_parameter_refs(  # noqa: empty-method-without-abstract-decorator
+    def resolve_parameter_refs(  # noqa: B027
         self, input_dict: Optional[Any], parameters: Dict[str, Any]
     ) -> Optional[Any]:
         """
         Subclass optionally implement this method to resolve the intrinsic function
         TODO: input_dict should not be None.
         """
 
-    def resolve_resource_refs(  # noqa: empty-method-without-abstract-decorator
+    def resolve_resource_refs(  # noqa: B027
         self, input_dict: Optional[Any], supported_resource_refs: Dict[str, Any]
     ) -> Optional[Any]:
         """
         Subclass optionally implement this method to resolve resource references
         TODO: input_dict should not be None.
         """
 
-    def resolve_resource_id_refs(  # noqa: empty-method-without-abstract-decorator
+    def resolve_resource_id_refs(  # noqa: B027
         self, input_dict: Optional[Any], supported_resource_id_refs: Dict[str, Any]
     ) -> Optional[Any]:
         """

samtranslator/intrinsics/resolver.py

@@ -1,5 +1,5 @@
 # Help resolve intrinsic functions
-from typing import Any, Callable, Dict, List, Optional, Union
+from typing import Any, Callable, Dict, List, Optional, Union, cast
 
 from samtranslator.intrinsics.actions import Action, GetAttAction, RefAction, SubAction
 from samtranslator.intrinsics.resource_refs import SupportedResourceReferences
@@ -49,7 +49,7 @@ def resolve_parameter_refs(self, _input: Any) -> Any:
 
     def resolve_sam_resource_refs(
         self, _input: Dict[str, Any], supported_resource_refs: SupportedResourceReferences
-    ) -> Any:
+    ) -> Dict[str, Any]:
         """
         Customers can provide a reference to a "derived" SAM resource such as Alias of a Function or Stage of an API
         resource. This method recursively walks the tree, converting all derived references to the real resource name,
@@ -71,7 +71,10 @@ def resolve_sam_resource_refs(
             references supported in this SAM template, along with the value they should resolve to.
         :return list errors: List of dictionary containing information about invalid reference. Empty list otherwise
         """
-        return self._traverse(_input, supported_resource_refs, self._try_resolve_sam_resource_refs)
+        # The _traverse() return type is the same as the input. Here the input is Dict[str, Any]
+        return cast(
+            Dict[str, Any], self._traverse(_input, supported_resource_refs, self._try_resolve_sam_resource_refs)
+        )
 
     def resolve_sam_resource_id_refs(self, _input: Dict[str, Any], supported_resource_id_refs: Dict[str, str]) -> Any:
         """
@@ -195,7 +198,7 @@ def _try_resolve_parameter_refs(self, _input: Dict[str, Any], parameters: Dict[s
         if not self._is_intrinsic_dict(_input):
             return _input
 
-        function_type = list(_input.keys())[0]
+        function_type = next(iter(_input.keys()))
         return self.supported_intrinsics[function_type].resolve_parameter_refs(_input, parameters)
 
     def _try_resolve_sam_resource_refs(
@@ -214,7 +217,7 @@ def _try_resolve_sam_resource_refs(
         if not self._is_intrinsic_dict(_input):
             return _input
 
-        function_type = list(_input.keys())[0]
+        function_type = next(iter(_input.keys()))
         return self.supported_intrinsics[function_type].resolve_resource_refs(_input, supported_resource_refs)
 
     def _try_resolve_sam_resource_id_refs(
@@ -232,7 +235,7 @@ def _try_resolve_sam_resource_id_refs(
         if not self._is_intrinsic_dict(_input):
             return _input
 
-        function_type = list(_input.keys())[0]
+        function_type = next(iter(_input.keys()))
         return self.supported_intrinsics[function_type].resolve_resource_id_refs(_input, supported_resource_id_refs)
 
     def _is_intrinsic_dict(self, _input: Dict[str, Any]) -> bool:
@@ -243,4 +246,4 @@ def _is_intrinsic_dict(self, _input: Dict[str, Any]) -> bool:
         :return: True, if the _input contains a supported intrinsic function.  False otherwise
         """
         # All intrinsic functions are dictionaries with just one key
-        return isinstance(_input, dict) and len(_input) == 1 and list(_input.keys())[0] in self.supported_intrinsics
+        return isinstance(_input, dict) and len(_input) == 1 and next(iter(_input.keys())) in self.supported_intrinsics

samtranslator/metrics/metrics.py

@@ -63,8 +63,8 @@ def _flush_metrics(self, namespace, metrics):  # type: ignore[no-untyped-def]
         try:
             if metric_data:
                 self.cloudwatch_client.put_metric_data(Namespace=namespace, MetricData=metric_data)
-        except Exception as e:
-            LOG.exception(f"Failed to report {len(metric_data)} metrics", exc_info=e)
+        except Exception:
+            LOG.exception("Failed to report %d metrics", len(metric_data))
 
 
 class DummyMetricsPublisher(MetricsPublisher):
@@ -73,7 +73,7 @@ def __init__(self) -> None:
 
     def publish(self, namespace: str, metrics: List["MetricDatum"]) -> None:
         """Do not publish any metric, this is a dummy publisher used for offline use."""
-        LOG.debug(f"Dummy publisher ignoring {len(metrics)} metrices")
+        LOG.debug("Dummy publisher ignoring %d metrices", len(metrics))
 
 
 class Unit:

samtranslator/model/__init__.py

@@ -330,9 +330,7 @@ def __setattr__(self, name, value):  # type: ignore[no-untyped-def]
 
         raise InvalidResourceException(
             self.logical_id,
-            "property {property_name} not defined for resource of type {resource_type}".format(
-                resource_type=self.resource_type, property_name=name
-            ),
+            f"property {name} not defined for resource of type {self.resource_type}",
         )
 
     # Note: For compabitliy issue, we should ONLY use this with new abstraction/resources.
@@ -599,7 +597,7 @@ def __init__(self, *modules: Any) -> None:
             for _, resource_class in inspect.getmembers(
                 module,
                 lambda cls: inspect.isclass(cls)
-                and cls.__module__ == module.__name__  # noqa: function-uses-loop-variable
+                and cls.__module__ == module.__name__  # noqa: B023
                 and hasattr(cls, "resource_type"),
             ):
                 self.resource_types[resource_class.resource_type] = resource_class

samtranslator/model/api/api_generator.py

@@ -169,7 +169,7 @@ def _set_condition(self, condition, template_conditions):  # type: ignore[no-unt
 
 
 class ApiGenerator:
-    def __init__(  # noqa: too-many-arguments
+    def __init__(  # noqa: PLR0913
         self,
         logical_id: str,
         cache_cluster_enabled: Optional[Intrinsicable[bool]],
@@ -450,7 +450,7 @@ def _construct_stage(
 
         return stage
 
-    def _construct_api_domain(  # noqa: too-many-branches
+    def _construct_api_domain(  # noqa: PLR0912, PLR0915
         self, rest_api: ApiGatewayRestApi, route53_record_set_groups: Any
     ) -> ApiDomainResponse:
         """
@@ -856,7 +856,7 @@ def _add_auth(self) -> None:
 
         self.definition_body = self._openapi_postprocess(swagger_editor.swagger)
 
-    def _construct_usage_plan(self, rest_api_stage: Optional[ApiGatewayStage] = None) -> Any:  # noqa: too-many-branches
+    def _construct_usage_plan(self, rest_api_stage: Optional[ApiGatewayStage] = None) -> Any:  # noqa: PLR0912
         """Constructs and returns the ApiGateway UsagePlan, ApiGateway UsagePlanKey, ApiGateway ApiKey for Auth.
 
         :param model.apigateway.ApiGatewayStage stage: the stage of rest api
@@ -1044,16 +1044,14 @@ def _add_gateway_responses(self) -> None:
             if not isinstance(responses_value, dict):
                 raise InvalidResourceException(
                     self.logical_id,
-                    "Invalid property type '{}' for GatewayResponses. "
-                    "Expected an object of type 'GatewayResponse'.".format(type(responses_value).__name__),
+                    f"Invalid property type '{type(responses_value).__name__}' for GatewayResponses. "
+                    "Expected an object of type 'GatewayResponse'.",
                 )
             for response_key in responses_value:
                 if response_key not in GatewayResponseProperties:
                     raise InvalidResourceException(
                         self.logical_id,
-                        "Invalid property '{}' in 'GatewayResponses' property '{}'.".format(
-                            response_key, responses_key
-                        ),
+                        f"Invalid property '{response_key}' in 'GatewayResponses' property '{responses_key}'.",
                     )
 
         if not SwaggerEditor.is_valid(self.definition_body):
@@ -1119,7 +1117,7 @@ def _add_models(self) -> None:
 
         self.definition_body = self._openapi_postprocess(swagger_editor.swagger)
 
-    def _openapi_postprocess(self, definition_body: Dict[str, Any]) -> Dict[str, Any]:  # noqa: too-many-branches
+    def _openapi_postprocess(self, definition_body: Dict[str, Any]) -> Dict[str, Any]:  # noqa: PLR0912
         """
         Convert definitions to openapi 3 in definition body if OpenApiVersion flag is specified.
 
@@ -1168,8 +1166,8 @@ def _openapi_postprocess(self, definition_body: Dict[str, Any]) -> Dict[str, Any
                     if path_item.get("options"):
                         SwaggerEditor.validate_is_dict(
                             path_item.get("options"),
-                            "Value of options method for path {} must be a "
-                            "dictionary according to Swagger spec.".format(path),
+                            f"Value of options method for path {path} must be a "
+                            "dictionary according to Swagger spec.",
                         )
                         options = path_item.get("options").copy()
                         for field, field_val in options.items():
@@ -1192,8 +1190,8 @@ def _openapi_postprocess(self, definition_body: Dict[str, Any]) -> Dict[str, Any
                                     continue
                                 SwaggerEditor.validate_is_dict(
                                     response_200_headers,
-                                    "Value of response's headers in options method for path {} must be a "
-                                    "dictionary according to Swagger spec.".format(path),
+                                    f"Value of response's headers in options method for path {path} must be a "
+                                    "dictionary according to Swagger spec.",
                                 )
                                 for header, header_val in response_200_headers.items():
                                     new_header_val_with_schema = Py27Dict()