chore: Upgrade mypy to 1.0.0 and type hint cw_timer

requirements/dev.txt

@@ -24,7 +24,7 @@ black==20.8b1
 ruamel.yaml==0.17.21  # It can parse yaml while perserving comments
 
 # type check
-mypy==0.971
+mypy~=1.0.0
 
 # types
 boto3-stubs[appconfig,serverlessrepo]>=1.19.5,==1.*

samtranslator/feature_toggle/feature_toggle.py

@@ -127,7 +127,7 @@ def config(self) -> Dict[str, Any]:
 class FeatureToggleAppConfigConfigProvider(FeatureToggleConfigProvider):
     """Feature toggle config provider which loads config from AppConfig."""
 
-    @cw_timer(prefix="External", name="AppConfig")  # type: ignore[misc]
+    @cw_timer(prefix="External", name="AppConfig")
     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

@@ -4,13 +4,15 @@
 import functools
 import logging
 from datetime import datetime
-from typing import Any, Callable, Optional, Union
+from typing import Callable, Optional, TypeVar, Union, overload
 
 from samtranslator.metrics.metrics import DummyMetricsPublisher, Metrics
 from samtranslator.model import Resource
 
 LOG = logging.getLogger(__name__)
 
+_RT = TypeVar("_RT")  # return value
+
 
 class MetricsMethodWrapperSingleton:
     """
@@ -76,9 +78,21 @@ def _send_cw_metric(prefix, name, execution_time_ms, func, args):  # type: ignor
         LOG.warning("Failed to add metrics", exc_info=e)
 
 
+@overload
+def cw_timer(
+    *, name: Optional[str] = None, prefix: Optional[str] = None
+) -> Callable[[Callable[..., _RT]], Callable[..., _RT]]:
+    ...
+
+
+@overload
+def cw_timer(_func: Callable[..., _RT], name: Optional[str] = None, prefix: Optional[str] = None) -> Callable[..., _RT]:
+    ...
+
+
 def cw_timer(
-    _func: Optional[Callable[..., Any]] = None, name: Optional[str] = None, prefix: Optional[str] = None
-) -> Union[Callable[..., Any], Callable[[Callable[..., Any]], Callable[..., Any]]]:
+    _func: Optional[Callable[..., _RT]] = None, name: Optional[str] = None, prefix: Optional[str] = None
+) -> Union[Callable[..., _RT], Callable[[Callable[..., _RT]], Callable[..., _RT]]]:
     """
     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.
