feat(parameters): Add force_update option

Changes:
- Add new force_update to always load even if there is a cached value

Author: Michael Brewer

aws_lambda_powertools/utilities/parameters/base.py

@@ -38,7 +38,12 @@ def _has_not_expired(self, key: Tuple[str, Optional[str]]) -> bool:
         return key in self.store and self.store[key].ttl >= datetime.now()
 
     def get(
-        self, name: str, max_age: int = DEFAULT_MAX_AGE_SECS, transform: Optional[str] = None, **sdk_options
+        self,
+        name: str,
+        max_age: int = DEFAULT_MAX_AGE_SECS,
+        transform: Optional[str] = None,
+        force_update: bool = False,
+        **sdk_options,
     ) -> Union[str, list, dict, bytes]:
         """
         Retrieve a parameter value or return the cached value
@@ -53,6 +58,8 @@ def get(
             Optional transformation of the parameter value. Supported values
             are "json" for JSON strings and "binary" for base 64 encoded
             values.
+        force_update: bool, optional
+            Force update even before a cached item has expired
         sdk_options: dict, optional
             Arguments that will be passed directly to the underlying API call
 
@@ -76,7 +83,7 @@ def get(
         # an acceptable tradeoff.
         key = (name, transform)
 
-        if self._has_not_expired(key):
+        if not force_update and self._has_not_expired(key):
             return self.store[key].value
 
         try:
@@ -105,6 +112,7 @@ def get_multiple(
         max_age: int = DEFAULT_MAX_AGE_SECS,
         transform: Optional[str] = None,
         raise_on_transform_error: bool = False,
+        force_update: bool = False,
         **sdk_options,
     ) -> Union[Dict[str, str], Dict[str, dict], Dict[str, bytes]]:
         """
@@ -123,6 +131,8 @@ def get_multiple(
         raise_on_transform_error: bool, optional
             Raises an exception if any transform fails, otherwise this will
             return a None value for each transform that failed
+        force_update: bool, optional
+            Force update even before a cached item has expired
         sdk_options: dict, optional
             Arguments that will be passed directly to the underlying API call
 
@@ -137,7 +147,7 @@ def get_multiple(
 
         key = (path, transform)
 
-        if self._has_not_expired(key):
+        if not force_update and self._has_not_expired(key):
             return self.store[key].value
 
         try:

aws_lambda_powertools/utilities/parameters/ssm.py

@@ -92,6 +92,7 @@ def get(
         max_age: int = DEFAULT_MAX_AGE_SECS,
         transform: Optional[str] = None,
         decrypt: bool = False,
+        force_update: bool = False,
         **sdk_options
     ) -> Union[str, list, dict, bytes]:
         """
@@ -109,6 +110,8 @@ def get(
             values.
         decrypt: bool, optional
             If the parameter value should be decrypted
+        force_update: bool, optional
+            Force update even before a cached item has expired
         sdk_options: dict, optional
             Arguments that will be passed directly to the underlying API call
 
@@ -124,7 +127,7 @@ def get(
         # Add to `decrypt` sdk_options to we can have an explicit option for this
         sdk_options["decrypt"] = decrypt
 
-        return super().get(name, max_age, transform, **sdk_options)
+        return super().get(name, max_age, transform, force_update, **sdk_options)
 
     def _get(self, name: str, decrypt: bool = False, **sdk_options) -> str:
         """

tests/functional/test_utilities_parameters.py

@@ -1663,3 +1663,48 @@ def test_get_transform_method_preserve_auto_unhandled(key):
     transform = parameters.base.get_transform_method(key, "auto")
 
     assert transform is None
+
+
+def test_base_provider_get_multiple_force_update(mock_name, mock_value):
+    """
+    Test BaseProvider.get_multiple()  with cached values and force_update is True
+    """
+
+    class TestProvider(BaseProvider):
+        def _get(self, name: str, **kwargs) -> str:
+            raise NotImplementedError()
+
+        def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]:
+            assert path == mock_name
+            return {"A": mock_value}
+
+    provider = TestProvider()
+
+    provider.store[(mock_name, None)] = ExpirableValue({"B": mock_value}, datetime.now() + timedelta(seconds=60))
+
+    value = provider.get_multiple(mock_name, force_update=True)
+
+    assert isinstance(value, dict)
+    assert value["A"] == mock_value
+
+
+def test_base_provider_get_force_update(mock_name, mock_value):
+    """
+    Test BaseProvider.get()  with cached values and force_update is True
+    """
+
+    class TestProvider(BaseProvider):
+        def _get(self, name: str, **kwargs) -> str:
+            return mock_value
+
+        def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]:
+            raise NotImplementedError()
+
+    provider = TestProvider()
+
+    provider.store[(mock_name, None)] = ExpirableValue("not-value", datetime.now() + timedelta(seconds=60))
+
+    value = provider.get(mock_name, force_update=True)
+
+    assert isinstance(value, str)
+    assert value == mock_value