Fix mypy errors

Author: Michal Ploski

.github/workflows/python_build.yml

@@ -30,6 +30,8 @@ jobs:
         run: make dev
       - name: Formatting and Linting
         run: make lint
+      - name: Typing
+        run: make mypy
       - name: Test with pytest
         run: make test
       - name: Security baseline

aws_lambda_powertools/metrics/metrics.py

@@ -53,7 +53,7 @@ def lambda_handler():
     ----------
     service : str, optional
         service name to be used as metric dimension, by default "service_undefined"
-    namespace : str
+    namespace : str, optional
         Namespace for metrics
 
     Raises
@@ -209,5 +209,6 @@ def __add_cold_start_metric(self, context: Any) -> None:
             logger.debug("Adding cold start metric and function_name dimension")
             with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1, namespace=self.namespace) as metric:
                 metric.add_dimension(name="function_name", value=context.function_name)
-                metric.add_dimension(name="service", value=self.service)
+                if self.service:
+                    metric.add_dimension(name="service", value=str(self.service))
                 is_cold_start = False

aws_lambda_powertools/shared/functions.py

@@ -1,4 +1,4 @@
-from typing import Any, Optional, Union
+from typing import Optional, Union
 
 
 def strtobool(value: str) -> bool:
@@ -38,21 +38,23 @@ def resolve_truthy_env_var_choice(env: str, choice: Optional[bool] = None) -> bo
     return choice if choice is not None else strtobool(env)
 
 
-def resolve_env_var_choice(env: Any, choice: Optional[Any] = None) -> Union[bool, Any]:
+def resolve_env_var_choice(
+    env: Optional[str] = None, choice: Optional[Union[str, float]] = None
+) -> Optional[Union[str, float]]:
     """Pick explicit choice over env, if available, otherwise return env value received
 
     NOTE: Environment variable should be resolved by the caller.
 
     Parameters
     ----------
-    env : Any
+    env : str, Optional
         environment variable actual value
-    choice : bool
+    choice : str|float, optional
         explicit choice
 
     Returns
     -------
-    choice : str
+    choice : str, Optional
         resolved choice as either bool or environment value
     """
     return choice if choice is not None else env

aws_lambda_powertools/utilities/batch/exceptions.py

@@ -5,7 +5,7 @@
 from types import TracebackType
 from typing import List, Optional, Tuple, Type
 
-ExceptionInfo = Tuple[Type[BaseException], BaseException, TracebackType]
+ExceptionInfo = Tuple[Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]]
 
 
 class BaseBatchProcessingError(Exception):

aws_lambda_powertools/utilities/batch/sqs.py

@@ -5,11 +5,13 @@
 """
 import logging
 import sys
-from typing import Callable, Dict, List, Optional, Tuple
+from typing import Callable, Dict, List, Optional, Tuple, cast
 
 import boto3
 from botocore.config import Config
 
+from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
+
 from ...middleware_factory import lambda_handler_decorator
 from .base import BasePartialProcessor
 from .exceptions import SQSBatchProcessingError
@@ -84,11 +86,14 @@ def _get_queue_url(self) -> Optional[str]:
         *_, account_id, queue_name = self.records[0]["eventSourceARN"].split(":")
         return f"{self.client._endpoint.host}/{account_id}/{queue_name}"
 
-    def _get_entries_to_clean(self) -> List:
+    def _get_entries_to_clean(self) -> List[Dict[str, str]]:
         """
         Format messages to use in batch deletion
         """
-        return [{"Id": msg["messageId"], "ReceiptHandle": msg["receiptHandle"]} for msg in self.success_messages]
+        return [
+            {"Id": msg["messageId"], "ReceiptHandle": msg["receiptHandle"]}
+            for msg in cast(List[SQSRecord], self.success_messages)
+        ]
 
     def _process_record(self, record) -> Tuple:
         """

aws_lambda_powertools/utilities/idempotency/idempotency.py

