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