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