@@ -112,7 +112,7 @@ def process_order(customer_id: str, order: dict, **kwargs):
             return {"StatusCode": 200}
     """
 
-    if function is None:
+    if not function:
         return cast(
             AnyCallableT,
             functools.partial(
@@ -132,7 +132,7 @@ def decorate(*args, **kwargs):
 
         payload = kwargs.get(data_keyword_argument)
 
-        if payload is None:
+        if not payload:
             raise RuntimeError(
                 f"Unable to extract '{data_keyword_argument}' from keyword arguments."
                 f" Ensure this exists in your function's signature as well as the caller used it as a keyword argument"

aws_lambda_powertools/utilities/idempotency/persistence/base.py

@@ -40,7 +40,7 @@ def __init__(
         idempotency_key,
         status: str = "",
         expiry_timestamp: Optional[int] = None,
-        response_data: Optional[str] = "",
+        response_data: str = "",
         payload_hash: Optional[str] = None,
     ) -> None:
         """
@@ -279,14 +279,14 @@ def _save_to_cache(self, data_record: DataRecord):
         -------
 
         """
-        if not self.use_local_cache:
+        if not self.use_local_cache or self._cache is None:
             return
         if data_record.status == STATUS_CONSTANTS["INPROGRESS"]:
             return
         self._cache[data_record.idempotency_key] = data_record
 
     def _retrieve_from_cache(self, idempotency_key: str):
-        if not self.use_local_cache:
+        if not self.use_local_cache or self._cache is None:
             return
         cached_record = self._cache.get(key=idempotency_key)
         if cached_record:
@@ -296,7 +296,7 @@ def _retrieve_from_cache(self, idempotency_key: str):
             self._delete_from_cache(idempotency_key=idempotency_key)
 
     def _delete_from_cache(self, idempotency_key: str):
-        if not self.use_local_cache:
+        if not self.use_local_cache or self._cache is None:
             return
         if idempotency_key in self._cache:
             del self._cache[idempotency_key]

aws_lambda_powertools/utilities/idempotency/persistence/dynamodb.py

@@ -130,7 +130,7 @@ def _item_to_data_record(self, item: Dict[str, Any]) -> DataRecord:
             idempotency_key=item[self.key_attr],
             status=item[self.status_attr],
             expiry_timestamp=item[self.expiry_attr],
-            response_data=item.get(self.data_attr),
+            response_data=item.get(self.data_attr, ""),
             payload_hash=item.get(self.validation_key_attr),
         )
 

aws_lambda_powertools/utilities/parameters/base.py

@@ -15,6 +15,7 @@
 ExpirableValue = namedtuple("ExpirableValue", ["value", "ttl"])
 # These providers will be dynamically initialized on first use of the helper functions
 DEFAULT_PROVIDERS: Dict[str, Any] = {}
+MULTIPLE_VALUES_TYPE: Union[Dict[str, str], Dict[str, dict], Dict[str, bytes], Dict[str, None]]
 TRANSFORM_METHOD_JSON = "json"
 TRANSFORM_METHOD_BINARY = "binary"
 SUPPORTED_TRANSFORM_METHODS = [TRANSFORM_METHOD_JSON, TRANSFORM_METHOD_BINARY]
@@ -44,7 +45,7 @@ def get(
         transform: Optional[str] = None,
         force_fetch: bool = False,
         **sdk_options,
-    ) -> Union[str, list, dict, bytes]:
+    ) -> Optional[Union[str, dict, bytes]]:
         """
         Retrieve a parameter value or return the cached value
 
