fix(metrics): AttributeError raised by MediaManager and Typing and docs (#357)

* chore(data-classes): Typing and docs

Changes:
- Correct typing to make mypy happier
- Correct highlighted lines

* fix(metrics): AttributeError raises when setting metadata_set

* chore: Correct the typing as suggested by mypy

* chore: Set boto types as Any

* chore(batch): Correct typing for processor

* fix(parser): Check python version_info for typing
Changes:
- Use sys.version_info to determine where we need typing_extensions

Reference: https://mypy.readthedocs.io/en/stable/runtime_troubles.html#using-new-additions-to-the-typing-module

* docs(validation): Correct typing for parameters

* chore: Code review changes

Author: Michael Brewer

aws_lambda_powertools/logging/lambda_context.py

@@ -1,3 +1,6 @@
+from typing import Any
+
+
 class LambdaContextModel:
     """A handful of Lambda Runtime Context fields
 
@@ -31,7 +34,7 @@ def __init__(
         self.function_request_id = function_request_id
 
 
-def build_lambda_context_model(context: object) -> LambdaContextModel:
+def build_lambda_context_model(context: Any) -> LambdaContextModel:
     """Captures Lambda function runtime info to be used across all log statements
 
     Parameters

aws_lambda_powertools/logging/logger.py

@@ -4,7 +4,7 @@
 import os
 import random
 import sys
-from typing import Any, Callable, Dict, Union
+from typing import Any, Callable, Dict, Optional, Union
 
 import jmespath
 
@@ -318,12 +318,12 @@ def set_correlation_id(self, value: str):
         self.structure_logs(append=True, correlation_id=value)
 
     @staticmethod
-    def _get_log_level(level: Union[str, int]) -> Union[str, int]:
+    def _get_log_level(level: Union[str, int, None]) -> Union[str, int]:
         """ Returns preferred log level set by the customer in upper case """
         if isinstance(level, int):
             return level
 
-        log_level: str = level or os.getenv("LOG_LEVEL")
+        log_level: Optional[str] = level or os.getenv("LOG_LEVEL")
         if log_level is None:
             return logging.INFO
 

aws_lambda_powertools/metrics/base.py

@@ -88,7 +88,7 @@ def __init__(
         self.service = resolve_env_var_choice(choice=service, env=os.getenv(constants.SERVICE_NAME_ENV))
         self._metric_units = [unit.value for unit in MetricUnit]
         self._metric_unit_options = list(MetricUnit.__members__)
-        self.metadata_set = self.metadata_set if metadata_set is not None else {}
+        self.metadata_set = metadata_set if metadata_set is not None else {}
 
     def add_metric(self, name: str, unit: Union[MetricUnit, str], value: float):
         """Adds given metric

aws_lambda_powertools/metrics/metric.py

@@ -1,7 +1,7 @@
 import json
 import logging
 from contextlib import contextmanager
-from typing import Dict
+from typing import Dict, Optional, Union
 
 from .base import MetricManager, MetricUnit
 
@@ -42,7 +42,7 @@ class SingleMetric(MetricManager):
         Inherits from `aws_lambda_powertools.metrics.base.MetricManager`
     """
 
-    def add_metric(self, name: str, unit: MetricUnit, value: float):
+    def add_metric(self, name: str, unit: Union[MetricUnit, str], value: float):
         """Method to prevent more than one metric being created
 
         Parameters
@@ -109,11 +109,11 @@ def single_metric(name: str, unit: MetricUnit, value: float, namespace: str = No
     SchemaValidationError
         When metric object fails EMF schema validation
     """
-    metric_set = None
+    metric_set: Optional[Dict] = None
     try:
         metric: SingleMetric = SingleMetric(namespace=namespace)
         metric.add_metric(name=name, unit=unit, value=value)
         yield metric
-        metric_set: Dict = metric.serialize_metric_set()
+        metric_set = metric.serialize_metric_set()
     finally:
         print(json.dumps(metric_set, separators=(",", ":")))

aws_lambda_powertools/metrics/metrics.py

@@ -2,7 +2,7 @@
 import json
 import logging
 import warnings
-from typing import Any, Callable
+from typing import Any, Callable, Dict, Optional
 
 from .base import MetricManager, MetricUnit
 from .metric import single_metric
@@ -71,15 +71,15 @@ def do_something():
         When metric object fails EMF schema validation
     """
 
-    _metrics = {}
-    _dimensions = {}
-    _metadata = {}
+    _metrics: Dict[str, Any] = {}
+    _dimensions: Dict[str, str] = {}
+    _metadata: Dict[str, Any] = {}
 
     def __init__(self, service: str = None, namespace: str = None):
         self.metric_set = self._metrics
         self.dimension_set = self._dimensions
         self.service = service
-        self.namespace = namespace
+        self.namespace: Optional[str] = namespace
         self.metadata_set = self._metadata
 
         super().__init__(

aws_lambda_powertools/shared/functions.py

@@ -1,5 +1,5 @@
 from distutils.util import strtobool
-from typing import Any, Union
+from typing import Any, Optional, Union
 
 
 def resolve_truthy_env_var_choice(env: Any, choice: bool = None) -> bool:
@@ -22,7 +22,7 @@ def resolve_truthy_env_var_choice(env: Any, choice: bool = None) -> bool:
     return choice if choice is not None else strtobool(env)
 
 
-def resolve_env_var_choice(env: Any, choice: bool = None) -> Union[bool, Any]:
+def resolve_env_var_choice(env: Any, choice: Optional[Any] = None) -> Union[bool, Any]:
     """Pick explicit choice over env, if available, otherwise return env value received
 
     NOTE: Environment variable should be resolved by the caller.

aws_lambda_powertools/utilities/batch/base.py

@@ -104,7 +104,7 @@ def failure_handler(self, record: Any, exception: Tuple):
 
 @lambda_handler_decorator
 def batch_processor(
-    handler: Callable, event: Dict, context: Dict, record_handler: Callable, processor: BasePartialProcessor = None
+    handler: Callable, event: Dict, context: Dict, record_handler: Callable, processor: BasePartialProcessor
 ):
     """
     Middleware to handle batch event processing

aws_lambda_powertools/utilities/batch/sqs.py

@@ -71,7 +71,7 @@ def _get_queue_url(self) -> Optional[str]:
         Format QueueUrl from first records entry
         """
         if not getattr(self, "records", None):
-            return
+            return None
 
         *_, account_id, queue_name = self.records[0]["eventSourceARN"].split(":")
         return f"{self.client._endpoint.host}/{account_id}/{queue_name}"

aws_lambda_powertools/utilities/data_classes/appsync/resolver_utils.py

@@ -27,8 +27,8 @@ def register_resolver(func):
 
         return register_resolver
 
-    def resolve(self, event: dict, context: LambdaContext) -> Any:
-        event = AppSyncResolverEvent(event)
+    def resolve(self, _event: dict, context: LambdaContext) -> Any:
+        event = AppSyncResolverEvent(_event)
         resolver, config = self._resolver(event.type_name, event.field_name)
         kwargs = self._kwargs(event, context, config)
         return resolver(**kwargs)

aws_lambda_powertools/utilities/data_classes/cognito_user_pool_event.py

@@ -292,26 +292,26 @@ class CustomMessageTriggerEventResponse(DictWrapper):
     def sms_message(self) -> str:
         return self["response"]["smsMessage"]
 
-    @property
-    def email_message(self) -> str:
-        return self["response"]["emailMessage"]
-
-    @property
-    def email_subject(self) -> str:
-        return self["response"]["emailSubject"]
-
     @sms_message.setter
     def sms_message(self, value: str):
         """The custom SMS message to be sent to your users.
         Must include the codeParameter value received in the request."""
         self["response"]["smsMessage"] = value
 
+    @property
+    def email_message(self) -> str:
+        return self["response"]["emailMessage"]
+
     @email_message.setter
     def email_message(self, value: str):
         """The custom email message to be sent to your users.
         Must include the codeParameter value received in the request."""
         self["response"]["emailMessage"] = value
 
+    @property
+    def email_subject(self) -> str:
+        return self["response"]["emailSubject"]
+
     @email_subject.setter
     def email_subject(self, value: str):
         """The subject line for the custom message."""
@@ -471,26 +471,26 @@ class ClaimsOverrideDetails(DictWrapper):
     def claims_to_add_or_override(self) -> Optional[Dict[str, str]]:
         return self.get("claimsToAddOrOverride")
 
-    @property
-    def claims_to_suppress(self) -> Optional[List[str]]:
-        return self.get("claimsToSuppress")
-
-    @property
-    def group_configuration(self) -> Optional[GroupOverrideDetails]:
-        group_override_details = self.get("groupOverrideDetails")
-        return None if group_override_details is None else GroupOverrideDetails(group_override_details)
-
     @claims_to_add_or_override.setter
     def claims_to_add_or_override(self, value: Dict[str, str]):
         """A map of one or more key-value pairs of claims to add or override.
         For group related claims, use groupOverrideDetails instead."""
         self._data["claimsToAddOrOverride"] = value
 
+    @property
+    def claims_to_suppress(self) -> Optional[List[str]]:
+        return self.get("claimsToSuppress")
+
     @claims_to_suppress.setter
     def claims_to_suppress(self, value: List[str]):
         """A list that contains claims to be suppressed from the identity token."""
         self._data["claimsToSuppress"] = value
 
+    @property
+    def group_configuration(self) -> Optional[GroupOverrideDetails]:
+        group_override_details = self.get("groupOverrideDetails")
+        return None if group_override_details is None else GroupOverrideDetails(group_override_details)
+
     @group_configuration.setter
     def group_configuration(self, value: Dict[str, Any]):
         """The output object containing the current group configuration.
@@ -609,25 +609,25 @@ class DefineAuthChallengeTriggerEventResponse(DictWrapper):
     def challenge_name(self) -> str:
         return self["response"]["challengeName"]
 
-    @property
-    def fail_authentication(self) -> bool:
-        return bool(self["response"]["failAuthentication"])
-
-    @property
-    def issue_tokens(self) -> bool:
-        return bool(self["response"]["issueTokens"])
-
     @challenge_name.setter
     def challenge_name(self, value: str):
         """A string containing the name of the next challenge.
         If you want to present a new challenge to your user, specify the challenge name here."""
         self["response"]["challengeName"] = value
 
+    @property
+    def fail_authentication(self) -> bool:
+        return bool(self["response"]["failAuthentication"])
+
     @fail_authentication.setter
     def fail_authentication(self, value: bool):
         """Set to true if you want to terminate the current authentication process, or false otherwise."""
         self["response"]["failAuthentication"] = value
 
+    @property
+    def issue_tokens(self) -> bool:
+        return bool(self["response"]["issueTokens"])
+
     @issue_tokens.setter
     def issue_tokens(self, value: bool):
         """Set to true if you determine that the user has been sufficiently authenticated by
@@ -695,21 +695,17 @@ class CreateAuthChallengeTriggerEventResponse(DictWrapper):
     def public_challenge_parameters(self) -> Dict[str, str]:
         return self["response"]["publicChallengeParameters"]
 
-    @property
-    def private_challenge_parameters(self) -> Dict[str, str]:
-        return self["response"]["privateChallengeParameters"]
-
-    @property
-    def challenge_metadata(self) -> str:
-        return self["response"]["challengeMetadata"]
-
     @public_challenge_parameters.setter
     def public_challenge_parameters(self, value: Dict[str, str]):
         """One or more key-value pairs for the client app to use in the challenge to be presented to the user.
         This parameter should contain all of the necessary information to accurately present the challenge to
         the user."""
         self["response"]["publicChallengeParameters"] = value
 
+    @property
+    def private_challenge_parameters(self) -> Dict[str, str]:
+        return self["response"]["privateChallengeParameters"]
+
     @private_challenge_parameters.setter
     def private_challenge_parameters(self, value: Dict[str, str]):
         """This parameter is only used by the Verify Auth Challenge Response Lambda trigger.
@@ -719,6 +715,10 @@ def private_challenge_parameters(self, value: Dict[str, str]):
         for the question."""
         self["response"]["privateChallengeParameters"] = value
 
+    @property
+    def challenge_metadata(self) -> str:
+        return self["response"]["challengeMetadata"]
+
     @challenge_metadata.setter
     def challenge_metadata(self, value: str):
         """Your name for the custom challenge, if this is a custom challenge."""