@@ -91,9 +105,9 @@ def cw_timer(
     If prefix is defined, it will be added in the beginning of what is been generated above
     """
 
-    def cw_timer_decorator(func: Callable[..., Any]) -> Callable[..., Any]:
+    def cw_timer_decorator(func: Callable[..., _RT]) -> Callable[..., _RT]:
         @functools.wraps(func)
-        def wrapper_cw_timer(*args, **kwargs):  # type: ignore[no-untyped-def]
+        def wrapper_cw_timer(*args, **kwargs) -> _RT:  # type: ignore[no-untyped-def]
             start_time = datetime.now()
 
             exec_result = func(*args, **kwargs)

samtranslator/model/__init__.py

@@ -1,7 +1,7 @@
 """ CloudFormation Resource serialization, deserialization, and validation """
 import inspect
 import re
-from abc import ABC, ABCMeta
+from abc import ABC, ABCMeta, abstractmethod
 from typing import Any, Callable, Dict, List, Optional, Tuple, Union
 
 from samtranslator.intrinsics.resolver import IntrinsicsResolver
@@ -409,6 +409,7 @@ def resources_to_link(self, resources):  # type: ignore[no-untyped-def]
         """
         return {}
 
+    @abstractmethod
     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

samtranslator/model/api/http_api_generator.py

@@ -712,7 +712,7 @@ def _add_title(self) -> None:
         open_api_editor.add_title(self.name)
         self.definition_body = open_api_editor.openapi
 
-    @cw_timer(prefix="Generator", name="HttpApi")  # type: ignore[misc]
+    @cw_timer(prefix="Generator", name="HttpApi")
     def to_cloudformation(
         self, route53_record_set_groups: Dict[str, Route53RecordSetGroup]
     ) -> Tuple[

samtranslator/model/eventsources/scheduler.py

@@ -73,7 +73,7 @@ class SchedulerEventSource(ResourceMacro):
 
     DEFAULT_FLEXIBLE_TIME_WINDOW = {"Mode": "OFF"}
 
-    @cw_timer(prefix=FUNCTION_EVETSOURCE_METRIC_PREFIX)  # type: ignore
+    @cw_timer(prefix=FUNCTION_EVETSOURCE_METRIC_PREFIX)
     def to_cloudformation(self, **kwargs: Dict[str, Any]) -> List[Resource]:
         """Returns the Scheduler Schedule and an IAM role.
 

samtranslator/model/sam_resources.py

@@ -1366,7 +1366,7 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
         :returns: a list of vanilla CloudFormation Resources, to which this Function expands
         :rtype: list
         """
-        resources = []
+        resources: List[Resource] = []
         intrinsics_resolver = kwargs["intrinsics_resolver"]
         self.CorsConfiguration = intrinsics_resolver.resolve_parameter_refs(self.CorsConfiguration)
         self.Domain = intrinsics_resolver.resolve_parameter_refs(self.Domain)
@@ -1816,7 +1816,7 @@ class SamConnector(SamResourceMacro):
     }
 
     @cw_timer
-    def to_cloudformation(self, **kwargs: Any) -> List[Resource]:  # type: ignore
+    def to_cloudformation(self, **kwargs: Any) -> List[Resource]:
         resource_resolver: ResourceResolver = kwargs["resource_resolver"]
         original_template = kwargs["original_template"]
 

tests/test_model.py

@@ -1,3 +1,5 @@
+from typing import Any, List
+
 import pytest
 
 from unittest import TestCase
@@ -220,6 +222,9 @@ class NewSamResource(SamResourceMacro):
             property_types = {}
             referable_properties = {"prop1": "resource_type1", "prop2": "resource_type2", "prop3": "resource_type3"}
 
+            def to_cloudformation(self, **kwargs: Any) -> List[Any]:
+                return []
+
         sam_resource = NewSamResource("SamLogicalId")
 
         cfn_resources = [self.ResourceType1("logicalId1"), self.ResourceType2("logicalId2")]
@@ -241,6 +246,9 @@ class NewSamResource(SamResourceMacro):
             property_types = {}
             referable_properties = {"prop1": "resource_type1", "prop2": "resource_type2", "prop3": "resource_type3"}
 
+            def to_cloudformation(self, **kwargs: Any) -> List[Any]:
+                return []
+
         sam_resource1 = NewSamResource("SamLogicalId1")
         sam_resource2 = NewSamResource("SamLogicalId2")
 
@@ -267,6 +275,9 @@ class NewSamResource(SamResourceMacro):
             property_types = {}
             referable_properties = {"prop1": "foo", "prop2": "bar"}
 
+            def to_cloudformation(self, **kwargs: Any) -> List[Any]:
+                return []
+
         sam_resource = NewSamResource("SamLogicalId")
 
         # None of the CFN resource types are in the referable list
@@ -282,6 +293,9 @@ class NewSamResource(SamResourceMacro):
             property_types = {}
             referable_properties = {}
 
+            def to_cloudformation(self, **kwargs: Any) -> List[Any]:
+                return []
+
         sam_resource = NewSamResource("SamLogicalId")
 
         cfn_resources = [self.ResourceType1("logicalId1"), self.ResourceType2("logicalId2")]
@@ -296,6 +310,9 @@ class NewSamResource(SamResourceMacro):
             property_types = {}
             referable_properties = {"prop1": "resource_type1"}
 
+            def to_cloudformation(self, **kwargs: Any) -> List[Any]:
+                return []
+
         sam_resource = NewSamResource("SamLogicalId")
 
         cfn_resources = []
@@ -310,6 +327,9 @@ class NewSamResource(SamResourceMacro):
             property_types = {}
             referable_properties = {"prop1": "resource_type1"}
 
+            def to_cloudformation(self, **kwargs: Any) -> List[Any]:
+                return []
+
         sam_resource = NewSamResource("SamLogicalId")
 
         cfn_resources = [self.ResourceType1("logicalId1")]