fix(parameters): AppConfigProvider when retrieving multiple unique configuration names

aws_lambda_powertools/utilities/parameters/appconfig.py

@@ -94,7 +94,7 @@ def __init__(
         self.environment = environment
         self.current_version = ""
 
-        self._next_token = ""  # nosec - token for get_latest_configuration executions
+        self._next_token: Dict[str, str] = {}  # nosec - token for get_latest_configuration executions
         self.last_returned_value = ""
 
     def _get(self, name: str, **sdk_options) -> str:
@@ -108,19 +108,19 @@ def _get(self, name: str, **sdk_options) -> str:
         sdk_options: dict, optional
             SDK options to propagate to `start_configuration_session` API call
         """
-        if not self._next_token:
+        if name not in self._next_token:
             sdk_options["ConfigurationProfileIdentifier"] = name
             sdk_options["ApplicationIdentifier"] = self.application
             sdk_options["EnvironmentIdentifier"] = self.environment
             response_configuration = self.client.start_configuration_session(**sdk_options)
-            self._next_token = response_configuration["InitialConfigurationToken"]
+            self._next_token[name] = response_configuration["InitialConfigurationToken"]
 
         # The new AppConfig APIs require two API calls to return the configuration
         # First we start the session and after that we retrieve the configuration
         # We need to store the token to use in the next execution
-        response = self.client.get_latest_configuration(ConfigurationToken=self._next_token)
+        response = self.client.get_latest_configuration(ConfigurationToken=self._next_token[name])
         return_value = response["Configuration"].read()
-        self._next_token = response["NextPollConfigurationToken"]
+        self._next_token[name] = response["NextPollConfigurationToken"]
 
         if return_value:
             self.last_returned_value = return_value

docs/utilities/parameters.md

@@ -91,6 +91,9 @@ You can fetch application configurations in AWS AppConfig using `get_app_config`
 
 The following will retrieve the latest version and store it in the cache.
 
+???+ warning
+	When fetching each unique parameter from the AppConfig for the first time, the system executes two API calls, incurring network latency in the process. To enhance performance and caching, it is advisable to consider increasing the `max_age` time.
+
 === "getting_started_appconfig.py"
     ```python hl_lines="5 12"
     --8<-- "examples/parameters/src/getting_started_appconfig.py"

tests/functional/test_utilities_parameters.py

@@ -30,6 +30,12 @@ def mock_name():
     return "".join(random.choices(string.ascii_letters + string.digits + "_.-/", k=random.randrange(3, 200)))
 
 
+@pytest.fixture(scope="function")
+def mock_name_multiple_calls():
+    # Parameter name must match [a-zA-Z0-9_.-/]+
+    return "".join(random.choices(string.ascii_letters + string.digits + "_.-/", k=random.randrange(3, 200)))
+
+
 @pytest.fixture(scope="function")
 def mock_value():
     # Standard parameters can be up to 4 KB
@@ -2179,6 +2185,65 @@ def test_appconf_provider_get_configuration_no_transform(mock_name, config):
         stubber.deactivate()
 
 
+def test_appconf_provider_multiple_calls_with_same_instance(mock_name, mock_name_multiple_calls, config):
+    """
+    Test appconfigprovider.get with default values
+    """
+
+    # Create a new provider
+    # GIVEN a provider instance, we should be able to retrieve multiple appconfig profiles.
+    environment = "dev"
+    application = "myapp"
+    provider = parameters.AppConfigProvider(environment=environment, application=application, config=config)
+
+    mock_body_json_first_call = {"myenvvar1": "Black Panther", "myenvvar2": 3}
+    encoded_message_first_call = json.dumps(mock_body_json_first_call).encode("utf-8")
+    mock_value_first_call = StreamingBody(BytesIO(encoded_message_first_call), len(encoded_message_first_call))
+
+    mock_body_json_second_call = {"myenvvar1": "Thor", "myenvvar2": 5}
+    encoded_message_second_call = json.dumps(mock_body_json_second_call).encode("utf-8")
+    mock_value_second_call = StreamingBody(BytesIO(encoded_message_second_call), len(encoded_message_second_call))
+
+    # WHEN making two API calls using the same provider instance.
+    stubber = stub.Stubber(provider.client)
+
+    response_get_latest_config_first_call = {
+        "Configuration": mock_value_first_call,
+        "NextPollConfigurationToken": "initial_token",
+        "ContentType": "application/json",
+    }
+
+    response_start_config_session = {"InitialConfigurationToken": "initial_token"}
+    stubber.add_response("start_configuration_session", response_start_config_session)
+    stubber.add_response("get_latest_configuration", response_get_latest_config_first_call)
+
+    response_get_latest_config_second_call = {
+        "Configuration": mock_value_second_call,
+        "NextPollConfigurationToken": "initial_token",
+        "ContentType": "application/json",
+    }
+    response_start_config_session = {"InitialConfigurationToken": "initial_token"}
+    stubber.add_response("start_configuration_session", response_start_config_session)
+    stubber.add_response("get_latest_configuration", response_get_latest_config_second_call)
+
+    stubber.activate()
+
+    try:
+        # THEN we should expect different return values.
+        value_first_call: str = provider.get(mock_name)
+        str_value_first_call = value_first_call.decode("utf-8")
+        assert str_value_first_call == json.dumps(mock_body_json_first_call)
+
+        value_second_call: str = provider.get(mock_name_multiple_calls)
+        str_value_second_call = value_second_call.decode("utf-8")
+        assert str_value_second_call == json.dumps(mock_body_json_second_call)
+
+        stubber.assert_no_pending_responses()
+
+    finally:
+        stubber.deactivate()
+
+
 def test_appconf_get_app_config_no_transform(monkeypatch, mock_name):
     """
     Test get_app_config()