chore: Type improvements in http api generator (#2586)

Author: aahung

samtranslator/feature_toggle/feature_toggle.py

@@ -125,7 +125,7 @@ def config(self):  # type: ignore[no-untyped-def]
 class FeatureToggleAppConfigConfigProvider(FeatureToggleConfigProvider):
     """Feature toggle config provider which loads config from AppConfig."""
 
-    @cw_timer(prefix="External", name="AppConfig")  # type: ignore[no-untyped-call]
+    @cw_timer(prefix="External", name="AppConfig")  # type: ignore[misc]
     def __init__(self, application_id, environment_id, configuration_profile_id, app_config_client=None):  # type: ignore[no-untyped-def]
         FeatureToggleConfigProvider.__init__(self)
         try:

samtranslator/metrics/method_decorator.py

@@ -6,6 +6,7 @@
 from samtranslator.metrics.metrics import DummyMetricsPublisher, Metrics
 from samtranslator.model import Resource
 import logging
+from typing import Any, Callable, Optional, Union
 
 LOG = logging.getLogger(__name__)
 
@@ -74,7 +75,9 @@ def _send_cw_metric(prefix, name, execution_time_ms, func, args):  # type: ignor
         LOG.warning("Failed to add metrics", exc_info=e)
 
 
-def cw_timer(_func=None, name=None, prefix=None):  # type: ignore[no-untyped-def]
+def cw_timer(
+    _func: Optional[Callable[..., Any]] = None, name: Optional[str] = None, prefix: Optional[str] = None
+) -> Union[Callable[..., Any], Callable[[Callable[..., Any]], Callable[..., Any]]]:
     """
     A method decorator, that will calculate execution time of the decorated method, and store this information as a
     metric in CloudWatch by calling the metrics singleton instance.
@@ -87,7 +90,7 @@ def cw_timer(_func=None, name=None, prefix=None):  # type: ignore[no-untyped-def
     If prefix is defined, it will be added in the beginning of what is been generated above
     """
 
-    def cw_timer_decorator(func):  # type: ignore[no-untyped-def]
+    def cw_timer_decorator(func: Callable[..., Any]) -> Callable[..., Any]:
         @functools.wraps(func)
         def wrapper_cw_timer(*args, **kwargs):  # type: ignore[no-untyped-def]
             start_time = datetime.now()
@@ -102,4 +105,4 @@ def wrapper_cw_timer(*args, **kwargs):  # type: ignore[no-untyped-def]
 
         return wrapper_cw_timer
 
-    return cw_timer_decorator if _func is None else cw_timer_decorator(_func)  # type: ignore[no-untyped-call]
+    return cw_timer_decorator if _func is None else cw_timer_decorator(_func)

samtranslator/model/__init__.py

@@ -376,7 +376,7 @@ def resources_to_link(self, resources):  # type: ignore[no-untyped-def]
         """
         return {}
 
-    def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
+    def to_cloudformation(self, **kwargs: Any) -> List[Any]:
         """Returns a list of Resource instances, representing vanilla CloudFormation resources, to which this macro
         expands. The caller should be able to update their template with the expanded resources by calling
         :func:`to_dict` on each resource returned, then updating their "Resources" mapping with the results.
@@ -455,7 +455,7 @@ def _construct_tag_list(self, tags, additional_tags=None):  # type: ignore[no-un
         # tags list. Changing this ordering will trigger a update on Lambda Function resource. Even though this
         # does not change the actual content of the tags, we don't want to trigger update of a resource without
         # customer's knowledge.
-        return get_tag_list(sam_tag) + get_tag_list(additional_tags) + get_tag_list(tags)  # type: ignore[no-untyped-call]
+        return get_tag_list(sam_tag) + get_tag_list(additional_tags) + get_tag_list(tags)
 
     def _check_tag(self, reserved_tag_name, tags):  # type: ignore[no-untyped-def]
         if reserved_tag_name in tags:

samtranslator/model/api/api_generator.py

@@ -338,15 +338,15 @@ def _construct_body_s3_dict(self):  # type: ignore[no-untyped-def]
             s3_pointer = self.definition_uri
 
         else:
-
             # DefinitionUri is a string
-            s3_pointer = parse_s3_uri(self.definition_uri)  # type: ignore[no-untyped-call]
-            if s3_pointer is None:
+            _parsed_s3_pointer = parse_s3_uri(self.definition_uri)
+            if _parsed_s3_pointer is None:
                 raise InvalidResourceException(
                     self.logical_id,
                     "'DefinitionUri' is not a valid S3 Uri of the form "
                     "'s3://bucket/key' with optional versionId query parameter.",
                 )
+            s3_pointer = _parsed_s3_pointer
 
             if isinstance(self.definition_uri, Py27UniStr):
                 # self.defintion_uri is a Py27UniStr instance if it is defined in the template
@@ -394,8 +394,8 @@ def _construct_stage(self, deployment, swagger, redeploy_restapi_parameters):  #
         if stage_name_prefix.isalnum():
             stage_logical_id = self.logical_id + stage_name_prefix + "Stage"
         else:
-            generator = LogicalIdGenerator(self.logical_id + "Stage", stage_name_prefix)  # type: ignore[no-untyped-call]
-            stage_logical_id = generator.gen()  # type: ignore[no-untyped-call]
+            generator = LogicalIdGenerator(self.logical_id + "Stage", stage_name_prefix)
+            stage_logical_id = generator.gen()
         stage = ApiGatewayStage(stage_logical_id, attributes=self.passthrough_resource_attributes)
         stage.RestApiId = ref(self.logical_id)
         stage.update_deployment_ref(deployment.logical_id)  # type: ignore[no-untyped-call]
@@ -414,7 +414,7 @@ def _construct_stage(self, deployment, swagger, redeploy_restapi_parameters):  #
             )
 
         if self.tags is not None:
