Module aws_lambda_powertools.utilities.feature_toggles.appconfig_fetcher

Expand source code
import logging
from typing import Any, Dict, Optional

from botocore.config import Config

from aws_lambda_powertools.utilities.parameters import AppConfigProvider, GetParameterError, TransformParameterError

from .exceptions import ConfigurationError
from .schema_fetcher import SchemaFetcher

logger = logging.getLogger(__name__)


TRANSFORM_TYPE = "json"


class AppConfigFetcher(SchemaFetcher):
    def __init__(
        self,
        environment: str,
        service: str,
        configuration_name: str,
        cache_seconds: int,
        config: Optional[Config] = None,
    ):
        """This class fetches JSON schemas from AWS AppConfig

        Parameters
        ----------
        environment: str
            what appconfig environment to use 'dev/test' etc.
        service: str
            what service name to use from the supplied environment
        configuration_name: str
            what configuration to take from the environment & service combination
        cache_seconds: int
            cache expiration time, how often to call AppConfig to fetch latest configuration
        config: Optional[Config]
            boto3 client configuration
        """
        super().__init__(configuration_name, cache_seconds)
        self._logger = logger
        self._conf_store = AppConfigProvider(environment=environment, application=service, config=config)

    def get_json_configuration(self) -> Dict[str, Any]:
        """Get configuration string from AWs AppConfig and return the parsed JSON dictionary

        Raises
        ------
        ConfigurationError
            Any validation error or appconfig error that can occur

        Returns
        -------
        Dict[str, Any]
            parsed JSON dictionary
        """
        try:
            return self._conf_store.get(
                name=self.configuration_name,
                transform=TRANSFORM_TYPE,
                max_age=self._cache_seconds,
            )  # parse result conf as JSON, keep in cache for self.max_age seconds
        except (GetParameterError, TransformParameterError) as exc:
            error_str = f"unable to get AWS AppConfig configuration file, exception={str(exc)}"
            self._logger.error(error_str)
            raise ConfigurationError(error_str)

Classes

class AppConfigFetcher (environment: str, service: str, configuration_name: str, cache_seconds: int, config: Union[botocore.config.Config, NoneType] = None)

Helper class that provides a standard way to create an ABC using inheritance.

This class fetches JSON schemas from AWS AppConfig

Parameters

environment : str
what appconfig environment to use 'dev/test' etc.
service : str
what service name to use from the supplied environment
configuration_name : str
what configuration to take from the environment & service combination
cache_seconds : int
cache expiration time, how often to call AppConfig to fetch latest configuration
config : Optional[Config]
boto3 client configuration
Expand source code
class AppConfigFetcher(SchemaFetcher):
    def __init__(
        self,
        environment: str,
        service: str,
        configuration_name: str,
        cache_seconds: int,
        config: Optional[Config] = None,
    ):
        """This class fetches JSON schemas from AWS AppConfig

        Parameters
        ----------
        environment: str
            what appconfig environment to use 'dev/test' etc.
        service: str
            what service name to use from the supplied environment
        configuration_name: str
            what configuration to take from the environment & service combination
        cache_seconds: int
            cache expiration time, how often to call AppConfig to fetch latest configuration
        config: Optional[Config]
            boto3 client configuration
        """
        super().__init__(configuration_name, cache_seconds)
        self._logger = logger
        self._conf_store = AppConfigProvider(environment=environment, application=service, config=config)

    def get_json_configuration(self) -> Dict[str, Any]:
        """Get configuration string from AWs AppConfig and return the parsed JSON dictionary

        Raises
        ------
        ConfigurationError
            Any validation error or appconfig error that can occur

        Returns
        -------
        Dict[str, Any]
            parsed JSON dictionary
        """
        try:
            return self._conf_store.get(
                name=self.configuration_name,
                transform=TRANSFORM_TYPE,
                max_age=self._cache_seconds,
            )  # parse result conf as JSON, keep in cache for self.max_age seconds
        except (GetParameterError, TransformParameterError) as exc:
            error_str = f"unable to get AWS AppConfig configuration file, exception={str(exc)}"
            self._logger.error(error_str)
            raise ConfigurationError(error_str)

Ancestors

Methods

def get_json_configuration(self) ‑> Dict[str, Any]

Get configuration string from AWs AppConfig and return the parsed JSON dictionary

Raises

ConfigurationError
Any validation error or appconfig error that can occur

Returns

Dict[str, Any]
parsed JSON dictionary
Expand source code
def get_json_configuration(self) -> Dict[str, Any]:
    """Get configuration string from AWs AppConfig and return the parsed JSON dictionary

    Raises
    ------
    ConfigurationError
        Any validation error or appconfig error that can occur

    Returns
    -------
    Dict[str, Any]
        parsed JSON dictionary
    """
    try:
        return self._conf_store.get(
            name=self.configuration_name,
            transform=TRANSFORM_TYPE,
            max_age=self._cache_seconds,
        )  # parse result conf as JSON, keep in cache for self.max_age seconds
    except (GetParameterError, TransformParameterError) as exc:
        error_str = f"unable to get AWS AppConfig configuration file, exception={str(exc)}"
        self._logger.error(error_str)
        raise ConfigurationError(error_str)