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