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