@@ -81,6 +82,7 @@ def get(
         # of supported transform is small and the probability that a given
         # parameter will always be used in a specific transform, this should be
         # an acceptable tradeoff.
+        value: Optional[Union[str, bytes, dict]] = None
         key = (name, transform)
 
         if not force_fetch and self._has_not_expired(key):
@@ -92,12 +94,12 @@ def get(
         except Exception as exc:
             raise GetParameterError(str(exc))
 
-        if transform is not None:
+        if transform:
             if isinstance(value, bytes):
                 value = value.decode("utf-8")
             value = transform_value(value, transform)
-
-        self.store[key] = ExpirableValue(value, datetime.now() + timedelta(seconds=max_age))
+            if value:
+                self.store[key] = ExpirableValue(value, datetime.now() + timedelta(seconds=max_age))
 
         return value
 
@@ -146,26 +148,25 @@ def get_multiple(
         TransformParameterError
             When the parameter provider fails to transform a parameter value.
         """
-
         key = (path, transform)
 
         if not force_fetch and self._has_not_expired(key):
             return self.store[key].value
 
         try:
-            values: Dict[str, Union[str, bytes, dict, None]] = self._get_multiple(path, **sdk_options)
+            values = self._get_multiple(path, **sdk_options)
         # Encapsulate all errors into a generic GetParameterError
         except Exception as exc:
             raise GetParameterError(str(exc))
 
-        if transform is not None:
-            for (key, value) in values.items():
-                _transform = get_transform_method(key, transform)
-                if _transform is None:
+        if transform:
+            transformed_values: dict = {}
+            for (item, value) in values.items():
+                _transform = get_transform_method(item, transform)
+                if not _transform:
                     continue
-
-                values[key] = transform_value(value, _transform, raise_on_transform_error)
-
+                transformed_values[item] = transform_value(value, _transform, raise_on_transform_error)
+            values.update(transformed_values)
         self.store[key] = ExpirableValue(values, datetime.now() + timedelta(seconds=max_age))
 
         return values
@@ -217,7 +218,9 @@ def get_transform_method(key: str, transform: Optional[str] = None) -> Optional[
     return None
 
 
-def transform_value(value: str, transform: str, raise_on_transform_error: bool = True) -> Union[dict, bytes, None]:
+def transform_value(
+    value: str, transform: str, raise_on_transform_error: Optional[bool] = True
+) -> Optional[Union[dict, bytes]]:
     """
     Apply a transform to a value
 

aws_lambda_powertools/utilities/parameters/ssm.py

@@ -95,7 +95,7 @@ def get(
         decrypt: bool = False,
         force_fetch: bool = False,
         **sdk_options
-    ) -> Union[str, list, dict, bytes]:
+    ) -> Optional[Union[str, dict, bytes]]:
         """
         Retrieve a parameter value or return the cached value
 

aws_lambda_powertools/utilities/parser/envelopes/apigw.py

@@ -27,6 +27,6 @@ def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model])
             Parsed detail payload with model provided
         """
         logger.debug(f"Parsing incoming data with Api Gateway model {APIGatewayProxyEventModel}")
-        parsed_envelope = APIGatewayProxyEventModel.parse_obj(data)
+        parsed_envelope: APIGatewayProxyEventModel = APIGatewayProxyEventModel.parse_obj(data)
         logger.debug(f"Parsing event payload in `detail` with {model}")
         return self._parse(data=parsed_envelope.body, model=model)

aws_lambda_powertools/utilities/parser/envelopes/apigwv2.py

@@ -27,6 +27,6 @@ def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model])
             Parsed detail payload with model provided
         """
         logger.debug(f"Parsing incoming data with Api Gateway model V2 {APIGatewayProxyEventV2Model}")
-        parsed_envelope = APIGatewayProxyEventV2Model.parse_obj(data)
+        parsed_envelope: APIGatewayProxyEventV2Model = APIGatewayProxyEventV2Model.parse_obj(data)
         logger.debug(f"Parsing event payload in `detail` with {model}")
         return self._parse(data=parsed_envelope.body, model=model)

aws_lambda_powertools/utilities/parser/envelopes/event_bridge.py

@@ -27,6 +27,6 @@ def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model])
             Parsed detail payload with model provided
         """
         logger.debug(f"Parsing incoming data with EventBridge model {EventBridgeModel}")
