chore: Remove "type: ignore" in samtranslator/translator (#2947)

Author: aahung

bin/sam-translate.py

@@ -106,7 +106,7 @@ def transform_template(input_file_path, output_file_path):  # type: ignore[no-un
         sam_template = yaml_parse(f)  # type: ignore[no-untyped-call]
 
     try:
-        cloud_formation_template = transform(sam_template, {}, ManagedPolicyLoader(iam_client))  # type: ignore[no-untyped-call]
+        cloud_formation_template = transform(sam_template, {}, ManagedPolicyLoader(iam_client))
         cloud_formation_template_prettified = json.dumps(cloud_formation_template, indent=1)
 
         with open(output_file_path, "w") as f:

samtranslator/feature_toggle/feature_toggle.py

@@ -1,7 +1,7 @@
 import json
 import logging
 from abc import ABC, abstractmethod
-from typing import Any, Dict, cast
+from typing import Any, Dict, Optional, cast
 
 import boto3
 from botocore.config import Config
@@ -28,7 +28,13 @@ class FeatureToggle:
         "account-percentile": SimpleAccountPercentileDialup,
     }
 
-    def __init__(self, config_provider, stage, account_id, region):  # type: ignore[no-untyped-def]
+    def __init__(
+        self,
+        config_provider: "FeatureToggleConfigProvider",
+        stage: Optional[str],
+        account_id: Optional[str],
+        region: Optional[str],
+    ) -> None:
         self.feature_config = config_provider.config
         self.stage = stage
         self.account_id = account_id

samtranslator/metrics/method_decorator.py

@@ -23,7 +23,7 @@ class MetricsMethodWrapperSingleton:
     This singleton will be alive until lambda receives shutdown event
     """
 
-    _DUMMY_INSTANCE = Metrics("ServerlessTransform", DummyMetricsPublisher())  # type: ignore[no-untyped-call, no-untyped-call]
+    _DUMMY_INSTANCE = Metrics("ServerlessTransform", DummyMetricsPublisher())
     _METRICS_INSTANCE = _DUMMY_INSTANCE
 
     @staticmethod

samtranslator/metrics/metrics.py

@@ -4,7 +4,7 @@
 import logging
 from abc import ABC, abstractmethod
 from datetime import datetime
-from typing import Any, Dict
+from typing import Any, Dict, List, Optional
 
 LOG = logging.getLogger(__name__)
 
@@ -13,7 +13,7 @@ class MetricsPublisher(ABC):
     """Interface for all MetricPublishers"""
 
     @abstractmethod
-    def publish(self, namespace, metrics):  # type: ignore[no-untyped-def]
+    def publish(self, namespace: str, metrics: List["MetricDatum"]) -> None:
         """
         Abstract method to publish all metrics to CloudWatch
 
@@ -117,15 +117,17 @@ def get_metric_data(self) -> Dict[str, Any]:
 
 
 class Metrics:
-    def __init__(self, namespace="ServerlessTransform", metrics_publisher=None):  # type: ignore[no-untyped-def]
+    def __init__(
+        self, namespace: str = "ServerlessTransform", metrics_publisher: Optional[MetricsPublisher] = None
+    ) -> None:
         """
         Constructor
 
         :param namespace: namespace under which all metrics will be published
         :param metrics_publisher: publisher to publish all metrics
         """
         self.metrics_publisher = metrics_publisher if metrics_publisher else DummyMetricsPublisher()
-        self.metrics_cache = {}
+        self.metrics_cache: Dict[str, List[MetricDatum]] = {}
         self.namespace = namespace
 
     def __del__(self) -> None:

samtranslator/parser/parser.py

@@ -1,11 +1,13 @@
 import logging
+from typing import Any, Dict
 
 from samtranslator.model.exceptions import (
     InvalidDocumentException,
     InvalidResourceAttributeTypeException,
     InvalidTemplateException,
 )
 from samtranslator.plugins import LifeCycleEvents
