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