Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions aws_lambda_powertools/utilities/parameters/appconfig.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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
Expand Down
3 changes: 3 additions & 0 deletions docs/utilities/parameters.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
We make two API calls to fetch each unique configuration name during the first time. This is by design in AppConfig. Please consider adjusting `max_age` parameter to enhance performance.

=== "getting_started_appconfig.py"
```python hl_lines="5 12"
--8<-- "examples/parameters/src/getting_started_appconfig.py"
Expand Down
56 changes: 55 additions & 1 deletion tests/functional/test_utilities_parameters.py
Original file line number Diff line number Diff line change
Expand Up @@ -2171,14 +2171,68 @@ def test_appconf_provider_get_configuration_no_transform(mock_name, config):
stubber.activate()

try:
value: str = provider.get(mock_name)
value: bytes = provider.get(mock_name)
str_value = value.decode("utf-8")
assert str_value == json.dumps(mock_body_json)
stubber.assert_no_pending_responses()
finally:
stubber.deactivate()


def test_appconf_provider_multiple_unique_config_names(mock_name, config):
"""
Test appconfig_provider.get with multiple config names
"""

# 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: bytes = provider.get(mock_name)
value_second_call: bytes = provider.get(f"{mock_name}_ second_config")

assert value_first_call != value_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()
Expand Down