-            stage.Tags = get_tag_list(self.tags)  # type: ignore[no-untyped-call]
+            stage.Tags = get_tag_list(self.tags)
 
         return stage
 
@@ -432,7 +432,7 @@ def _construct_api_domain(self, rest_api, route53_record_set_groups):  # type: i
             )
 
         self.domain["ApiDomainName"] = "{}{}".format(
-            "ApiGatewayDomainName", LogicalIdGenerator("", self.domain.get("DomainName")).gen()  # type: ignore[no-untyped-call, no-untyped-call]
+            "ApiGatewayDomainName", LogicalIdGenerator("", self.domain.get("DomainName")).gen()
         )
 
         domain = ApiGatewayDomainName(self.domain.get("ApiDomainName"), attributes=self.passthrough_resource_attributes)
@@ -537,7 +537,7 @@ def _construct_api_domain(self, rest_api, route53_record_set_groups):  # type: i
                     "HostedZoneId or HostedZoneName is required to enable Route53 support on Custom Domains.",
                 )
 
-            logical_id_suffix = LogicalIdGenerator(  # type: ignore[no-untyped-call, no-untyped-call]
+            logical_id_suffix = LogicalIdGenerator(
                 "", route53.get("HostedZoneId") or route53.get("HostedZoneName")
             ).gen()
             logical_id = "RecordSetGroup" + logical_id_suffix
@@ -593,7 +593,7 @@ def _construct_alias_target(self, domain):  # type: ignore[no-untyped-def]
             alias_target["DNSName"] = route53.get("DistributionDomainName")
         return alias_target
 
-    @cw_timer(prefix="Generator", name="Api")  # type: ignore[no-untyped-call]
+    @cw_timer(prefix="Generator", name="Api")
     def to_cloudformation(self, redeploy_restapi_parameters, route53_record_set_groups):  # type: ignore[no-untyped-def]
         """Generates CloudFormation resources from a SAM API resource