From 0af2a3c3a9d87e60b803bfe398bb03b0765f19b2 Mon Sep 17 00:00:00 2001 From: Eric Nielsen <4120606+ericbn@users.noreply.github.com> Date: Wed, 14 Aug 2024 20:41:05 -0500 Subject: [PATCH 1/3] refactor(parameters): add from __future__ import annotations and update code according to ruff rules TCH, UP006, UP007, UP037 and FA100. --- .../utilities/parameters/appconfig.py | 40 ++-- .../utilities/parameters/base.py | 60 +++-- .../utilities/parameters/dynamodb.py | 18 +- .../utilities/parameters/secrets.py | 50 ++--- .../utilities/parameters/ssm.py | 207 +++++++++--------- 5 files changed, 186 insertions(+), 189 deletions(-) diff --git a/aws_lambda_powertools/utilities/parameters/appconfig.py b/aws_lambda_powertools/utilities/parameters/appconfig.py index be98fcecaf3..c232086aa17 100644 --- a/aws_lambda_powertools/utilities/parameters/appconfig.py +++ b/aws_lambda_powertools/utilities/parameters/appconfig.py @@ -2,27 +2,29 @@ AWS App Config configuration retrieval and caching utility """ +from __future__ import annotations + import os import warnings -from typing import TYPE_CHECKING, Dict, Optional, Union +from typing import TYPE_CHECKING import boto3 -from botocore.config import Config - -from aws_lambda_powertools.utilities.parameters.types import TransformOptions -from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning - -if TYPE_CHECKING: - from mypy_boto3_appconfigdata.client import AppConfigDataClient from aws_lambda_powertools.shared import constants from aws_lambda_powertools.shared.functions import ( resolve_env_var_choice, resolve_max_age, ) +from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning from .base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider +if TYPE_CHECKING: + from botocore.config import Config + from mypy_boto3_appconfigdata.client import AppConfigDataClient + + from aws_lambda_powertools.utilities.parameters.types import TransformOptions + class AppConfigProvider(BaseProvider): """ @@ -72,11 +74,11 @@ class AppConfigProvider(BaseProvider): def __init__( self, environment: str, - application: Optional[str] = None, - config: Optional[Config] = None, - boto_config: Optional[Config] = None, - boto3_session: Optional[boto3.session.Session] = None, - boto3_client: Optional["AppConfigDataClient"] = None, + application: str | None = None, + config: Config | None = None, + boto_config: Config | None = None, + boto3_session: boto3.session.Session | None = None, + boto3_client: AppConfigDataClient | None = None, ): """ Initialize the App Config client @@ -105,9 +107,9 @@ def __init__( self.environment = environment self.current_version = "" - self._next_token: Dict[str, str] = {} # nosec - token for get_latest_configuration executions + self._next_token: dict[str, str] = {} # nosec - token for get_latest_configuration executions # Dict to store the recently retrieved value for a specific configuration. - self.last_returned_value: Dict[str, bytes] = {} + self.last_returned_value: dict[str, bytes] = {} super().__init__(client=self.client) @@ -145,7 +147,7 @@ def _get(self, name: str, **sdk_options) -> bytes: return self.last_returned_value[name] - def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]: + def _get_multiple(self, path: str, **sdk_options) -> dict[str, str]: """ Retrieving multiple parameter values is not supported with AWS App Config Provider """ @@ -155,12 +157,12 @@ def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]: def get_app_config( name: str, environment: str, - application: Optional[str] = None, + application: str | None = None, transform: TransformOptions = None, force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, **sdk_options, -) -> Union[str, list, dict, bytes]: +) -> str | bytes | list | dict: """ Retrieve a configuration value from AWS App Config. diff --git a/aws_lambda_powertools/utilities/parameters/base.py b/aws_lambda_powertools/utilities/parameters/base.py index f58e8db77a7..a476ada6f43 100644 --- a/aws_lambda_powertools/utilities/parameters/base.py +++ b/aws_lambda_powertools/utilities/parameters/base.py @@ -9,28 +9,20 @@ import os from abc import ABC, abstractmethod from datetime import datetime, timedelta -from typing import ( - Any, - Callable, - Dict, - NamedTuple, - Optional, - Tuple, - Union, - cast, - overload, -) +from typing import TYPE_CHECKING, Any, Callable, NamedTuple, cast, overload from aws_lambda_powertools.shared import constants, user_agent from aws_lambda_powertools.shared.functions import resolve_max_age -from aws_lambda_powertools.utilities.parameters.types import TransformOptions from .exceptions import GetParameterError, TransformParameterError +if TYPE_CHECKING: + from aws_lambda_powertools.utilities.parameters.types import TransformOptions + DEFAULT_MAX_AGE_SECS = "300" # These providers will be dynamically initialized on first use of the helper functions -DEFAULT_PROVIDERS: Dict[str, Any] = {} +DEFAULT_PROVIDERS: dict[str, Any] = {} TRANSFORM_METHOD_JSON = "json" TRANSFORM_METHOD_BINARY = "binary" SUPPORTED_TRANSFORM_METHODS = [TRANSFORM_METHOD_JSON, TRANSFORM_METHOD_BINARY] @@ -45,7 +37,7 @@ class ExpirableValue(NamedTuple): - value: str | bytes | Dict[str, Any] + value: str | bytes | dict[str, Any] ttl: datetime @@ -54,7 +46,7 @@ class BaseProvider(ABC): Abstract Base Class for Parameter providers """ - store: Dict[Tuple, ExpirableValue] + store: dict[tuple, ExpirableValue] def __init__(self, *, client=None, resource=None): """ @@ -65,19 +57,19 @@ def __init__(self, *, client=None, resource=None): if resource is not None: user_agent.register_feature_to_resource(resource=resource, feature="parameters") - self.store: Dict[Tuple, ExpirableValue] = {} + self.store: dict[tuple, ExpirableValue] = {} - def has_not_expired_in_cache(self, key: Tuple) -> bool: + def has_not_expired_in_cache(self, key: tuple) -> bool: return key in self.store and self.store[key].ttl >= datetime.now() def get( self, name: str, - max_age: Optional[int] = None, + max_age: int | None = None, transform: TransformOptions = None, force_fetch: bool = False, **sdk_options, - ) -> Optional[Union[str, dict, bytes]]: + ) -> str | bytes | dict | None: """ Retrieve a parameter value or return the cached value @@ -114,7 +106,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 + value: str | bytes | dict | None = None key = self._build_cache_key(name=name, transform=transform) # If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS @@ -139,7 +131,7 @@ def get( return value @abstractmethod - def _get(self, name: str, **sdk_options) -> Union[str, bytes, Dict[str, Any]]: + def _get(self, name: str, **sdk_options) -> str | bytes | dict[str, Any]: """ Retrieve parameter value from the underlying parameter store """ @@ -154,12 +146,12 @@ def set(self, name: str, value: Any, *, overwrite: bool = False, **kwargs): def get_multiple( self, path: str, - max_age: Optional[int] = None, + max_age: int | None = None, transform: TransformOptions = None, raise_on_transform_error: bool = False, force_fetch: bool = False, **sdk_options, - ) -> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]: + ) -> dict[str, str] | dict[str, bytes] | dict[str, dict]: """ Retrieve multiple parameters based on a path prefix @@ -211,7 +203,7 @@ def get_multiple( return values @abstractmethod - def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]: + def _get_multiple(self, path: str, **sdk_options) -> dict[str, str]: """ Retrieve multiple parameter values from the underlying parameter store """ @@ -220,10 +212,10 @@ def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]: def clear_cache(self): self.store.clear() - def fetch_from_cache(self, key: Tuple): + def fetch_from_cache(self, key: tuple): return self.store[key].value if key in self.store else {} - def add_to_cache(self, key: Tuple, value: Any, max_age: int): + def add_to_cache(self, key: tuple, value: Any, max_age: int): if max_age <= 0: return @@ -248,7 +240,7 @@ def _build_cache_key( Returns ------- - Tuple[str, TransformOptions, bool] + tuple[str, TransformOptions, bool] Cache key """ return (name, transform, is_nested) @@ -294,28 +286,28 @@ def get_transform_method(value: str, transform: TransformOptions = None) -> Call @overload def transform_value( - value: Dict[str, Any], + value: dict[str, Any], transform: TransformOptions, raise_on_transform_error: bool = False, key: str = "", -) -> Dict[str, Any]: ... +) -> dict[str, Any]: ... @overload def transform_value( - value: Union[str, bytes, Dict[str, Any]], + value: str | bytes | dict[str, Any], transform: TransformOptions, raise_on_transform_error: bool = False, key: str = "", -) -> Optional[Union[str, bytes, Dict[str, Any]]]: ... +) -> str | bytes | dict[str, Any] | None: ... def transform_value( - value: Union[str, bytes, Dict[str, Any]], + value: str | bytes | dict[str, Any], transform: TransformOptions, raise_on_transform_error: bool = True, key: str = "", -) -> Optional[Union[str, bytes, Dict[str, Any]]]: +) -> str | bytes | dict[str, Any] | None: """ Transform a value using one of the available options. @@ -348,7 +340,7 @@ def transform_value( # where one of the keys might fail during transform, e.g. `{"a": "valid", "b": "{"}` # expected: `{"a": "valid", "b": None}` - transformed_values: Dict[str, Any] = {} + transformed_values: dict[str, Any] = {} for dict_key, dict_value in value.items(): transform_method = get_transform_method(value=dict_key, transform=transform) try: diff --git a/aws_lambda_powertools/utilities/parameters/dynamodb.py b/aws_lambda_powertools/utilities/parameters/dynamodb.py index 934c1d927b3..c7175cdde51 100644 --- a/aws_lambda_powertools/utilities/parameters/dynamodb.py +++ b/aws_lambda_powertools/utilities/parameters/dynamodb.py @@ -2,18 +2,20 @@ Amazon DynamoDB parameter retrieval and caching utility """ +from __future__ import annotations + import warnings -from typing import TYPE_CHECKING, Dict, Optional +from typing import TYPE_CHECKING import boto3 from boto3.dynamodb.conditions import Key -from botocore.config import Config from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning from .base import BaseProvider if TYPE_CHECKING: + from botocore.config import Config from mypy_boto3_dynamodb.service_resource import DynamoDBServiceResource @@ -156,11 +158,11 @@ def __init__( key_attr: str = "id", sort_attr: str = "sk", value_attr: str = "value", - endpoint_url: Optional[str] = None, - config: Optional[Config] = None, - boto_config: Optional[Config] = None, - boto3_session: Optional[boto3.session.Session] = None, - boto3_client: Optional["DynamoDBServiceResource"] = None, + endpoint_url: str | None = None, + config: Config | None = None, + boto_config: Config | None = None, + boto3_session: boto3.session.Session | None = None, + boto3_client: DynamoDBServiceResource | None = None, ): """ Initialize the DynamoDB client @@ -203,7 +205,7 @@ def _get(self, name: str, **sdk_options) -> str: # without a breaking change within ABC return type return self.table.get_item(**sdk_options)["Item"][self.value_attr] # type: ignore[return-value] - def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]: + def _get_multiple(self, path: str, **sdk_options) -> dict[str, str]: """ Retrieve multiple parameter values from Amazon DynamoDB diff --git a/aws_lambda_powertools/utilities/parameters/secrets.py b/aws_lambda_powertools/utilities/parameters/secrets.py index d1ef331dfa5..f8224624a1d 100644 --- a/aws_lambda_powertools/utilities/parameters/secrets.py +++ b/aws_lambda_powertools/utilities/parameters/secrets.py @@ -8,23 +8,23 @@ import logging import os import warnings -from typing import TYPE_CHECKING, Dict, Literal, Optional, Union, overload +from typing import TYPE_CHECKING, Literal, overload import boto3 -from botocore.config import Config +from aws_lambda_powertools.shared import constants +from aws_lambda_powertools.shared.functions import resolve_max_age +from aws_lambda_powertools.shared.json_encoder import Encoder +from aws_lambda_powertools.utilities.parameters.base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider +from aws_lambda_powertools.utilities.parameters.exceptions import SetSecretError from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning if TYPE_CHECKING: + from botocore.config import Config from mypy_boto3_secretsmanager.client import SecretsManagerClient from mypy_boto3_secretsmanager.type_defs import CreateSecretResponseTypeDef -from aws_lambda_powertools.shared import constants -from aws_lambda_powertools.shared.functions import resolve_max_age -from aws_lambda_powertools.shared.json_encoder import Encoder -from aws_lambda_powertools.utilities.parameters.base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider -from aws_lambda_powertools.utilities.parameters.exceptions import SetSecretError -from aws_lambda_powertools.utilities.parameters.types import TransformOptions + from aws_lambda_powertools.utilities.parameters.types import TransformOptions logger = logging.getLogger(__name__) @@ -80,10 +80,10 @@ class SecretsProvider(BaseProvider): def __init__( self, - config: Optional[Config] = None, - boto_config: Optional[Config] = None, - boto3_session: Optional[boto3.session.Session] = None, - boto3_client: Optional[SecretsManagerClient] = None, + config: Config | None = None, + boto_config: Config | None = None, + boto3_session: boto3.session.Session | None = None, + boto3_client: SecretsManagerClient | None = None, ): """ Initialize the Secrets Manager client @@ -103,7 +103,7 @@ def __init__( super().__init__(client=self.client) - def _get(self, name: str, **sdk_options) -> Union[str, bytes]: + def _get(self, name: str, **sdk_options) -> str | bytes: """ Retrieve a parameter value from AWS Systems Manager Parameter Store @@ -125,7 +125,7 @@ def _get(self, name: str, **sdk_options) -> Union[str, bytes]: return secret_value["SecretBinary"] - def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]: + def _get_multiple(self, path: str, **sdk_options) -> dict[str, str]: """ Retrieving multiple parameter values is not supported with AWS Secrets Manager """ @@ -168,9 +168,9 @@ def _update_secret(self, name: str, **sdk_options): def set( self, name: str, - value: Union[str, dict, bytes], + value: str | bytes | dict, *, # force keyword arguments - client_request_token: Optional[str] = None, + client_request_token: str | None = None, **sdk_options, ) -> CreateSecretResponseTypeDef: """ @@ -265,7 +265,7 @@ def get_secret( name: str, transform: None = None, force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, **sdk_options, ) -> str: ... @@ -275,7 +275,7 @@ def get_secret( name: str, transform: Literal["json"], force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, **sdk_options, ) -> dict: ... @@ -285,9 +285,9 @@ def get_secret( name: str, transform: Literal["binary"], force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, **sdk_options, -) -> Union[str, dict, bytes]: ... +) -> str | bytes | dict: ... @overload @@ -295,7 +295,7 @@ def get_secret( name: str, transform: Literal["auto"], force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, **sdk_options, ) -> bytes: ... @@ -304,9 +304,9 @@ def get_secret( name: str, transform: TransformOptions = None, force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, **sdk_options, -) -> Union[str, dict, bytes]: +) -> str | bytes | dict: """ Retrieve a parameter value from AWS Secrets Manager @@ -370,9 +370,9 @@ def get_secret( def set_secret( name: str, - value: Union[str, bytes], + value: str | bytes, *, # force keyword arguments - client_request_token: Optional[str] = None, + client_request_token: str | None = None, **sdk_options, ) -> CreateSecretResponseTypeDef: """ diff --git a/aws_lambda_powertools/utilities/parameters/ssm.py b/aws_lambda_powertools/utilities/parameters/ssm.py index 6d29881cdf6..bbae4cad645 100644 --- a/aws_lambda_powertools/utilities/parameters/ssm.py +++ b/aws_lambda_powertools/utilities/parameters/ssm.py @@ -7,10 +7,9 @@ import logging import os import warnings -from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Tuple, Union, overload +from typing import TYPE_CHECKING, Any, Literal, overload import boto3 -from botocore.config import Config from aws_lambda_powertools.shared import constants from aws_lambda_powertools.shared.functions import ( @@ -25,13 +24,15 @@ transform_value, ) from aws_lambda_powertools.utilities.parameters.exceptions import GetParameterError, SetParameterError -from aws_lambda_powertools.utilities.parameters.types import TransformOptions from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning if TYPE_CHECKING: + from botocore.config import Config from mypy_boto3_ssm.client import SSMClient from mypy_boto3_ssm.type_defs import GetParametersResultTypeDef, PutParameterResultTypeDef + from aws_lambda_powertools.utilities.parameters.types import TransformOptions + SSM_PARAMETER_TYPES = Literal["String", "StringList", "SecureString"] SSM_PARAMETER_TIER = Literal["Standard", "Advanced", "Intelligent-Tiering"] @@ -108,10 +109,10 @@ class SSMProvider(BaseProvider): def __init__( self, - config: Optional[Config] = None, - boto_config: Optional[Config] = None, - boto3_session: Optional[boto3.session.Session] = None, - boto3_client: Optional[SSMClient] = None, + config: Config | None = None, + boto_config: Config | None = None, + boto3_session: boto3.session.Session | None = None, + boto3_client: SSMClient | None = None, ): """ Initialize the SSM Parameter Store client @@ -134,14 +135,14 @@ def __init__( def get_multiple( # type: ignore[override] self, path: str, - max_age: Optional[int] = None, + max_age: int | None = None, transform: TransformOptions = None, raise_on_transform_error: bool = False, - decrypt: Optional[bool] = None, + decrypt: bool | None = None, force_fetch: bool = False, recursive: bool = False, **sdk_options, - ) -> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]: + ) -> dict[str, str] | dict[str, bytes] | dict[str, dict]: """ Retrieve multiple parameters based on a path prefix @@ -193,12 +194,12 @@ def get_multiple( # type: ignore[override] def get( # type: ignore[override] self, name: str, - max_age: Optional[int] = None, + max_age: int | None = None, transform: TransformOptions = None, - decrypt: Optional[bool] = None, + decrypt: bool | None = None, force_fetch: bool = False, **sdk_options, - ) -> Optional[Union[str, dict, bytes]]: + ) -> str | bytes | dict | None: """ Retrieve a parameter value or return the cached value @@ -385,10 +386,10 @@ def _get(self, name: str, decrypt: bool = False, **sdk_options) -> str: def _get_multiple( self, path: str, - decrypt: Optional[bool] = None, + decrypt: bool | None = None, recursive: bool = False, **sdk_options, - ) -> Dict[str, str]: + ) -> dict[str, str]: """ Retrieve multiple parameter values from AWS Systems Manager Parameter Store @@ -428,12 +429,12 @@ def _get_multiple( # NOTE: When bandwidth permits, allocate a week to refactor to lower cognitive load def get_parameters_by_name( self, - parameters: Dict[str, Dict], + parameters: dict[str, dict], transform: TransformOptions = None, - decrypt: Optional[bool] = None, - max_age: Optional[int] = None, + decrypt: bool | None = None, + max_age: int | None = None, raise_on_error: bool = True, - ) -> Dict[str, str] | Dict[str, bytes] | Dict[str, dict]: + ) -> dict[str, str] | dict[str, bytes] | dict[str, dict]: """ Retrieve multiple parameter values by name from SSM or cache. @@ -459,7 +460,7 @@ def get_parameters_by_name( Parameters ---------- - parameters: List[Dict[str, Dict]] + parameters: dict[str, dict] List of parameter names, and any optional overrides transform: str, optional Transforms the content from a JSON object ('json') or base64 binary string ('binary') @@ -488,11 +489,11 @@ def get_parameters_by_name( ) # Init potential batch/decrypt batch responses and errors - batch_ret: Dict[str, Any] = {} - decrypt_ret: Dict[str, Any] = {} - batch_err: List[str] = [] - decrypt_err: List[str] = [] - response: Dict[str, Any] = {} + batch_ret: dict[str, Any] = {} + decrypt_ret: dict[str, Any] = {} + batch_err: list[str] = [] + decrypt_err: list[str] = [] + response: dict[str, Any] = {} # NOTE: We fail early to avoid unintended graceful errors being replaced with their '_errors' param values self._raise_if_errors_key_is_present(parameters, self._ERRORS_KEY, raise_on_error) @@ -519,11 +520,11 @@ def get_parameters_by_name( def _get_parameters_by_name_with_decrypt_option( self, - batch: Dict[str, Dict], + batch: dict[str, dict], raise_on_error: bool, - ) -> Tuple[Dict, List]: - response: Dict[str, Any] = {} - errors: List[str] = [] + ) -> tuple[dict, list]: + response: dict[str, Any] = {} + errors: list[str] = [] # Decided for single-thread as it outperforms in 128M and 1G + reduce timeout risk # see: https://github.com/aws-powertools/powertools-lambda-python/issues/1040#issuecomment-1299954613 @@ -540,12 +541,12 @@ def _get_parameters_by_name_with_decrypt_option( def _get_parameters_batch_by_name( self, - batch: Dict[str, Dict], + batch: dict[str, dict], raise_on_error: bool = True, decrypt: bool = False, - ) -> Tuple[Dict, List]: + ) -> tuple[dict, list]: """Slice batch and fetch parameters using GetParameters by max permitted""" - errors: List[str] = [] + errors: list[str] = [] # Fetch each possible batch param from cache and return if entire batch is cached cached_params = self._get_parameters_by_name_from_cache(batch) @@ -557,7 +558,7 @@ def _get_parameters_batch_by_name( return {**cached_params, **batch_ret}, errors - def _get_parameters_by_name_from_cache(self, batch: Dict[str, Dict]) -> Dict[str, Any]: + def _get_parameters_by_name_from_cache(self, batch: dict[str, dict]) -> dict[str, Any]: """Fetch each parameter from batch that hasn't been expired""" cache = {} for name, options in batch.items(): @@ -569,14 +570,14 @@ def _get_parameters_by_name_from_cache(self, batch: Dict[str, Dict]) -> Dict[str def _get_parameters_by_name_in_chunks( self, - batch: Dict[str, Dict], - cache: Dict[str, Any], + batch: dict[str, dict], + cache: dict[str, Any], raise_on_error: bool, decrypt: bool = False, - ) -> Tuple[Dict, List]: + ) -> tuple[dict, list]: """Take out differences from cache and batch, slice it and fetch from SSM""" - response: Dict[str, Any] = {} - errors: List[str] = [] + response: dict[str, Any] = {} + errors: list[str] = [] diff = {key: value for key, value in batch.items() if key not in cache} @@ -593,22 +594,22 @@ def _get_parameters_by_name_in_chunks( def _get_parameters_by_name( self, - parameters: Dict[str, Dict], + parameters: dict[str, dict], raise_on_error: bool = True, decrypt: bool = False, - ) -> Tuple[Dict[str, Any], List[str]]: + ) -> tuple[dict[str, Any], list[str]]: """Use SSM GetParameters to fetch parameters, hydrate cache, and handle partial failure Parameters ---------- - parameters : Dict[str, Dict] + parameters : dict[str, dict] Parameters to fetch raise_on_error : bool, optional Whether to fail-fast or fail gracefully by including "_errors" key in the response, by default True Returns ------- - Dict[str, Any] + dict[str, Any] Retrieved parameters as key names and their values Raises @@ -616,8 +617,8 @@ def _get_parameters_by_name( GetParameterError When one or more parameters failed on fetching, and raise_on_error is enabled """ - ret: Dict[str, Any] = {} - batch_errors: List[str] = [] + ret: dict[str, Any] = {} + batch_errors: list[str] = [] parameter_names = list(parameters.keys()) # All params in the batch must be decrypted @@ -639,10 +640,10 @@ def _get_parameters_by_name( def _transform_and_cache_get_parameters_response( self, api_response: GetParametersResultTypeDef, - parameters: Dict[str, Any], + parameters: dict[str, Any], raise_on_error: bool = True, - ) -> Dict[str, Any]: - response: Dict[str, Any] = {} + ) -> dict[str, Any]: + response: dict[str, Any] = {} for parameter in api_response["Parameters"]: name = parameter["Name"] @@ -665,7 +666,7 @@ def _transform_and_cache_get_parameters_response( def _handle_any_invalid_get_parameter_errors( api_response: GetParametersResultTypeDef, raise_on_error: bool = True, - ) -> List[str]: + ) -> list[str]: """GetParameters is non-atomic. Failures don't always reflect in exceptions so we need to collect.""" failed_parameters = api_response["InvalidParameters"] if failed_parameters: @@ -678,16 +679,16 @@ def _handle_any_invalid_get_parameter_errors( @staticmethod def _split_batch_and_decrypt_parameters( - parameters: Dict[str, Dict], + parameters: dict[str, dict], transform: TransformOptions, max_age: int, decrypt: bool, - ) -> Tuple[Dict[str, Dict], Dict[str, Dict]]: + ) -> tuple[dict[str, dict], dict[str, dict]]: """Split parameters that can be fetched by GetParameters vs GetParameter Parameters ---------- - parameters : Dict[str, Dict] + parameters : dict[str, dict] Parameters containing names as key and optional config override as value transform : TransformOptions Transform configuration @@ -698,11 +699,11 @@ def _split_batch_and_decrypt_parameters( Returns ------- - Tuple[Dict[str, Dict], Dict[str, Dict]] + tuple[dict[str, dict], dict[str, dict]] GetParameters and GetParameter parameters dict along with their overrides/globals merged """ - batch_parameters: Dict[str, Dict] = {} - decrypt_parameters: Dict[str, Any] = {} + batch_parameters: dict[str, dict] = {} + decrypt_parameters: dict[str, Any] = {} for parameter, options in parameters.items(): # NOTE: TypeDict later @@ -725,7 +726,7 @@ def _split_batch_and_decrypt_parameters( return batch_parameters, decrypt_parameters @staticmethod - def _raise_if_errors_key_is_present(parameters: Dict, reserved_parameter: str, raise_on_error: bool): + def _raise_if_errors_key_is_present(parameters: dict, reserved_parameter: str, raise_on_error: bool): """Raise GetParameterError if fail-fast is disabled and '_errors' key is in parameters batch""" if not raise_on_error and reserved_parameter in parameters: raise GetParameterError( @@ -737,9 +738,9 @@ def _raise_if_errors_key_is_present(parameters: Dict, reserved_parameter: str, r def get_parameter( name: str, transform: None = None, - decrypt: Optional[bool] = None, + decrypt: bool | None = None, force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, **sdk_options, ) -> str: ... @@ -748,9 +749,9 @@ def get_parameter( def get_parameter( name: str, transform: Literal["json"], - decrypt: Optional[bool] = None, + decrypt: bool | None = None, force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, **sdk_options, ) -> dict: ... @@ -759,20 +760,20 @@ def get_parameter( def get_parameter( name: str, transform: Literal["binary"], - decrypt: Optional[bool] = None, + decrypt: bool | None = None, force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, **sdk_options, -) -> Union[str, dict, bytes]: ... +) -> str | bytes | dict: ... @overload def get_parameter( name: str, transform: Literal["auto"], - decrypt: Optional[bool] = None, + decrypt: bool | None = None, force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, **sdk_options, ) -> bytes: ... @@ -780,11 +781,11 @@ def get_parameter( def get_parameter( name: str, transform: TransformOptions = None, - decrypt: Optional[bool] = None, + decrypt: bool | None = None, force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, **sdk_options, -) -> Union[str, dict, bytes]: +) -> str | bytes | dict: """ Retrieve a parameter value from AWS Systems Manager (SSM) Parameter Store @@ -860,12 +861,12 @@ def get_parameters( path: str, transform: None = None, recursive: bool = True, - decrypt: Optional[bool] = None, + decrypt: bool | None = None, force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, raise_on_transform_error: bool = False, **sdk_options, -) -> Dict[str, str]: ... +) -> dict[str, str]: ... @overload @@ -873,12 +874,12 @@ def get_parameters( path: str, transform: Literal["json"], recursive: bool = True, - decrypt: Optional[bool] = None, + decrypt: bool | None = None, force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, raise_on_transform_error: bool = False, **sdk_options, -) -> Dict[str, dict]: ... +) -> dict[str, dict]: ... @overload @@ -886,12 +887,12 @@ def get_parameters( path: str, transform: Literal["binary"], recursive: bool = True, - decrypt: Optional[bool] = None, + decrypt: bool | None = None, force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, raise_on_transform_error: bool = False, **sdk_options, -) -> Dict[str, bytes]: ... +) -> dict[str, bytes]: ... @overload @@ -899,24 +900,24 @@ def get_parameters( path: str, transform: Literal["auto"], recursive: bool = True, - decrypt: Optional[bool] = None, + decrypt: bool | None = None, force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, raise_on_transform_error: bool = False, **sdk_options, -) -> Union[Dict[str, bytes], Dict[str, dict], Dict[str, str]]: ... +) -> dict[str, str] | dict[str, bytes] | dict[str, dict]: ... def get_parameters( path: str, transform: TransformOptions = None, recursive: bool = True, - decrypt: Optional[bool] = None, + decrypt: bool | None = None, force_fetch: bool = False, - max_age: Optional[int] = None, + max_age: int | None = None, raise_on_transform_error: bool = False, **sdk_options, -) -> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]: +) -> dict[str, str] | dict[str, bytes] | dict[str, dict]: """ Retrieve multiple parameter values from AWS Systems Manager (SSM) Parameter Store @@ -1072,57 +1073,57 @@ def set_parameter( @overload def get_parameters_by_name( - parameters: Dict[str, Dict], + parameters: dict[str, dict], transform: None = None, - decrypt: Optional[bool] = None, - max_age: Optional[int] = None, + decrypt: bool | None = None, + max_age: int | None = None, raise_on_error: bool = True, -) -> Dict[str, str]: ... +) -> dict[str, str]: ... @overload def get_parameters_by_name( - parameters: Dict[str, Dict], + parameters: dict[str, dict], transform: Literal["binary"], - decrypt: Optional[bool] = None, - max_age: Optional[int] = None, + decrypt: bool | None = None, + max_age: int | None = None, raise_on_error: bool = True, -) -> Dict[str, bytes]: ... +) -> dict[str, bytes]: ... @overload def get_parameters_by_name( - parameters: Dict[str, Dict], + parameters: dict[str, dict], transform: Literal["json"], - decrypt: Optional[bool] = None, - max_age: Optional[int] = None, + decrypt: bool | None = None, + max_age: int | None = None, raise_on_error: bool = True, -) -> Dict[str, Dict[str, Any]]: ... +) -> dict[str, dict[str, Any]]: ... @overload def get_parameters_by_name( - parameters: Dict[str, Dict], + parameters: dict[str, dict], transform: Literal["auto"], - decrypt: Optional[bool] = None, - max_age: Optional[int] = None, + decrypt: bool | None = None, + max_age: int | None = None, raise_on_error: bool = True, -) -> Union[Dict[str, str], Dict[str, dict]]: ... +) -> dict[str, str] | dict[str, dict]: ... def get_parameters_by_name( - parameters: Dict[str, Any], + parameters: dict[str, Any], transform: TransformOptions = None, - decrypt: Optional[bool] = None, - max_age: Optional[int] = None, + decrypt: bool | None = None, + max_age: int | None = None, raise_on_error: bool = True, -) -> Union[Dict[str, str], Dict[str, bytes], Dict[str, dict]]: +) -> dict[str, str] | dict[str, bytes] | dict[str, dict]: """ Retrieve multiple parameter values by name from AWS Systems Manager (SSM) Parameter Store Parameters ---------- - parameters: List[Dict[str, Dict]] + parameters: dict[str, Any] List of parameter names, and any optional overrides transform: str, optional Transforms the content from a JSON object ('json') or base64 binary string ('binary') From 2d4415deafe310593355bc3251f49e5b5ae9d405 Mon Sep 17 00:00:00 2001 From: Leandro Damascena Date: Thu, 15 Aug 2024 12:38:50 +0100 Subject: [PATCH 2/3] Fixing constants + absolute imports --- .../utilities/parameters/appconfig.py | 4 ++-- .../utilities/parameters/base.py | 23 +++++-------------- .../utilities/parameters/constants.py | 22 ++++++++++++++++++ .../utilities/parameters/dynamodb.py | 3 +-- .../utilities/parameters/secrets.py | 3 ++- .../utilities/parameters/ssm.py | 11 +++++---- 6 files changed, 39 insertions(+), 27 deletions(-) create mode 100644 aws_lambda_powertools/utilities/parameters/constants.py diff --git a/aws_lambda_powertools/utilities/parameters/appconfig.py b/aws_lambda_powertools/utilities/parameters/appconfig.py index c232086aa17..dd4c779d1c1 100644 --- a/aws_lambda_powertools/utilities/parameters/appconfig.py +++ b/aws_lambda_powertools/utilities/parameters/appconfig.py @@ -15,10 +15,10 @@ resolve_env_var_choice, resolve_max_age, ) +from aws_lambda_powertools.utilities.parameters.base import BaseProvider +from aws_lambda_powertools.utilities.parameters.constants import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning -from .base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider - if TYPE_CHECKING: from botocore.config import Config from mypy_boto3_appconfigdata.client import AppConfigDataClient diff --git a/aws_lambda_powertools/utilities/parameters/base.py b/aws_lambda_powertools/utilities/parameters/base.py index a476ada6f43..5c6d4dcc955 100644 --- a/aws_lambda_powertools/utilities/parameters/base.py +++ b/aws_lambda_powertools/utilities/parameters/base.py @@ -4,8 +4,6 @@ from __future__ import annotations -import base64 -import json import os from abc import ABC, abstractmethod from datetime import datetime, timedelta @@ -19,21 +17,12 @@ if TYPE_CHECKING: from aws_lambda_powertools.utilities.parameters.types import TransformOptions -DEFAULT_MAX_AGE_SECS = "300" - -# These providers will be dynamically initialized on first use of the helper functions -DEFAULT_PROVIDERS: dict[str, Any] = {} -TRANSFORM_METHOD_JSON = "json" -TRANSFORM_METHOD_BINARY = "binary" -SUPPORTED_TRANSFORM_METHODS = [TRANSFORM_METHOD_JSON, TRANSFORM_METHOD_BINARY] - -TRANSFORM_METHOD_MAPPING = { - TRANSFORM_METHOD_JSON: json.loads, - TRANSFORM_METHOD_BINARY: base64.b64decode, - ".json": json.loads, - ".binary": base64.b64decode, - None: lambda x: x, -} + +from aws_lambda_powertools.utilities.parameters.constants import ( + DEFAULT_MAX_AGE_SECS, + DEFAULT_PROVIDERS, + TRANSFORM_METHOD_MAPPING, +) class ExpirableValue(NamedTuple): diff --git a/aws_lambda_powertools/utilities/parameters/constants.py b/aws_lambda_powertools/utilities/parameters/constants.py new file mode 100644 index 00000000000..55380b0bd55 --- /dev/null +++ b/aws_lambda_powertools/utilities/parameters/constants.py @@ -0,0 +1,22 @@ +from __future__ import annotations + +import base64 +import json +from typing import Any, Literal + +SSM_PARAMETER_TYPES = Literal["String", "StringList", "SecureString"] +SSM_PARAMETER_TIER = Literal["Standard", "Advanced", "Intelligent-Tiering"] + +DEFAULT_MAX_AGE_SECS = "300" + +# These providers will be dynamically initialized on first use of the helper functions +DEFAULT_PROVIDERS: dict[str, Any] = {} +TRANSFORM_METHOD_JSON = "json" +TRANSFORM_METHOD_BINARY = "binary" +TRANSFORM_METHOD_MAPPING = { + TRANSFORM_METHOD_JSON: json.loads, + TRANSFORM_METHOD_BINARY: base64.b64decode, + ".json": json.loads, + ".binary": base64.b64decode, + None: lambda x: x, +} diff --git a/aws_lambda_powertools/utilities/parameters/dynamodb.py b/aws_lambda_powertools/utilities/parameters/dynamodb.py index c7175cdde51..3203a785bae 100644 --- a/aws_lambda_powertools/utilities/parameters/dynamodb.py +++ b/aws_lambda_powertools/utilities/parameters/dynamodb.py @@ -10,10 +10,9 @@ import boto3 from boto3.dynamodb.conditions import Key +from aws_lambda_powertools.utilities.parameters.base import BaseProvider from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning -from .base import BaseProvider - if TYPE_CHECKING: from botocore.config import Config from mypy_boto3_dynamodb.service_resource import DynamoDBServiceResource diff --git a/aws_lambda_powertools/utilities/parameters/secrets.py b/aws_lambda_powertools/utilities/parameters/secrets.py index f8224624a1d..228318031a3 100644 --- a/aws_lambda_powertools/utilities/parameters/secrets.py +++ b/aws_lambda_powertools/utilities/parameters/secrets.py @@ -15,7 +15,8 @@ from aws_lambda_powertools.shared import constants from aws_lambda_powertools.shared.functions import resolve_max_age from aws_lambda_powertools.shared.json_encoder import Encoder -from aws_lambda_powertools.utilities.parameters.base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider +from aws_lambda_powertools.utilities.parameters.base import BaseProvider +from aws_lambda_powertools.utilities.parameters.constants import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS from aws_lambda_powertools.utilities.parameters.exceptions import SetSecretError from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning diff --git a/aws_lambda_powertools/utilities/parameters/ssm.py b/aws_lambda_powertools/utilities/parameters/ssm.py index bbae4cad645..4ec3081a6ca 100644 --- a/aws_lambda_powertools/utilities/parameters/ssm.py +++ b/aws_lambda_powertools/utilities/parameters/ssm.py @@ -18,11 +18,15 @@ slice_dictionary, ) from aws_lambda_powertools.utilities.parameters.base import ( - DEFAULT_MAX_AGE_SECS, - DEFAULT_PROVIDERS, BaseProvider, transform_value, ) +from aws_lambda_powertools.utilities.parameters.constants import ( + DEFAULT_MAX_AGE_SECS, + DEFAULT_PROVIDERS, + SSM_PARAMETER_TIER, + SSM_PARAMETER_TYPES, +) from aws_lambda_powertools.utilities.parameters.exceptions import GetParameterError, SetParameterError from aws_lambda_powertools.warnings import PowertoolsDeprecationWarning @@ -33,9 +37,6 @@ from aws_lambda_powertools.utilities.parameters.types import TransformOptions -SSM_PARAMETER_TYPES = Literal["String", "StringList", "SecureString"] -SSM_PARAMETER_TIER = Literal["Standard", "Advanced", "Intelligent-Tiering"] - logger = logging.getLogger(__name__) From 4aed3f9aec15ec3de1e45c8ed89ce3059cb49557 Mon Sep 17 00:00:00 2001 From: Leandro Damascena Date: Thu, 15 Aug 2024 12:52:00 +0100 Subject: [PATCH 3/3] Fixing constants + absolute imports --- aws_lambda_powertools/utilities/parameters/base.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/aws_lambda_powertools/utilities/parameters/base.py b/aws_lambda_powertools/utilities/parameters/base.py index 5c6d4dcc955..897cd4ace57 100644 --- a/aws_lambda_powertools/utilities/parameters/base.py +++ b/aws_lambda_powertools/utilities/parameters/base.py @@ -11,8 +11,7 @@ from aws_lambda_powertools.shared import constants, user_agent from aws_lambda_powertools.shared.functions import resolve_max_age - -from .exceptions import GetParameterError, TransformParameterError +from aws_lambda_powertools.utilities.parameters.exceptions import GetParameterError, TransformParameterError if TYPE_CHECKING: from aws_lambda_powertools.utilities.parameters.types import TransformOptions