config_ninja.contrib.appconfig

  1"""Integrate with the AWS AppConfig service.
  2
  3## Example
  4
  5The following `config-ninja`_ settings file configures the `AppConfigBackend` to install
  6`/opt/aws/amazon-cloudwatch-agent/etc/amazon-cloudwatch-agent.json` from the latest version deployed
  7through AWS AppConfig:
  8
  9```yaml
 10.. include:: ../../../examples/appconfig-backend.yaml
 11```
 12
 13.. _config-ninja: https://bryant-finney.github.io/config-ninja/config_ninja.html
 14"""
 15
 16from __future__ import annotations
 17
 18import asyncio
 19import logging
 20import warnings
 21from typing import TYPE_CHECKING, Any, AsyncIterator, Iterator, Literal
 22
 23import boto3
 24
 25from config_ninja.backend import Backend
 26
 27try:  # pragma: no cover
 28    from typing import TypeAlias, TypedDict  # type: ignore[attr-defined,unused-ignore]
 29except ImportError:  # pragma: no cover
 30    from typing_extensions import TypeAlias, TypedDict
 31
 32
 33if TYPE_CHECKING:  # pragma: no cover
 34    from mypy_boto3_appconfig.client import AppConfigClient
 35    from mypy_boto3_appconfigdata import AppConfigDataClient
 36
 37__all__ = ['AppConfigBackend']
 38
 39MINIMUM_POLL_INTERVAL_SECONDS = 60
 40
 41
 42OperationName: TypeAlias = Literal['list_applications', 'list_configuration_profiles', 'list_environments']
 43
 44logger = logging.getLogger(__name__)
 45
 46
 47class ErrorT(TypedDict):
 48    """These properties are returned in the `BadRequestException` response's `Error` object.
 49
 50    # ref: https://botocore.amazonaws.com/v1/documentation/api/latest/reference/services/appconfigdata/client/exceptions/BadRequestException.html#badrequestexception
 51    """
 52
 53    Code: str
 54    Message: str
 55
 56
 57class BadRequestExceptionResponse(TypedDict):
 58    """Type information for the `AppConfigData.Client.exceptions.BadRequestException.response`.
 59
 60    # ref: https://botocore.amazonaws.com/v1/documentation/api/latest/reference/services/appconfigdata/client/exceptions/BadRequestException.html#badrequestexception
 61    """
 62
 63    Error: ErrorT
 64    Message: str
 65
 66
 67class AppConfigBackend(Backend):
 68    """Retrieve the deployed configuration from AWS AppConfig.
 69
 70    ## Usage
 71
 72    To retrieve the configuration, use the `AppConfigBackend.get()` method:
 73
 74    >>> backend = AppConfigBackend(appconfigdata_client, 'app-id', 'conf-id', 'env-id')
 75    >>> print(backend.get())
 76    key_0: value_0
 77    key_1: 1
 78    key_2: true
 79    key_3:
 80        - 1
 81        - 2
 82        - 3
 83    """
 84
 85    client: AppConfigDataClient
 86    """The `boto3` client used to communicate with the AWS AppConfig service."""
 87
 88    application_id: str
 89    """See [Creating a namespace for your application in AWS AppConfig](https://docs.aws.amazon.com/appconfig/latest/userguide/appconfig-creating-namespace.html)"""
 90    configuration_profile_id: str
 91    """See [Creating a configuration profile in AWS AppConfig](https://docs.aws.amazon.com/appconfig/latest/userguide/appconfig-creating-configuration-profile.html)"""
 92    environment_id: str
 93    """See [Creating environments for your application in AWS AppConfig](https://docs.aws.amazon.com/appconfig/latest/userguide/appconfig-creating-environment.html)"""
 94
 95    def __init__(
 96        self,
 97        client: AppConfigDataClient,
 98        app_id: str,
 99        config_profile_id: str,
100        env_id: str,
101    ) -> None:
102        """Initialize the backend."""
103        logger.debug(
104            "Initialize: %s(client=%s, app_id='%s', conf_id='%s', env_id='%s')",
105            self.__class__.__name__,
106            client,
107            app_id,
108            config_profile_id,
109            env_id,
110        )
111        self.client = client
112
113        self.application_id = app_id
114        self.configuration_profile_id = config_profile_id
115        self.environment_id = env_id
116
117    def __str__(self) -> str:
118        """Include properties in the string representation.
119
120        >>> print(str( AppConfigBackend(appconfigdata_client, 'app-id', 'conf-id', 'env-id') ))
121        boto3.client('appconfigdata').start_configuration_session(ApplicationIdentifier='app-id', ConfigurationProfileIdentifier='conf-id', EnvironmentIdentifier='env-id')
122        """
123        return (
124            "boto3.client('appconfigdata').start_configuration_session("
125            f"ApplicationIdentifier='{self.application_id}', "
126            f"ConfigurationProfileIdentifier='{self.configuration_profile_id}', "
127            f"EnvironmentIdentifier='{self.environment_id}')"
128        )
129
130    @staticmethod
131    def _get_id_from_name(name: str, operation_name: OperationName, client: AppConfigClient, **kwargs: Any) -> str:
132        out: Iterator[str] = (
133            client.get_paginator(operation_name).paginate(**kwargs).search(f'Items[?Name == `{name}`].Id')
134        )
135        ids = list(out)
136
137        if not ids:
138            raise ValueError(f'no "{operation_name}" results found for Name="{name}"')
139
140        if len(ids) > 1:
141            warnings.warn(
142                f"'{operation_name}' found {len(ids)} results for Name='{name}'; "
143                f"'{ids[0]}' will be used and the others ignored: {ids[1:]}",
144                category=RuntimeWarning,
145                stacklevel=3,
146            )
147
148        return ids[0]
149
150    def get(self) -> str:
151        """Retrieve the latest configuration deployment as a string."""
152        logger.debug('Retrieve latest configuration (%s)', self)
153        token = self.client.start_configuration_session(
154            ApplicationIdentifier=self.application_id,
155            EnvironmentIdentifier=self.environment_id,
156            ConfigurationProfileIdentifier=self.configuration_profile_id,
157            RequiredMinimumPollIntervalInSeconds=MINIMUM_POLL_INTERVAL_SECONDS,
158        )['InitialConfigurationToken']
159
160        resp = self.client.get_latest_configuration(ConfigurationToken=token)
161        return resp['Configuration'].read().decode()
162
163    @classmethod
164    def get_application_id(cls, name: str, client: AppConfigClient) -> str:
165        """Retrieve the application ID for the given application name."""
166        return cls._get_id_from_name(name, 'list_applications', client)
167
168    @classmethod
169    def get_configuration_profile_id(cls, name: str, client: AppConfigClient, application_id: str) -> str:
170        """Retrieve the configuration profile ID for the given configuration profile name."""
171        return cls._get_id_from_name(name, 'list_configuration_profiles', client, ApplicationId=application_id)
172
173    @classmethod
174    def get_environment_id(cls, name: str, client: AppConfigClient, application_id: str) -> str:
175        """Retrieve the environment ID for the given environment name & application ID."""
176        return cls._get_id_from_name(name, 'list_environments', client, ApplicationId=application_id)
177
178    @classmethod
179    def new(  # pylint: disable=arguments-differ  # pyright: ignore[reportIncompatibleMethodOverride]
180        cls,
181        application_name: str,
182        configuration_profile_name: str,
183        environment_name: str,
184        session: boto3.Session | None = None,
185    ) -> AppConfigBackend:
186        """Create a new instance of the backend.
187
188        ## Usage: `AppConfigBackend.new()`
189
190        <!-- fixture is used for doctest but excluded from documentation
191        >>> session = getfixture('mock_session_with_1_id')
192
193        -->
194
195        Use `boto3` to fetch IDs for based on name:
196
197        >>> backend = AppConfigBackend.new('app-name', 'conf-name', 'env-name', session)
198        >>> print(f"{backend}")
199        boto3.client('appconfigdata').start_configuration_session(ApplicationIdentifier='id-1', ConfigurationProfileIdentifier='id-1', EnvironmentIdentifier='id-1')
200
201        ### Error: No IDs Found
202
203        >>> session = getfixture('mock_session_with_0_ids')  # fixture for doctest
204
205        A `ValueError` is raised if no IDs are found for the given name:
206
207        >>> backend = AppConfigBackend.new('app-name', 'conf-name', 'env-name', session)
208        Traceback (most recent call last):
209        ...
210        ValueError: no "list_applications" results found for Name="app-name"
211
212        ### Warning: Multiple IDs Found
213
214        >>> session = getfixture('mock_session_with_2_ids')
215
216        The first ID is used and the others ignored.
217
218        >>> with pytest.warns(RuntimeWarning):
219        ...     backend = AppConfigBackend.new('app-name', 'conf-name', 'env-name', session)
220        """
221        logger.info(
222            'Create new instance: %s(app="%s", conf="%s", env="%s")',
223            cls.__name__,
224            application_name,
225            configuration_profile_name,
226            environment_name,
227        )
228
229        session = session or boto3.Session()
230        appconfig_client = session.client('appconfig')  # pyright: ignore[reportUnknownMemberType]
231        application_id = cls.get_application_id(application_name, appconfig_client)
232        configuration_profile_id = cls.get_configuration_profile_id(
233            configuration_profile_name, appconfig_client, application_id
234        )
235        environment_id = cls.get_environment_id(environment_name, appconfig_client, application_id)
236
237        client: AppConfigDataClient = session.client('appconfigdata')  # pyright: ignore[reportUnknownMemberType]
238
239        return cls(client, application_id, configuration_profile_id, environment_id)
240
241    async def poll(self, interval: int = MINIMUM_POLL_INTERVAL_SECONDS) -> AsyncIterator[str]:
242        """Poll the AppConfig service for configuration changes.
243
244        .. note::
245            Methods written for `asyncio` need to jump through hoops to run as `doctest` tests.
246            To improve the readability of this documentation, each Python code block corresponds to
247            a `doctest` test defined in a private method.
248
249        ## Usage: `AppConfigBackend.poll()`
250
251        ```py
252        In [1]: async for content in backend.poll():
253           ...:     print(content)  # ← executes each time the configuration changes
254        ```
255        ```yaml
256        key_0: value_0
257        key_1: 1
258        key_2: true
259        key_3:
260            - 1
261            - 2
262            - 3
263        ```
264
265        .. note::
266            If polling is done too quickly, the AWS AppConfig client will raise a
267            `BadRequestException`. This is handled automatically by the backend, which will retry
268            the request after waiting for half the given `interval`.
269        """
270        token = self.client.start_configuration_session(
271            ApplicationIdentifier=self.application_id,
272            EnvironmentIdentifier=self.environment_id,
273            ConfigurationProfileIdentifier=self.configuration_profile_id,
274            RequiredMinimumPollIntervalInSeconds=interval,
275        )['InitialConfigurationToken']
276
277        while True:
278            logger.debug('Poll for configuration changes')
279            try:
280                resp = self.client.get_latest_configuration(ConfigurationToken=token)
281            except self.client.exceptions.BadRequestException as exc:
282                exc_resp: BadRequestExceptionResponse = exc.response  # type: ignore[assignment]
283                if exc_resp['Error']['Message'] != 'Request too early':  # pragma: no cover
284                    raise
285                logger.debug('Request too early; retrying in %d seconds', interval / 2)
286                await asyncio.sleep(interval / 2)
287                continue
288
289            token = resp['NextPollConfigurationToken']
290            if content := resp['Configuration'].read():
291                yield content.decode()
292            else:
293                logger.debug('No configuration changes')
294
295            await asyncio.sleep(resp['NextPollIntervalInSeconds'])
296
297    def _async_doctests(self) -> None:
298        """Define `async` `doctest` tests in this method to improve documentation.
299
300        Verify that an empty response to the `boto3` client is handled and the polling continues:
301        >>> backend = AppConfigBackend(appconfigdata_client_first_empty, 'app-id', 'conf-id', 'env-id')
302        >>> content = asyncio.run(anext(backend.poll(interval=0.01)))
303        >>> print(content)
304        key_0: value_0
305        key_1: 1
306        key_2: true
307        key_3:
308            - 1
309            - 2
310            - 3
311
312
313        >>> client = getfixture('mock_poll_too_early')    # seed a `BadRequestException`
314
315        >>> backend = AppConfigBackend(client, 'app-id', 'conf-id', 'env-id')
316        >>> content = asyncio.run(anext(backend.poll(interval=0.01)))   # it is handled successfully
317        >>> print(content)
318        key_0: value_0
319        key_1: 1
320        key_2: true
321        key_3:
322            - 1
323            - 2
324            - 3
325        """
326
327
328logger.debug('successfully imported %s', __name__)