feat(parameters): Configure max_age and decrypt parameters via environment variables

aws_lambda_powertools/shared/constants.py

@@ -22,6 +22,9 @@
 
 IDEMPOTENCY_DISABLED_ENV: str = "POWERTOOLS_IDEMPOTENCY_DISABLED"
 
+PARAMETERS_SSM_DECRYPT_ENV: str = "POWERTOOLS_PARAMETERS_SSM_DECRYPT"
+PARAMETERS_MAX_AGE_ENV: str = "POWERTOOLS_PARAMETERS_MAX_AGE"
+
 LOGGER_LAMBDA_CONTEXT_KEYS = [
     "function_arn",
     "function_memory_size",

aws_lambda_powertools/shared/functions.py

@@ -51,6 +51,11 @@ 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_max_age(env: str, choice: Optional[int]) -> int:
+    """Resolve max age value"""
+    return choice if choice is not None else int(env)
+
+
 @overload
 def resolve_env_var_choice(env: Optional[str], choice: float) -> float:
     ...

aws_lambda_powertools/utilities/parameters/appconfig.py

@@ -14,8 +14,12 @@
 if TYPE_CHECKING:
     from mypy_boto3_appconfigdata import AppConfigDataClient
 
-from ...shared import constants
-from ...shared.functions import resolve_env_var_choice
+from aws_lambda_powertools.shared import constants
+from aws_lambda_powertools.shared.functions import (
+    resolve_env_var_choice,
+    resolve_max_age,
+)
+
 from .base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider
 
 
@@ -136,7 +140,7 @@ def get_app_config(
     application: Optional[str] = None,
     transform: TransformOptions = None,
     force_fetch: bool = False,
-    max_age: int = DEFAULT_MAX_AGE_SECS,
+    max_age: Optional[int] = None,
     **sdk_options
 ) -> Union[str, list, dict, bytes]:
     """
@@ -154,7 +158,7 @@ def get_app_config(
         Transforms the content from a JSON object ('json') or base64 binary string ('binary')
     force_fetch: bool, optional
         Force update even before a cached item has expired, defaults to False
-    max_age: int
+    max_age: int, optional
         Maximum age of the cached value
     sdk_options: dict, optional
         SDK options to propagate to `start_configuration_session` API call
@@ -187,6 +191,8 @@ def get_app_config(
         >>> print(value)
         My configuration's JSON value
     """
+    # If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS
+    max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE_ENV, DEFAULT_MAX_AGE_SECS), choice=max_age)
 
     # Only create the provider if this function is called at least once
     if "appconfig" not in DEFAULT_PROVIDERS:

aws_lambda_powertools/utilities/parameters/base.py

@@ -5,6 +5,7 @@
 
 import base64
 import json
+import os
 from abc import ABC, abstractmethod
 from datetime import datetime, timedelta
 from typing import (
@@ -24,6 +25,8 @@
 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.utilities.parameters.types import TransformOptions
 
 from .exceptions import GetParameterError, TransformParameterError
@@ -35,7 +38,8 @@
     from mypy_boto3_ssm import SSMClient
 
 
-DEFAULT_MAX_AGE_SECS = 5
+DEFAULT_MAX_AGE_SECS = "5"
+
 # These providers will be dynamically initialized on first use of the helper functions
 DEFAULT_PROVIDERS: Dict[str, Any] = {}
 TRANSFORM_METHOD_JSON = "json"
@@ -77,7 +81,7 @@ def has_not_expired_in_cache(self, key: Tuple[str, TransformOptions]) -> bool:
     def get(
         self,
         name: str,
-        max_age: int = DEFAULT_MAX_AGE_SECS,
+        max_age: Optional[int] = None,
         transform: TransformOptions = None,
         force_fetch: bool = False,
         **sdk_options,
@@ -121,6 +125,9 @@ def get(
         value: Optional[Union[str, bytes, dict]] = None
         key = (name, transform)
 
+        # If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS
+        max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE_ENV, DEFAULT_MAX_AGE_SECS), choice=max_age)
+
         if not force_fetch and self.has_not_expired_in_cache(key):
             return self.store[key].value
 
@@ -149,7 +156,7 @@ def _get(self, name: str, **sdk_options) -> Union[str, bytes]:
     def get_multiple(
         self,
         path: str,
-        max_age: int = DEFAULT_MAX_AGE_SECS,
+        max_age: Optional[int] = None,
         transform: TransformOptions = None,
         raise_on_transform_error: bool = False,
         force_fetch: bool = False,
@@ -186,6 +193,9 @@ def get_multiple(
         """
         key = (path, transform)
 
+        # If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS
+        max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE_ENV, DEFAULT_MAX_AGE_SECS), choice=max_age)
+
         if not force_fetch and self.has_not_expired_in_cache(key):
             return self.store[key].value  # type: ignore # need to revisit entire typing here
 

aws_lambda_powertools/utilities/parameters/secrets.py

@@ -3,6 +3,7 @@
 """
 
 
+import os
 from typing import TYPE_CHECKING, Any, Dict, Optional, Union
 
 import boto3
@@ -11,6 +12,9 @@
 if TYPE_CHECKING:
     from mypy_boto3_secretsmanager import SecretsManagerClient
 
+from aws_lambda_powertools.shared import constants
+from aws_lambda_powertools.shared.functions import resolve_max_age
+
 from .base import DEFAULT_MAX_AGE_SECS, DEFAULT_PROVIDERS, BaseProvider
 
 
@@ -111,11 +115,7 @@ def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
 
 
 def get_secret(
-    name: str,
-    transform: Optional[str] = None,
-    force_fetch: bool = False,
-    max_age: int = DEFAULT_MAX_AGE_SECS,
-    **sdk_options
+    name: str, transform: Optional[str] = None, force_fetch: bool = False, max_age: Optional[int] = None, **sdk_options
 ) -> Union[str, dict, bytes]:
     """
     Retrieve a parameter value from AWS Secrets Manager
@@ -128,7 +128,7 @@ def get_secret(
         Transforms the content from a JSON object ('json') or base64 binary string ('binary')
     force_fetch: bool, optional
         Force update even before a cached item has expired, defaults to False
-    max_age: int
+    max_age: int, optional
         Maximum age of the cached value
     sdk_options: dict, optional
         Dictionary of options that will be passed to the get_secret_value call
@@ -162,6 +162,9 @@ def get_secret(
         >>> get_secret("my-secret", VersionId="f658cac0-98a5-41d9-b993-8a76a7799194")
     """
 
+    # If max_age is not set, resolve it from the environment variable, defaulting to DEFAULT_MAX_AGE_SECS
+    max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE_ENV, DEFAULT_MAX_AGE_SECS), choice=max_age)
+
     # Only create the provider if this function is called at least once
     if "secrets" not in DEFAULT_PROVIDERS:
         DEFAULT_PROVIDERS["secrets"] = SecretsProvider()