refactor(parameters): add from __future__ import annotations

aws_lambda_powertools/utilities/parameters/appconfig.py

@@ -2,26 +2,28 @@
 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.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
+
+if TYPE_CHECKING:
+    from botocore.config import Config
+    from mypy_boto3_appconfigdata.client import AppConfigDataClient
 
-from .base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider
+    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.
 

aws_lambda_powertools/utilities/parameters/base.py

@@ -4,48 +4,28 @@
 
 from __future__ import annotations
 
-import base64
-import json
 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
+from aws_lambda_powertools.utilities.parameters.exceptions import GetParameterError, TransformParameterError
 
-DEFAULT_MAX_AGE_SECS = "300"
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.parameters.types import TransformOptions
 
-# 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):
-    value: str | bytes | Dict[str, Any]
+    value: str | bytes | dict[str, Any]
     ttl: datetime
 
 
@@ -54,7 +34,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 +45,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 +94,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 +119,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 +134,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 +191,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 +200,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 +228,7 @@ def _build_cache_key(
 
         Returns
         -------
-        Tuple[str, TransformOptions, bool]
+        tuple[str, TransformOptions, bool]
             Cache key
         """
         return (name, transform, is_nested)
@@ -294,28 +274,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 +328,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:

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,
+}

aws_lambda_powertools/utilities/parameters/dynamodb.py

@@ -2,18 +2,19 @@
 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.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
 
 
@@ -156,11 +157,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 +204,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