Module aws_lambda_powertools.utilities.feature_flags.appconfig
Expand source code
import logging
import traceback
from typing import Any, Dict, Optional, cast
from botocore.config import Config
from aws_lambda_powertools.utilities.parameters import AppConfigProvider, GetParameterError, TransformParameterError
from ...shared import jmespath_utils
from .base import StoreProvider
from .exceptions import ConfigurationStoreError, StoreClientError
logger = logging.getLogger(__name__)
TRANSFORM_TYPE = "json"
class AppConfigStore(StoreProvider):
def __init__(
self,
environment: str,
application: str,
name: str,
max_age: int = 5,
sdk_config: Optional[Config] = None,
envelope: Optional[str] = "",
jmespath_options: Optional[Dict] = 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: Optional[Config]
Botocore Config object to pass during client initialization
envelope : Optional[str]
JMESPath expression to pluck feature flags data from config
jmespath_options : Optional[Dict]
Alternative JMESPath options to be included when filtering expr
"""
super().__init__()
self.environment = environment
self.application = application
self.name = name
self.cache_seconds = max_age
self.config = sdk_config
self.envelope = envelope
self.jmespath_options = jmespath_options
self._conf_store = AppConfigProvider(environment=environment, application=application, config=sdk_config)
def get_configuration(self) -> Dict[str, Any]:
"""Fetch feature schema configuration from AWS AppConfig
Raises
------
ConfigurationStoreError
Any validation error or AppConfig error that can occur
Returns
-------
Dict[str, Any]
parsed JSON dictionary
"""
try:
# parse result conf as JSON, keep in cache for self.max_age seconds
config = cast(
dict,
self._conf_store.get(
name=self.name,
transform=TRANSFORM_TYPE,
max_age=self.cache_seconds,
),
)
if self.envelope:
config = jmespath_utils.extract_data_from_envelope(
data=config, envelope=self.envelope, jmespath_options=self.jmespath_options
)
return config
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
Classes
class AppConfigStore (environment: str, application: str, name: str, max_age: int = 5, sdk_config: Optional[botocore.config.Config] = None, envelope: Optional[str] = '', jmespath_options: Optional[Dict[~KT, ~VT]] = 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
:Optional[Config]
- Botocore Config object to pass during client initialization
envelope
:Optional[str]
- JMESPath expression to pluck feature flags data from config
jmespath_options
:Optional[Dict]
- Alternative JMESPath options to be included when filtering expr
Expand source code
class AppConfigStore(StoreProvider): def __init__( self, environment: str, application: str, name: str, max_age: int = 5, sdk_config: Optional[Config] = None, envelope: Optional[str] = "", jmespath_options: Optional[Dict] = 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: Optional[Config] Botocore Config object to pass during client initialization envelope : Optional[str] JMESPath expression to pluck feature flags data from config jmespath_options : Optional[Dict] Alternative JMESPath options to be included when filtering expr """ super().__init__() self.environment = environment self.application = application self.name = name self.cache_seconds = max_age self.config = sdk_config self.envelope = envelope self.jmespath_options = jmespath_options self._conf_store = AppConfigProvider(environment=environment, application=application, config=sdk_config) def get_configuration(self) -> Dict[str, Any]: """Fetch feature schema configuration from AWS AppConfig Raises ------ ConfigurationStoreError Any validation error or AppConfig error that can occur Returns ------- Dict[str, Any] parsed JSON dictionary """ try: # parse result conf as JSON, keep in cache for self.max_age seconds config = cast( dict, self._conf_store.get( name=self.name, transform=TRANSFORM_TYPE, max_age=self.cache_seconds, ), ) if self.envelope: config = jmespath_utils.extract_data_from_envelope( data=config, envelope=self.envelope, jmespath_options=self.jmespath_options ) return config 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
Ancestors
- StoreProvider
- abc.ABC
Methods
def get_configuration(self) ‑> Dict[str, Any]
-
Fetch feature schema configuration from AWS AppConfig
Raises
ConfigurationStoreError
- Any validation error or AppConfig error that can occur
Returns
Dict[str, Any]
- parsed JSON dictionary
Expand source code
def get_configuration(self) -> Dict[str, Any]: """Fetch feature schema configuration from AWS AppConfig Raises ------ ConfigurationStoreError Any validation error or AppConfig error that can occur Returns ------- Dict[str, Any] parsed JSON dictionary """ try: # parse result conf as JSON, keep in cache for self.max_age seconds config = cast( dict, self._conf_store.get( name=self.name, transform=TRANSFORM_TYPE, max_age=self.cache_seconds, ), ) if self.envelope: config = jmespath_utils.extract_data_from_envelope( data=config, envelope=self.envelope, jmespath_options=self.jmespath_options ) return config 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