feat(parameters) - environment default variables - refactor maxage code

Author: leandrodamascena

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]:
     """
@@ -187,6 +191,8 @@ def get_app_config(
         >>> print(value)
         My configuration's JSON value
     """
+    # Resolving if will use the default value (5), the value passed by parameter or the environment variable
+    max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE, 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

@@ -26,6 +26,7 @@
 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
@@ -38,7 +39,7 @@
 
 
 # If the environment variable is not set, the default value is 5
-DEFAULT_MAX_AGE_SECS: int = int(os.getenv(constants.PARAMETERS_MAX_AGE, 5))
+DEFAULT_MAX_AGE_SECS = "5"
 
 # These providers will be dynamically initialized on first use of the helper functions
 DEFAULT_PROVIDERS: Dict[str, Any] = {}
@@ -81,7 +82,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,
@@ -125,6 +126,9 @@ def get(
         value: Optional[Union[str, bytes, dict]] = None
         key = (name, transform)
 
+        # Resolving if will use the default value (5), the value passed by parameter or the environment variable
+        max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE, DEFAULT_MAX_AGE_SECS), choice=max_age)
+
         if not force_fetch and self.has_not_expired_in_cache(key):
             return self.store[key].value
 
@@ -153,7 +157,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,
@@ -190,6 +194,9 @@ def get_multiple(
         """
         key = (path, transform)
 
+        # Resolving if will use the default value (5), the value passed by parameter or the environment variable
+        max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE, 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
@@ -162,6 +162,9 @@ def get_secret(
         >>> get_secret("my-secret", VersionId="f658cac0-98a5-41d9-b993-8a76a7799194")
     """
 
+    # Resolving if will use the default value (5), the value passed by parameter or the environment variable
+    max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE, 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()

aws_lambda_powertools/utilities/parameters/ssm.py

@@ -12,6 +12,7 @@
 
 from aws_lambda_powertools.shared import constants
 from aws_lambda_powertools.shared.functions import (
+    resolve_max_age,
     resolve_truthy_env_var_choice,
     slice_dictionary,
 )
@@ -115,7 +116,7 @@ def __init__(
     def get(  # type: ignore[override]
         self,
         name: str,
-        max_age: int = DEFAULT_MAX_AGE_SECS,
+        max_age: Optional[int] = None,
         transform: TransformOptions = None,
         decrypt: Optional[bool] = None,
         force_fetch: bool = False,
@@ -150,7 +151,10 @@ def get(  # type: ignore[override]
             When the parameter provider fails to transform a parameter value.
         """
 
-        # Resolving if will use the value passed by parameter or the environment
+        # Resolving if will use the default value (5), the value passed by parameter or the environment variable
+        max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE, DEFAULT_MAX_AGE_SECS), choice=max_age)
+
+        # Resolving if will use the default value (False), the value passed by parameter or the environment variable
         decrypt = resolve_truthy_env_var_choice(
             env=os.getenv(constants.PARAMETERS_DEFAULT_DECRYPT, "false"), choice=decrypt
         )
@@ -223,7 +227,7 @@ def get_parameters_by_name(
         parameters: Dict[str, Dict],
         transform: TransformOptions = None,
         decrypt: Optional[bool] = None,
-        max_age: int = DEFAULT_MAX_AGE_SECS,
+        max_age: Optional[int] = None,
         raise_on_error: bool = True,
     ) -> Dict[str, str] | Dict[str, bytes] | Dict[str, dict]:
         """
@@ -270,6 +274,9 @@ def get_parameters_by_name(
             When "_errors" reserved key is in parameters to be fetched from SSM.
         """
 
+        # Resolving if will use the default value (5), the value passed by parameter or the environment variable
+        max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE, DEFAULT_MAX_AGE_SECS), choice=max_age)
+
         # Resolving if will use the default value (False), the value passed by parameter or the environment variable
         decrypt = resolve_truthy_env_var_choice(
             env=os.getenv(constants.PARAMETERS_DEFAULT_DECRYPT, "false"), choice=decrypt
@@ -505,7 +512,7 @@ def get_parameter(
     transform: Optional[str] = None,
     decrypt: Optional[bool] = None,
     force_fetch: bool = False,
-    max_age: int = DEFAULT_MAX_AGE_SECS,
+    max_age: Optional[int] = None,
     **sdk_options,
 ) -> Union[str, dict, bytes]:
     """
@@ -559,6 +566,9 @@ def get_parameter(
     if "ssm" not in DEFAULT_PROVIDERS:
         DEFAULT_PROVIDERS["ssm"] = SSMProvider()
 
+    # Resolving if will use the default value (5), the value passed by parameter or the environment variable
+    max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE, DEFAULT_MAX_AGE_SECS), choice=max_age)
+
     # Resolving if will use the default value (False), the value passed by parameter or the environment variable
     decrypt = resolve_truthy_env_var_choice(
         env=os.getenv(constants.PARAMETERS_DEFAULT_DECRYPT, "false"), choice=decrypt
@@ -578,7 +588,7 @@ def get_parameters(
     recursive: bool = True,
     decrypt: Optional[bool] = None,
     force_fetch: bool = False,
-    max_age: int = DEFAULT_MAX_AGE_SECS,
+    max_age: Optional[int] = None,
     raise_on_transform_error: bool = False,
     **sdk_options,
 ) -> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]:
@@ -638,6 +648,9 @@ def get_parameters(
     if "ssm" not in DEFAULT_PROVIDERS:
         DEFAULT_PROVIDERS["ssm"] = SSMProvider()
 
+    # Resolving if will use the default value (5), the value passed by parameter or the environment variable
+    max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE, DEFAULT_MAX_AGE_SECS), choice=max_age)
+
     # Resolving if will use the default value (False), the value passed by parameter or the environment variable
     decrypt = resolve_truthy_env_var_choice(
         env=os.getenv(constants.PARAMETERS_DEFAULT_DECRYPT, "false"), choice=decrypt
@@ -661,7 +674,7 @@ def get_parameters_by_name(
     parameters: Dict[str, Dict],
     transform: None = None,
     decrypt: Optional[bool] = None,
-    max_age: int = DEFAULT_MAX_AGE_SECS,
+    max_age: Optional[int] = None,
     raise_on_error: bool = True,
 ) -> Dict[str, str]:
     ...
@@ -672,7 +685,7 @@ def get_parameters_by_name(
     parameters: Dict[str, Dict],
     transform: Literal["binary"],
     decrypt: Optional[bool] = None,
-    max_age: int = DEFAULT_MAX_AGE_SECS,
+    max_age: Optional[int] = None,
     raise_on_error: bool = True,
 ) -> Dict[str, bytes]:
     ...
@@ -683,7 +696,7 @@ def get_parameters_by_name(
     parameters: Dict[str, Dict],
     transform: Literal["json"],
     decrypt: Optional[bool] = None,
-    max_age: int = DEFAULT_MAX_AGE_SECS,
+    max_age: Optional[int] = None,
     raise_on_error: bool = True,
 ) -> Dict[str, Dict[str, Any]]:
     ...
@@ -694,7 +707,7 @@ def get_parameters_by_name(
     parameters: Dict[str, Dict],
     transform: Literal["auto"],
     decrypt: Optional[bool] = None,
-    max_age: int = DEFAULT_MAX_AGE_SECS,
+    max_age: Optional[int] = None,
     raise_on_error: bool = True,
 ) -> Union[Dict[str, str], Dict[str, dict]]:
     ...
@@ -704,7 +717,7 @@ def get_parameters_by_name(
     parameters: Dict[str, Any],
     transform: TransformOptions = None,
     decrypt: Optional[bool] = None,
-    max_age: int = DEFAULT_MAX_AGE_SECS,
+    max_age: Optional[int] = None,
     raise_on_error: bool = True,
 ) -> Union[Dict[str, str], Dict[str, bytes], Dict[str, dict]]:
     """
@@ -758,6 +771,9 @@ def get_parameters_by_name(
     # NOTE: Decided against using multi-thread due to single-thread outperforming in 128M and 1G + timeout risk
     # see: https://github.com/awslabs/aws-lambda-powertools-python/issues/1040#issuecomment-1299954613
 
+    # Resolving if will use the default value (5), the value passed by parameter or the environment variable
+    max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE, DEFAULT_MAX_AGE_SECS), choice=max_age)
+
     # Resolving if will use the default value (False), the value passed by parameter or the environment variable
     decrypt = resolve_truthy_env_var_choice(
         env=os.getenv(constants.PARAMETERS_DEFAULT_DECRYPT, "false"), choice=decrypt

tests/unit/test_shared_functions.py

@@ -1,3 +1,4 @@
+import os
 import warnings
 from dataclasses import dataclass
 
@@ -10,10 +11,12 @@
     powertools_debug_is_set,
     powertools_dev_is_set,
     resolve_env_var_choice,
+    resolve_max_age,
     resolve_truthy_env_var_choice,
     strtobool,
 )
 from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+from aws_lambda_powertools.utilities.parameters.base import DEFAULT_MAX_AGE_SECS
 
 
 def test_resolve_env_var_choice_explicit_wins_over_env_var():
@@ -103,3 +106,35 @@ class DummyDataclass:
 @pytest.mark.parametrize("data", [False, True, "", 10, [], {}, object])
 def test_extract_event_any(data):
     assert extract_event_from_common_models(data) == data
+
+
+def test_resolve_max_age_explicit_wins_over_env_var(monkeypatch: pytest.MonkeyPatch):
+    # GIVEN POWERTOOLS_PARAMETERS_MAX_AGE environment variable is set
+    monkeypatch.setenv(constants.PARAMETERS_MAX_AGE, "20")
+
+    # WHEN the choice is set explicitly
+    max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE, DEFAULT_MAX_AGE_SECS), choice=10)
+
+    # THEN the result must be the choice
+    assert max_age == 10
+
+
+def test_resolve_max_age_with_default_value():
+    # GIVEN POWERTOOLS_PARAMETERS_MAX_AGE is not set
+
+    # WHEN the choice is set to None
+    max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE, DEFAULT_MAX_AGE_SECS), choice=None)
+
+    # THEN the result must be the default value (DEFAULT_MAX_AGE_SECS)
+    assert max_age == int(DEFAULT_MAX_AGE_SECS)
+
+
+def test_resolve_max_age_env_var_wins_over_default_value(monkeypatch: pytest.MonkeyPatch):
+    # GIVEN POWERTOOLS_PARAMETERS_MAX_AGE environment variable is set
+    monkeypatch.setenv(constants.PARAMETERS_MAX_AGE, "20")
+
+    # WHEN the choice is set to None
+    max_age = resolve_max_age(env=os.getenv(constants.PARAMETERS_MAX_AGE, DEFAULT_MAX_AGE_SECS), choice=None)
+
+    # THEN the result must be the environment variable value
+    assert max_age == 20