chore: Remove "type: ignore" in samtranslator/translator

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(

samtranslator/translator/managed_policy_translator.py

@@ -1,19 +1,21 @@
 import logging
-from typing import Dict, cast
+from typing import Dict, Optional, cast
+
+from botocore.client import BaseClient
 
 from samtranslator.metrics.method_decorator import cw_timer
 
 LOG = logging.getLogger(__name__)
 
 
 class ManagedPolicyLoader:
-    def __init__(self, iam_client):  # type: ignore[no-untyped-def]
+    def __init__(self, iam_client: BaseClient) -> None:
         self._iam_client = iam_client
-        self._policy_map = None
+        self._policy_map: Optional[Dict[str, str]] = None
         self.max_items = 1000
 
     @cw_timer(prefix="External", name="IAM")
-    def _load_policies_from_iam(self):  # type: ignore[no-untyped-def]
+    def _load_policies_from_iam(self) -> None:
         LOG.info("Loading policies from IAM...")
 
         paginator = self._iam_client.get_paginator("list_policies")
@@ -23,7 +25,7 @@ def _load_policies_from_iam(self):  # type: ignore[no-untyped-def]
         # Note(jfuss): boto3 PaginationConfig MaxItems does not control the number of items returned from the API
         # call. This is actually controlled by PageSize.
         page_iterator = paginator.paginate(Scope="AWS", PaginationConfig={"PageSize": self.max_items})
-        name_to_arn_map = {}  # type: ignore[var-annotated]
+        name_to_arn_map: Dict[str, str] = {}
 
         for page in page_iterator:
             name_to_arn_map.update(map(lambda x: (x["PolicyName"], x["Arn"]), page["Policies"]))

samtranslator/translator/transform.py

@@ -1,13 +1,20 @@
 from functools import lru_cache
-from typing import Dict
+from typing import Any, Dict, Optional
 
+from samtranslator.feature_toggle.feature_toggle import FeatureToggle
 from samtranslator.parser.parser import Parser
 from samtranslator.translator.managed_policy_translator import ManagedPolicyLoader
 from samtranslator.translator.translator import Translator
 from samtranslator.utils.py27hash_fix import to_py27_compatible_template, undo_mark_unicode_str_in_template
 
 
-def transform(input_fragment, parameter_values, managed_policy_loader: ManagedPolicyLoader, feature_toggle=None, passthrough_metadata=False):  # type: ignore[no-untyped-def]
+def transform(
+    input_fragment: Dict[str, Any],
+    parameter_values: Dict[str, Any],
+    managed_policy_loader: ManagedPolicyLoader,
+    feature_toggle: Optional[FeatureToggle] = None,
+    passthrough_metadata: Optional[bool] = False,
+) -> Dict[str, Any]:
     """Translates the SAM manifest provided in the and returns the translation to CloudFormation.
 
     :param dict input_fragment: the SAM template to transform
@@ -18,7 +25,7 @@ def transform(input_fragment, parameter_values, managed_policy_loader: ManagedPo
 
     sam_parser = Parser()
     to_py27_compatible_template(input_fragment, parameter_values)
-    translator = Translator(  # type: ignore[no-untyped-call]
+    translator = Translator(
         None,
         sam_parser,
     )
@@ -34,4 +41,4 @@ def get_managed_policy_map() -> Dict[str, str]:
         passthrough_metadata=passthrough_metadata,
         get_managed_policy_map=get_managed_policy_map,
     )
-    return undo_mark_unicode_str_in_template(transformed)  # type: ignore[no-untyped-call]
+    return undo_mark_unicode_str_in_template(transformed)