Module aws_lambda_powertools.utilities.feature_flags.appconfig

Classes

class AppConfigStore (environment: str, application: str, name: str, max_age: int = 5, sdk_config: Config | None = None, envelope: str | None = '', jmespath_options: dict | None = None, logger: logging.Logger | Logger | None = None, boto_config: Config | None = None, boto3_session: boto3.session.Session | None = None, boto3_client: AppConfigDataClient | None = 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
Appconfig environment, e.g. 'dev/test' etc.
application : str
AppConfig application name, e.g. 'powertools'
name : str
AppConfig configuration name e.g. my_conf
max_age : int
cache expiration time in seconds, or how often to call AppConfig to fetch latest configuration
sdk_config : Config | None
Botocore Config object to pass during client initialization
envelope : str | None
JMESPath expression to pluck feature flags data from config
jmespath_options : dict | None
Alternative JMESPath options to be included when filtering expr
logger : A logging object
Used to log messages. If None is supplied, one will be created.
boto_config : botocore.config.Config, optional
Botocore configuration to pass during client initialization
boto3_session : boto3.Session, optional
Boto3 session to use for AWS API communication
boto3_client : AppConfigDataClient, optional
Boto3 AppConfigDataClient Client to use, boto3_session and boto_config will be ignored if both are provided
Expand source code
class AppConfigStore(StoreProvider):
    def __init__(
        self,
        environment: str,
        application: str,
        name: str,
        max_age: int = 5,
        sdk_config: Config | None = None,
        envelope: str | None = "",
        jmespath_options: dict | None = None,
        logger: logging.Logger | Logger | None = None,
        boto_config: Config | None = None,
        boto3_session: boto3.session.Session | None = None,
        boto3_client: AppConfigDataClient | None = None,
    ):
        """This class fetches JSON schemas from AWS AppConfig

        Parameters
        ----------
        environment: str
            Appconfig environment, e.g. 'dev/test' etc.
        application: str
            AppConfig application name, e.g. 'powertools'
        name: str
            AppConfig configuration name e.g. `my_conf`
        max_age: int
            cache expiration time in seconds, or how often to call AppConfig to fetch latest configuration
        sdk_config: Config | None
            Botocore Config object to pass during client initialization
        envelope : str | None
            JMESPath expression to pluck feature flags data from config
        jmespath_options : dict | None
            Alternative JMESPath options to be included when filtering expr
        logger: A logging object
            Used to log messages. If None is supplied, one will be created.
        boto_config: botocore.config.Config, optional
            Botocore configuration to pass during client initialization
        boto3_session : boto3.Session, optional
            Boto3 session to use for AWS API communication
        boto3_client : AppConfigDataClient, optional
            Boto3 AppConfigDataClient Client to use, boto3_session and boto_config will be ignored if both are provided
        """
        super().__init__()
        self.logger = logger or logging.getLogger(__name__)
        self.environment = environment
        self.application = application
        self.name = name
        self.cache_seconds = max_age
        self.config = sdk_config or boto_config
        self.envelope = envelope
        self.jmespath_options = jmespath_options
        self._conf_store = AppConfigProvider(
            environment=environment,
            application=application,
            config=sdk_config or boto_config,
            boto3_client=boto3_client,
            boto3_session=boto3_session,
        )

    @property
    def get_raw_configuration(self) -> dict[str, Any]:
        """Fetch feature schema configuration from AWS AppConfig"""
        try:
            # parse result conf as JSON, keep in cache for self.max_age seconds
            self.logger.debug(
                "Fetching configuration from the store",
                extra={"param_name": self.name, "max_age": self.cache_seconds},
            )
            return cast(
                dict,
                self._conf_store.get(
                    name=self.name,
                    transform="json",
                    max_age=self.cache_seconds,
                ),
            )
        except (GetParameterError, TransformParameterError) as exc:
            err_msg = traceback.format_exc()
            if "AccessDenied" in err_msg:
                raise StoreClientError(err_msg) from exc
            raise ConfigurationStoreError("Unable to get AWS AppConfig configuration file") from exc

    def get_configuration(self) -> dict[str, Any]:
        """Fetch feature schema configuration from AWS AppConfig

        If envelope is set, it'll extract and return feature flags from configuration,
        otherwise it'll return the entire configuration fetched from AWS AppConfig.

        Raises
        ------
        ConfigurationStoreError
            Any validation error or AppConfig error that can occur

        Returns
        -------
        dict[str, Any]
            parsed JSON dictionary
        """
        config = self.get_raw_configuration

        if self.envelope:
            self.logger.debug("Envelope enabled; extracting data from config", extra={"envelope": self.envelope})
            config = jmespath_utils.query(
                data=config,
                envelope=self.envelope,
                jmespath_options=self.jmespath_options,
            )

        return config

Ancestors

Instance variables

prop get_raw_configuration : dict[str, Any]

Fetch feature schema configuration from AWS AppConfig

Expand source code
@property
def get_raw_configuration(self) -> dict[str, Any]:
    """Fetch feature schema configuration from AWS AppConfig"""
    try:
        # parse result conf as JSON, keep in cache for self.max_age seconds
        self.logger.debug(
            "Fetching configuration from the store",
            extra={"param_name": self.name, "max_age": self.cache_seconds},
        )
        return cast(
            dict,
            self._conf_store.get(
                name=self.name,
                transform="json",
                max_age=self.cache_seconds,
            ),
        )
    except (GetParameterError, TransformParameterError) as exc:
        err_msg = traceback.format_exc()
        if "AccessDenied" in err_msg:
            raise StoreClientError(err_msg) from exc
        raise ConfigurationStoreError("Unable to get AWS AppConfig configuration file") from exc

Methods

def get_configuration(self) ‑> dict[str, typing.Any]

Fetch feature schema configuration from AWS AppConfig

If envelope is set, it'll extract and return feature flags from configuration, otherwise it'll return the entire configuration fetched from AWS AppConfig.

Raises

ConfigurationStoreError
Any validation error or AppConfig error that can occur

Returns

dict[str, Any]
parsed JSON dictionary