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