+from samtranslator.plugins.sam_plugins import SamPlugins
 from samtranslator.public.sdk.template import SamTemplate
 from samtranslator.validator.validator import SamTemplateValidator
 from samtranslator.validator.value_validator import sam_expect
@@ -17,7 +19,7 @@ class Parser:
     def __init__(self) -> None:
         pass
 
-    def parse(self, sam_template, parameter_values, sam_plugins):  # type: ignore[no-untyped-def]
+    def parse(self, sam_template: Dict[str, Any], parameter_values: Dict[str, Any], sam_plugins: SamPlugins) -> None:
         self._validate(sam_template, parameter_values)  # type: ignore[no-untyped-call]
         sam_plugins.act(LifeCycleEvents.before_transform_template, sam_template)
 

samtranslator/plugins/application/serverless_app_plugin.py

@@ -2,9 +2,10 @@
 import json
 import logging
 from time import sleep
-from typing import Any, Dict, Tuple
+from typing import Any, Dict, List, Optional, Tuple
 
 import boto3
+from botocore.client import BaseClient
 from botocore.config import Config
 from botocore.exceptions import ClientError, EndpointConnectionError
 
@@ -50,7 +51,13 @@ class ServerlessAppPlugin(BasePlugin):
     LOCATION_KEY = "Location"
     TEMPLATE_URL_KEY = "TemplateUrl"
 
-    def __init__(self, sar_client=None, wait_for_template_active_status=False, validate_only=False, parameters=None):  # type: ignore[no-untyped-def]
+    def __init__(
+        self,
+        sar_client: Optional[BaseClient] = None,
+        wait_for_template_active_status: bool = False,
+        validate_only: bool = False,
+        parameters: Optional[Dict[str, Any]] = None,
+    ):
         """
         Initialize the plugin.
 
@@ -62,9 +69,9 @@ def __init__(self, sar_client=None, wait_for_template_active_status=False, valid
         super().__init__()
         if parameters is None:
             parameters = {}
-        self._applications = {}
-        self._in_progress_templates = []
-        self._sar_client = sar_client
+        self._applications: Dict[Tuple[str, str], Any] = {}
+        self._in_progress_templates: List[Tuple[str, str]] = []
+        self.__sar_client = sar_client
         self._wait_for_template_active_status = wait_for_template_active_status
         self._validate_only = validate_only
         self._parameters = parameters
@@ -75,6 +82,15 @@ def __init__(self, sar_client=None, wait_for_template_active_status=False, valid
             message = "Cannot set both validate_only and wait_for_template_active_status flags to True."
             raise InvalidPluginException(ServerlessAppPlugin.__name__, message)
 
+    @property
+    def _sar_client(self) -> BaseClient:
+        # Lazy initialization of the client- create it when it is needed
+        if not self.__sar_client:
+            # a SAR call could take a while to finish, leaving the read_timeout default (60s).
+            client_config = Config(connect_timeout=BOTO3_CONNECT_TIMEOUT)
+            self.__sar_client = boto3.client("serverlessrepo", config=client_config)
+        return self.__sar_client
+
     @staticmethod
     def _make_app_key(app_id: Any, semver: Any) -> Tuple[str, str]:
         """Generate a key that is always hashable."""
@@ -127,11 +143,6 @@ def on_before_transform_template(self, template_dict):  # type: ignore[no-untype
                         raise InvalidResourceException(
                             logical_id, "Serverless Application Repository is not available in this region."
                         )
-                    # Lazy initialization of the client- create it when it is needed
-                    if not self._sar_client:
-                        # a SAR call could take a while to finish, leaving the read_timeout default (60s).
-                        client_config = Config(connect_timeout=BOTO3_CONNECT_TIMEOUT)
-                        self._sar_client = boto3.client("serverlessrepo", config=client_config)
                     self._make_service_call_with_retry(service_call, app_id, semver, key, logical_id)  # type: ignore[no-untyped-call]
                 except InvalidResourceException as e:
                     # Catch all InvalidResourceExceptions, raise those in the before_resource_transform target.
@@ -421,17 +432,17 @@ def _resource_is_supported(self, resource_type):  # type: ignore[no-untyped-def]
         return resource_type == self.SUPPORTED_RESOURCE_TYPE
 
     def _get_application(self, app_id, semver):  # type: ignore[no-untyped-def]
-        return self._sar_client.get_application(
+        return self._sar_client.get_application(  # type: ignore[attr-defined]
             ApplicationId=self._sanitize_sar_str_param(app_id), SemanticVersion=self._sanitize_sar_str_param(semver)  # type: ignore[no-untyped-call]
         )
 
     def _create_cfn_template(self, app_id, semver):  # type: ignore[no-untyped-def]
-        return self._sar_client.create_cloud_formation_template(
+        return self._sar_client.create_cloud_formation_template(  # type: ignore[attr-defined]
             ApplicationId=self._sanitize_sar_str_param(app_id), SemanticVersion=self._sanitize_sar_str_param(semver)  # type: ignore[no-untyped-call]
         )
 
     def _get_cfn_template(self, app_id, template_id):  # type: ignore[no-untyped-def]
-        return self._sar_client.get_cloud_formation_template(
+        return self._sar_client.get_cloud_formation_template(  # type: ignore[attr-defined]
             ApplicationId=self._sanitize_sar_str_param(app_id),  # type: ignore[no-untyped-call]
             TemplateId=self._sanitize_sar_str_param(template_id),  # type: ignore[no-untyped-call]
         )

samtranslator/plugins/policies/policy_templates_plugin.py

@@ -4,6 +4,7 @@
 from samtranslator.model.resource_policies import PolicyTypes, ResourcePolicies
 from samtranslator.plugins import BasePlugin
 from samtranslator.policy_template_processor.exceptions import InsufficientParameterValues, InvalidParameterValues
+from samtranslator.policy_template_processor.processor import PolicyTemplatesProcessor
 
 
 class PolicyTemplatesForResourcePlugin(BasePlugin):
@@ -17,7 +18,7 @@ class PolicyTemplatesForResourcePlugin(BasePlugin):
     _plugin_name = ""
     SUPPORTED_RESOURCE_TYPE = {"AWS::Serverless::Function", "AWS::Serverless::StateMachine"}
 
-    def __init__(self, policy_template_processor):  # type: ignore[no-untyped-def]
+    def __init__(self, policy_template_processor: PolicyTemplatesProcessor) -> None:
         """
         Initialize the plugin.
 

