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