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,11 +19,11 @@ 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
-mypy~=1.1.0
+mypy~=1.3.0
 
 # types
 boto3-stubs[appconfig,serverlessrepo]>=1.19.5,==1.*

ruff.toml

@@ -30,6 +30,11 @@ 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",
+    # Logging statement uses f-string
+    "G004",
 ]
 
 # Mininal python version we support is 3.7

samtranslator/feature_toggle/feature_toggle.py

@@ -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:
         """
@@ -112,6 +115,8 @@ def _traverse(
         :param resolver_method: Method that will be called to actually resolve an intrinsic function. This method
             is called with the parameters `(input, resolution_data)`.
         :return: Modified `input` with intrinsics resolved
+
+        TODO: type this and make _traverse generic.
         """
 
         # There is data to help with resolution. Skip the traversal altogether
@@ -195,7 +200,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 +219,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 +237,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 +248,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(f"Failed to report {len(metric_data)} metrics")
 
 
 class DummyMetricsPublisher(MetricsPublisher):

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()

samtranslator/model/api/http_api_generator.py

@@ -35,7 +35,7 @@
 
 
 class HttpApiGenerator:
-    def __init__(  # noqa: too-many-arguments
+    def __init__(  # noqa: PLR0913
         self,
         logical_id: str,
         stage_variables: Optional[Dict[str, Intrinsicable[str]]],
@@ -239,7 +239,7 @@ def _update_default_path(self) -> None:
         if DefaultStageName in paths:
             paths[f"/{DefaultStageName}"] = paths.pop(DefaultStageName)
 
-    def _construct_api_domain(  # noqa: too-many-branches
+    def _construct_api_domain(  # noqa: PLR0912, PLR0915
         self, http_api: ApiGatewayV2HttpApi, route53_record_set_groups: Dict[str, Route53RecordSetGroup]
     ) -> Tuple[
         Optional[ApiGatewayV2DomainName],

samtranslator/model/apigateway.py

@@ -92,7 +92,7 @@ class ApiGatewayDeployment(Resource):
 
     runtime_attrs = {"deployment_id": lambda self: ref(self.logical_id)}
 
-    def make_auto_deployable(  # noqa: too-many-arguments
+    def make_auto_deployable(  # noqa: PLR0913
         self,
         stage: ApiGatewayStage,
         openapi_version: Optional[Union[Dict[str, Any], str]] = None,
@@ -281,7 +281,7 @@ class ApiGatewayApiKey(Resource):
 class ApiGatewayAuthorizer:
     _VALID_FUNCTION_PAYLOAD_TYPES = [None, "TOKEN", "REQUEST"]
 
-    def __init__(  # type: ignore[no-untyped-def]# noqa: too-many-arguments
+    def __init__(  # type: ignore[no-untyped-def]# noqa: PLR0913
         self,
         api_logical_id=None,
         name=None,