diff --git a/aws_lambda_powertools/logging/lambda_context.py b/aws_lambda_powertools/logging/lambda_context.py index 75da8711f03..65e9e652a92 100644 --- a/aws_lambda_powertools/logging/lambda_context.py +++ b/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 diff --git a/aws_lambda_powertools/logging/logger.py b/aws_lambda_powertools/logging/logger.py index 98ecfc4c449..2e9cbb78d2e 100644 --- a/aws_lambda_powertools/logging/logger.py +++ b/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 diff --git a/aws_lambda_powertools/metrics/base.py b/aws_lambda_powertools/metrics/base.py index b7f3862b590..dc4fe34ee12 100644 --- a/aws_lambda_powertools/metrics/base.py +++ b/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 diff --git a/aws_lambda_powertools/metrics/metric.py b/aws_lambda_powertools/metrics/metric.py index 353defab0c6..8bdd0d800b8 100644 --- a/aws_lambda_powertools/metrics/metric.py +++ b/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=(",", ":"))) diff --git a/aws_lambda_powertools/metrics/metrics.py b/aws_lambda_powertools/metrics/metrics.py index 5178116a717..59d3b18e0e4 100644 --- a/aws_lambda_powertools/metrics/metrics.py +++ b/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__( diff --git a/aws_lambda_powertools/shared/functions.py b/aws_lambda_powertools/shared/functions.py index 079b5d29c49..b8f5cb9f74b 100644 --- a/aws_lambda_powertools/shared/functions.py +++ b/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. diff --git a/aws_lambda_powertools/utilities/batch/base.py b/aws_lambda_powertools/utilities/batch/base.py index 56d99c6d19c..a0ad18a9ec1 100644 --- a/aws_lambda_powertools/utilities/batch/base.py +++ b/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 diff --git a/aws_lambda_powertools/utilities/batch/sqs.py b/aws_lambda_powertools/utilities/batch/sqs.py index 597faa4c72e..e37fdbd3fb5 100644 --- a/aws_lambda_powertools/utilities/batch/sqs.py +++ b/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}" diff --git a/aws_lambda_powertools/utilities/data_classes/appsync/resolver_utils.py b/aws_lambda_powertools/utilities/data_classes/appsync/resolver_utils.py index b4b49888e24..9329b8effe8 100644 --- a/aws_lambda_powertools/utilities/data_classes/appsync/resolver_utils.py +++ b/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) diff --git a/aws_lambda_powertools/utilities/data_classes/cognito_user_pool_event.py b/aws_lambda_powertools/utilities/data_classes/cognito_user_pool_event.py index 461b906b230..e467875305f 100644 --- a/aws_lambda_powertools/utilities/data_classes/cognito_user_pool_event.py +++ b/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,14 +695,6 @@ 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. @@ -710,6 +702,10 @@ def public_challenge_parameters(self, value: Dict[str, str]): 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.""" diff --git a/aws_lambda_powertools/utilities/data_classes/common.py b/aws_lambda_powertools/utilities/data_classes/common.py index 65dfb4a0173..6f393cccb60 100644 --- a/aws_lambda_powertools/utilities/data_classes/common.py +++ b/aws_lambda_powertools/utilities/data_classes/common.py @@ -25,7 +25,9 @@ def raw_event(self) -> Dict[str, Any]: return self._data -def get_header_value(headers: Dict[str, str], name: str, default_value: str, case_sensitive: bool) -> Optional[str]: +def get_header_value( + headers: Dict[str, str], name: str, default_value: Optional[str], case_sensitive: Optional[bool] +) -> Optional[str]: """Get header value by name""" if case_sensitive: return headers.get(name, default_value) diff --git a/aws_lambda_powertools/utilities/idempotency/persistence/base.py b/aws_lambda_powertools/utilities/idempotency/persistence/base.py index 37c9968b3e0..263414a9573 100644 --- a/aws_lambda_powertools/utilities/idempotency/persistence/base.py +++ b/aws_lambda_powertools/utilities/idempotency/persistence/base.py @@ -40,7 +40,7 @@ def __init__( idempotency_key, status: str = "", expiry_timestamp: int = None, - response_data: str = "", + response_data: Optional[str] = "", payload_hash: str = None, ) -> None: """ diff --git a/aws_lambda_powertools/utilities/parameters/appconfig.py b/aws_lambda_powertools/utilities/parameters/appconfig.py index 03aac781c92..4490e260364 100644 --- a/aws_lambda_powertools/utilities/parameters/appconfig.py +++ b/aws_lambda_powertools/utilities/parameters/appconfig.py @@ -4,7 +4,7 @@ import os -from typing import Dict, Optional, Union +from typing import Any, Dict, Optional, Union from uuid import uuid4 import boto3 @@ -58,7 +58,7 @@ class AppConfigProvider(BaseProvider): """ - client = None + client: Any = None def __init__(self, environment: str, application: Optional[str] = None, config: Optional[Config] = None): """ diff --git a/aws_lambda_powertools/utilities/parameters/base.py b/aws_lambda_powertools/utilities/parameters/base.py index da73a26c07d..b3b907bc18b 100644 --- a/aws_lambda_powertools/utilities/parameters/base.py +++ b/aws_lambda_powertools/utilities/parameters/base.py @@ -7,14 +7,14 @@ from abc import ABC, abstractmethod from collections import namedtuple from datetime import datetime, timedelta -from typing import Dict, Optional, Tuple, Union +from typing import Any, Dict, Optional, Tuple, Union from .exceptions import GetParameterError, TransformParameterError DEFAULT_MAX_AGE_SECS = 5 ExpirableValue = namedtuple("ExpirableValue", ["value", "ttl"]) # These providers will be dynamically initialized on first use of the helper functions -DEFAULT_PROVIDERS = {} +DEFAULT_PROVIDERS: Dict[str, Any] = {} TRANSFORM_METHOD_JSON = "json" TRANSFORM_METHOD_BINARY = "binary" SUPPORTED_TRANSFORM_METHODS = [TRANSFORM_METHOD_JSON, TRANSFORM_METHOD_BINARY] @@ -25,7 +25,7 @@ class BaseProvider(ABC): Abstract Base Class for Parameter providers """ - store = None + store: Any = None def __init__(self): """ diff --git a/aws_lambda_powertools/utilities/parameters/dynamodb.py b/aws_lambda_powertools/utilities/parameters/dynamodb.py index dcb447b6060..a4b05121aa2 100644 --- a/aws_lambda_powertools/utilities/parameters/dynamodb.py +++ b/aws_lambda_powertools/utilities/parameters/dynamodb.py @@ -3,7 +3,7 @@ """ -from typing import Dict, Optional +from typing import Any, Dict, Optional import boto3 from boto3.dynamodb.conditions import Key @@ -139,7 +139,7 @@ class DynamoDBProvider(BaseProvider): c Parameter value c """ - table = None + table: Any = None key_attr = None sort_attr = None value_attr = None diff --git a/aws_lambda_powertools/utilities/parameters/secrets.py b/aws_lambda_powertools/utilities/parameters/secrets.py index f14e4703ba8..6b7ea21fdf6 100644 --- a/aws_lambda_powertools/utilities/parameters/secrets.py +++ b/aws_lambda_powertools/utilities/parameters/secrets.py @@ -3,7 +3,7 @@ """ -from typing import Dict, Optional, Union +from typing import Any, Dict, Optional, Union import boto3 from botocore.config import Config @@ -56,7 +56,7 @@ class SecretsProvider(BaseProvider): My parameter value """ - client = None + client: Any = None def __init__(self, config: Optional[Config] = None): """ diff --git a/aws_lambda_powertools/utilities/parameters/ssm.py b/aws_lambda_powertools/utilities/parameters/ssm.py index 97910eda8ed..4bbef8bfc15 100644 --- a/aws_lambda_powertools/utilities/parameters/ssm.py +++ b/aws_lambda_powertools/utilities/parameters/ssm.py @@ -3,7 +3,7 @@ """ -from typing import Dict, Optional, Union +from typing import Any, Dict, Optional, Union import boto3 from botocore.config import Config @@ -72,7 +72,7 @@ class SSMProvider(BaseProvider): /my/path/prefix/c Parameter value c """ - client = None + client: Any = None def __init__(self, config: Optional[Config] = None): """ diff --git a/aws_lambda_powertools/utilities/parser/types.py b/aws_lambda_powertools/utilities/parser/types.py index b19f9c9a87f..2565e52c764 100644 --- a/aws_lambda_powertools/utilities/parser/types.py +++ b/aws_lambda_powertools/utilities/parser/types.py @@ -1,13 +1,14 @@ """Generics and other shared types used across parser""" +import sys from typing import TypeVar from pydantic import BaseModel # We only need typing_extensions for python versions <3.8 -try: +if sys.version_info >= (3, 8): from typing import Literal # noqa: F401 -except ImportError: +else: from typing_extensions import Literal # noqa: F401 Model = TypeVar("Model", bound=BaseModel) diff --git a/aws_lambda_powertools/utilities/validation/base.py b/aws_lambda_powertools/utilities/validation/base.py index a5c82503735..eb84f300ded 100644 --- a/aws_lambda_powertools/utilities/validation/base.py +++ b/aws_lambda_powertools/utilities/validation/base.py @@ -40,7 +40,7 @@ def validate_data_against_schema(data: Dict, schema: Dict, formats: Optional[Dic raise SchemaValidationError(message) -def unwrap_event_from_envelope(data: Dict, envelope: str, jmespath_options: Dict) -> Any: +def unwrap_event_from_envelope(data: Dict, envelope: str, jmespath_options: Optional[Dict]) -> Any: """Searches data using JMESPath expression Parameters diff --git a/aws_lambda_powertools/utilities/validation/validator.py b/aws_lambda_powertools/utilities/validation/validator.py index 23a7241fd32..c962f8fff76 100644 --- a/aws_lambda_powertools/utilities/validation/validator.py +++ b/aws_lambda_powertools/utilities/validation/validator.py @@ -133,7 +133,7 @@ def handler(event, context): def validate( event: Dict, - schema: Dict = None, + schema: Dict, formats: Optional[Dict] = None, envelope: str = None, jmespath_options: Dict = None, diff --git a/docs/utilities/data_classes.md b/docs/utilities/data_classes.md index c7c11b6b2f9..fa85c719243 100644 --- a/docs/utilities/data_classes.md +++ b/docs/utilities/data_classes.md @@ -554,7 +554,7 @@ This example is based on the AWS Blog post [Introducing Amazon S3 Object Lambda === "app.py" - ```python hl_lines="4-5 10 12" + ```python hl_lines="5-6 12 14" import boto3 import requests diff --git a/tests/functional/test_metrics.py b/tests/functional/test_metrics.py index 6386e76e42f..3090a1228d2 100644 --- a/tests/functional/test_metrics.py +++ b/tests/functional/test_metrics.py @@ -739,3 +739,13 @@ def test_serialize_metric_set_metric_definition_multiple_values( assert "Timestamp" in metric_definition_output["_aws"] remove_timestamp(metrics=[metric_definition_output, expected_metric_definition]) assert metric_definition_output == expected_metric_definition + + +def test_metric_manage_metadata_set(): + expected_dict = {"setting": "On"} + + try: + metric = MetricManager(metadata_set=expected_dict) + assert metric.metadata_set == expected_dict + except AttributeError: + pytest.fail("AttributeError should not be raised")