-        parsed_envelope = EventBridgeModel.parse_obj(data)
+        parsed_envelope: EventBridgeModel = EventBridgeModel.parse_obj(data)
         logger.debug(f"Parsing event payload in `detail` with {model}")
         return self._parse(data=parsed_envelope.detail, model=model)

aws_lambda_powertools/utilities/parser/envelopes/sns.py

@@ -69,6 +69,6 @@ def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Type[Model])
         parsed_envelope = SqsModel.parse_obj(data)
         output = []
         for record in parsed_envelope.Records:
-            sns_notification = SnsNotificationModel.parse_raw(record.body)
+            sns_notification: SnsNotificationModel = SnsNotificationModel.parse_raw(record.body)
             output.append(self._parse(data=sns_notification.Message, model=model))
         return output

aws_lambda_powertools/utilities/parser/models/alb.py

@@ -1,4 +1,4 @@
-from typing import Dict, Union
+from typing import Dict, Generic, Union
 
 from pydantic import BaseModel
 
@@ -13,7 +13,7 @@ class AlbRequestContext(BaseModel):
     elb: AlbRequestContextData
 
 
-class AlbModel(BaseModel):
+class AlbModel(BaseModel, Generic[Model]):
     httpMethod: str
     path: str
     body: Union[str, Model]

aws_lambda_powertools/utilities/parser/models/apigw.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, Generic, List, Optional, Union
 
 from pydantic import BaseModel, root_validator
 from pydantic.networks import IPvAnyNetwork
@@ -76,7 +76,7 @@ def check_message_id(cls, values):
         return values
 
 
-class APIGatewayProxyEventModel(BaseModel):
+class APIGatewayProxyEventModel(BaseModel, Generic[Model]):
     version: Optional[str]
     resource: str
     path: str

aws_lambda_powertools/utilities/parser/models/apigwv2.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, Generic, List, Optional, Union
 
 from pydantic import BaseModel, Field
 from pydantic.networks import IPvAnyNetwork
@@ -56,7 +56,7 @@ class RequestContextV2(BaseModel):
     http: RequestContextV2Http
 
 
-class APIGatewayProxyEventV2Model(BaseModel):
+class APIGatewayProxyEventV2Model(BaseModel, Generic[Model]):
     version: str
     routeKey: str
     rawPath: str

aws_lambda_powertools/utilities/parser/models/cloudwatch.py

@@ -3,7 +3,7 @@
 import logging
 import zlib
 from datetime import datetime
-from typing import List, Union
+from typing import Generic, List, Union
 
 from pydantic import BaseModel, Field, validator
 
@@ -12,7 +12,7 @@
 logger = logging.getLogger(__name__)
 
 
-class CloudWatchLogsLogEvent(BaseModel):
+class CloudWatchLogsLogEvent(BaseModel, Generic[Model]):
     id: str  # noqa AA03 VNE003
     timestamp: datetime
     message: Union[str, Model]

aws_lambda_powertools/utilities/parser/models/dynamodb.py

@@ -1,12 +1,12 @@
 from datetime import date
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, Generic, List, Optional, Union
 
 from pydantic import BaseModel
 
 from aws_lambda_powertools.utilities.parser.types import Literal, Model
 
 
-class DynamoDBStreamChangedRecordModel(BaseModel):
+class DynamoDBStreamChangedRecordModel(BaseModel, Generic[Model]):
     ApproximateCreationDateTime: Optional[date]
     Keys: Dict[str, Dict[str, Any]]
     NewImage: Optional[Union[Dict[str, Any], Model]]

aws_lambda_powertools/utilities/parser/models/event_bridge.py

@@ -1,12 +1,12 @@
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, Generic, List, Optional, Union
 
 from pydantic import BaseModel, Field
 
 from aws_lambda_powertools.utilities.parser.types import Model
 
 
-class EventBridgeModel(BaseModel):
+class EventBridgeModel(BaseModel, Generic[Model]):
     version: str
     id: str  # noqa: A003,VNE003
     source: str