samtranslator/region_configuration.py

@@ -47,7 +47,7 @@ def is_service_supported(cls, service, region=None):  # type: ignore[no-untyped-
             # need to handle when region is None so that it won't break
             if region is None:
                 if ArnGenerator.BOTO_SESSION_REGION_NAME is not None:
-                    region = ArnGenerator.BOTO_SESSION_REGION_NAME  # type: ignore[unreachable]
+                    region = ArnGenerator.BOTO_SESSION_REGION_NAME
                 else:
                     raise NoRegionFound("AWS Region cannot be found")
 

samtranslator/sdk/parameter.py

@@ -1,7 +1,8 @@
 import copy
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 
 import boto3
+from boto3 import Session
 
 from samtranslator.translator.arn_generator import ArnGenerator, NoRegionFound
 
@@ -64,7 +65,7 @@ def add_default_parameter_values(self, sam_template: Dict[str, Any]) -> Any:
 
         return None
 
-    def add_pseudo_parameter_values(self, session=None):  # type: ignore[no-untyped-def]
+    def add_pseudo_parameter_values(self, session: Optional[Session] = None) -> None:
         """
         Add pseudo parameter values
         :return: parameter values that have pseudo parameter in it

samtranslator/translator/arn_generator.py

@@ -30,7 +30,7 @@ def _region_to_partition(region: str) -> str:
 
 
 class ArnGenerator:
-    BOTO_SESSION_REGION_NAME = None
+    BOTO_SESSION_REGION_NAME: Optional[str] = None
 
     @classmethod
     def generate_arn(