Module aws_lambda_powertools.utilities.feature_toggles.configuration_store
Expand source code
import logging
from typing import Any, Dict, List, Optional, cast
from . import schema
from .exceptions import ConfigurationError
from .schema_fetcher import SchemaFetcher
logger = logging.getLogger(__name__)
class ConfigurationStore:
def __init__(self, schema_fetcher: SchemaFetcher):
"""constructor
Parameters
----------
schema_fetcher: SchemaFetcher
A schema JSON fetcher, can be AWS AppConfig, Hashicorp Consul etc.
"""
self._logger = logger
self._schema_fetcher = schema_fetcher
self._schema_validator = schema.SchemaValidator(self._logger)
def _match_by_action(self, action: str, condition_value: Any, context_value: Any) -> bool:
if not context_value:
return False
mapping_by_action = {
schema.ACTION.EQUALS.value: lambda a, b: a == b,
schema.ACTION.STARTSWITH.value: lambda a, b: a.startswith(b),
schema.ACTION.ENDSWITH.value: lambda a, b: a.endswith(b),
schema.ACTION.CONTAINS.value: lambda a, b: a in b,
}
try:
func = mapping_by_action.get(action, lambda a, b: False)
return func(context_value, condition_value)
except Exception as exc:
self._logger.error(f"caught exception while matching action, action={action}, exception={str(exc)}")
return False
def _is_rule_matched(self, feature_name: str, rule: Dict[str, Any], rules_context: Dict[str, Any]) -> bool:
rule_name = rule.get(schema.RULE_NAME_KEY, "")
rule_default_value = rule.get(schema.RULE_DEFAULT_VALUE)
conditions = cast(List[Dict], rule.get(schema.CONDITIONS_KEY))
for condition in conditions:
context_value = rules_context.get(str(condition.get(schema.CONDITION_KEY)))
if not self._match_by_action(
condition.get(schema.CONDITION_ACTION, ""),
condition.get(schema.CONDITION_VALUE),
context_value,
):
logger.debug(
f"rule did not match action, rule_name={rule_name}, rule_default_value={rule_default_value}, "
f"feature_name={feature_name}, context_value={str(context_value)} "
)
# context doesn't match condition
return False
# if we got here, all conditions match
logger.debug(
f"rule matched, rule_name={rule_name}, rule_default_value={rule_default_value}, "
f"feature_name={feature_name}"
)
return True
return False
def _handle_rules(
self,
*,
feature_name: str,
rules_context: Dict[str, Any],
feature_default_value: bool,
rules: List[Dict[str, Any]],
) -> bool:
for rule in rules:
rule_default_value = rule.get(schema.RULE_DEFAULT_VALUE)
if self._is_rule_matched(feature_name, rule, rules_context):
return bool(rule_default_value)
# no rule matched, return default value of feature
logger.debug(
f"no rule matched, returning default value of feature, feature_default_value={feature_default_value}, "
f"feature_name={feature_name}"
)
return feature_default_value
return False
def get_configuration(self) -> Dict[str, Any]:
"""Get configuration string from AWs AppConfig and returned the parsed JSON dictionary
Raises
------
ConfigurationError
Any validation error or appconfig error that can occur
Returns
------
Dict[str, Any]
parsed JSON dictionary
"""
# parse result conf as JSON, keep in cache for self.max_age seconds
config = self._schema_fetcher.get_json_configuration()
# validate schema
self._schema_validator.validate_json_schema(config)
return config
def get_feature_toggle(
self, *, feature_name: str, rules_context: Optional[Dict[str, Any]] = None, value_if_missing: bool
) -> bool:
"""Get a feature toggle boolean value. Value is calculated according to a set of rules and conditions.
See below for explanation.
Parameters
----------
feature_name: str
feature name that you wish to fetch
rules_context: Optional[Dict[str, Any]]
dict of attributes that you would like to match the rules
against, can be {'tenant_id: 'X', 'username':' 'Y', 'region': 'Z'} etc.
value_if_missing: bool
this will be the returned value in case the feature toggle doesn't exist in
the schema or there has been an error while fetching the
configuration from appconfig
Returns
------
bool
calculated feature toggle value. several possibilities:
1. if the feature doesn't appear in the schema or there has been an error fetching the
configuration -> error/warning log would appear and value_if_missing is returned
2. feature exists and has no rules or no rules have matched -> return feature_default_value of
the defined feature
3. feature exists and a rule matches -> rule_default_value of rule is returned
"""
if rules_context is None:
rules_context = {}
try:
toggles_dict: Dict[str, Any] = self.get_configuration()
except ConfigurationError:
logger.error("unable to get feature toggles JSON, returning provided value_if_missing value")
return value_if_missing
feature: Dict[str, Dict] = toggles_dict.get(schema.FEATURES_KEY, {}).get(feature_name, None)
if feature is None:
logger.warning(
f"feature does not appear in configuration, using provided value_if_missing, "
f"feature_name={feature_name}, value_if_missing={value_if_missing}"
)
return value_if_missing
rules_list = feature.get(schema.RULES_KEY)
feature_default_value = feature.get(schema.FEATURE_DEFAULT_VAL_KEY)
if not rules_list:
# not rules but has a value
logger.debug(
f"no rules found, returning feature default value, feature_name={feature_name}, "
f"default_value={feature_default_value}"
)
return bool(feature_default_value)
# look for first rule match
logger.debug(
f"looking for rule match, feature_name={feature_name}, feature_default_value={feature_default_value}"
)
return self._handle_rules(
feature_name=feature_name,
rules_context=rules_context,
feature_default_value=bool(feature_default_value),
rules=cast(List, rules_list),
)
def get_all_enabled_feature_toggles(self, *, rules_context: Optional[Dict[str, Any]] = None) -> List[str]:
"""Get all enabled feature toggles while also taking into account rule_context
(when a feature has defined rules)
Parameters
----------
rules_context: Optional[Dict[str, Any]]
dict of attributes that you would like to match the rules
against, can be `{'tenant_id: 'X', 'username':' 'Y', 'region': 'Z'}` etc.
Returns
----------
List[str]
a list of all features name that are enabled by also taking into account
rule_context (when a feature has defined rules)
"""
if rules_context is None:
rules_context = {}
try:
toggles_dict: Dict[str, Any] = self.get_configuration()
except ConfigurationError:
logger.error("unable to get feature toggles JSON")
return []
ret_list = []
features: Dict[str, Any] = toggles_dict.get(schema.FEATURES_KEY, {})
for feature_name, feature_dict_def in features.items():
rules_list = feature_dict_def.get(schema.RULES_KEY, [])
feature_default_value = feature_dict_def.get(schema.FEATURE_DEFAULT_VAL_KEY)
if feature_default_value and not rules_list:
self._logger.debug(
f"feature is enabled by default and has no defined rules, feature_name={feature_name}"
)
ret_list.append(feature_name)
elif self._handle_rules(
feature_name=feature_name,
rules_context=rules_context,
feature_default_value=feature_default_value,
rules=rules_list,
):
self._logger.debug(f"feature's calculated value is True, feature_name={feature_name}")
ret_list.append(feature_name)
return ret_list
Classes
class ConfigurationStore (schema_fetcher: SchemaFetcher)
-
constructor
Parameters
schema_fetcher
:SchemaFetcher
- A schema JSON fetcher, can be AWS AppConfig, Hashicorp Consul etc.
Expand source code
class ConfigurationStore: def __init__(self, schema_fetcher: SchemaFetcher): """constructor Parameters ---------- schema_fetcher: SchemaFetcher A schema JSON fetcher, can be AWS AppConfig, Hashicorp Consul etc. """ self._logger = logger self._schema_fetcher = schema_fetcher self._schema_validator = schema.SchemaValidator(self._logger) def _match_by_action(self, action: str, condition_value: Any, context_value: Any) -> bool: if not context_value: return False mapping_by_action = { schema.ACTION.EQUALS.value: lambda a, b: a == b, schema.ACTION.STARTSWITH.value: lambda a, b: a.startswith(b), schema.ACTION.ENDSWITH.value: lambda a, b: a.endswith(b), schema.ACTION.CONTAINS.value: lambda a, b: a in b, } try: func = mapping_by_action.get(action, lambda a, b: False) return func(context_value, condition_value) except Exception as exc: self._logger.error(f"caught exception while matching action, action={action}, exception={str(exc)}") return False def _is_rule_matched(self, feature_name: str, rule: Dict[str, Any], rules_context: Dict[str, Any]) -> bool: rule_name = rule.get(schema.RULE_NAME_KEY, "") rule_default_value = rule.get(schema.RULE_DEFAULT_VALUE) conditions = cast(List[Dict], rule.get(schema.CONDITIONS_KEY)) for condition in conditions: context_value = rules_context.get(str(condition.get(schema.CONDITION_KEY))) if not self._match_by_action( condition.get(schema.CONDITION_ACTION, ""), condition.get(schema.CONDITION_VALUE), context_value, ): logger.debug( f"rule did not match action, rule_name={rule_name}, rule_default_value={rule_default_value}, " f"feature_name={feature_name}, context_value={str(context_value)} " ) # context doesn't match condition return False # if we got here, all conditions match logger.debug( f"rule matched, rule_name={rule_name}, rule_default_value={rule_default_value}, " f"feature_name={feature_name}" ) return True return False def _handle_rules( self, *, feature_name: str, rules_context: Dict[str, Any], feature_default_value: bool, rules: List[Dict[str, Any]], ) -> bool: for rule in rules: rule_default_value = rule.get(schema.RULE_DEFAULT_VALUE) if self._is_rule_matched(feature_name, rule, rules_context): return bool(rule_default_value) # no rule matched, return default value of feature logger.debug( f"no rule matched, returning default value of feature, feature_default_value={feature_default_value}, " f"feature_name={feature_name}" ) return feature_default_value return False def get_configuration(self) -> Dict[str, Any]: """Get configuration string from AWs AppConfig and returned the parsed JSON dictionary Raises ------ ConfigurationError Any validation error or appconfig error that can occur Returns ------ Dict[str, Any] parsed JSON dictionary """ # parse result conf as JSON, keep in cache for self.max_age seconds config = self._schema_fetcher.get_json_configuration() # validate schema self._schema_validator.validate_json_schema(config) return config def get_feature_toggle( self, *, feature_name: str, rules_context: Optional[Dict[str, Any]] = None, value_if_missing: bool ) -> bool: """Get a feature toggle boolean value. Value is calculated according to a set of rules and conditions. See below for explanation. Parameters ---------- feature_name: str feature name that you wish to fetch rules_context: Optional[Dict[str, Any]] dict of attributes that you would like to match the rules against, can be {'tenant_id: 'X', 'username':' 'Y', 'region': 'Z'} etc. value_if_missing: bool this will be the returned value in case the feature toggle doesn't exist in the schema or there has been an error while fetching the configuration from appconfig Returns ------ bool calculated feature toggle value. several possibilities: 1. if the feature doesn't appear in the schema or there has been an error fetching the configuration -> error/warning log would appear and value_if_missing is returned 2. feature exists and has no rules or no rules have matched -> return feature_default_value of the defined feature 3. feature exists and a rule matches -> rule_default_value of rule is returned """ if rules_context is None: rules_context = {} try: toggles_dict: Dict[str, Any] = self.get_configuration() except ConfigurationError: logger.error("unable to get feature toggles JSON, returning provided value_if_missing value") return value_if_missing feature: Dict[str, Dict] = toggles_dict.get(schema.FEATURES_KEY, {}).get(feature_name, None) if feature is None: logger.warning( f"feature does not appear in configuration, using provided value_if_missing, " f"feature_name={feature_name}, value_if_missing={value_if_missing}" ) return value_if_missing rules_list = feature.get(schema.RULES_KEY) feature_default_value = feature.get(schema.FEATURE_DEFAULT_VAL_KEY) if not rules_list: # not rules but has a value logger.debug( f"no rules found, returning feature default value, feature_name={feature_name}, " f"default_value={feature_default_value}" ) return bool(feature_default_value) # look for first rule match logger.debug( f"looking for rule match, feature_name={feature_name}, feature_default_value={feature_default_value}" ) return self._handle_rules( feature_name=feature_name, rules_context=rules_context, feature_default_value=bool(feature_default_value), rules=cast(List, rules_list), ) def get_all_enabled_feature_toggles(self, *, rules_context: Optional[Dict[str, Any]] = None) -> List[str]: """Get all enabled feature toggles while also taking into account rule_context (when a feature has defined rules) Parameters ---------- rules_context: Optional[Dict[str, Any]] dict of attributes that you would like to match the rules against, can be `{'tenant_id: 'X', 'username':' 'Y', 'region': 'Z'}` etc. Returns ---------- List[str] a list of all features name that are enabled by also taking into account rule_context (when a feature has defined rules) """ if rules_context is None: rules_context = {} try: toggles_dict: Dict[str, Any] = self.get_configuration() except ConfigurationError: logger.error("unable to get feature toggles JSON") return [] ret_list = [] features: Dict[str, Any] = toggles_dict.get(schema.FEATURES_KEY, {}) for feature_name, feature_dict_def in features.items(): rules_list = feature_dict_def.get(schema.RULES_KEY, []) feature_default_value = feature_dict_def.get(schema.FEATURE_DEFAULT_VAL_KEY) if feature_default_value and not rules_list: self._logger.debug( f"feature is enabled by default and has no defined rules, feature_name={feature_name}" ) ret_list.append(feature_name) elif self._handle_rules( feature_name=feature_name, rules_context=rules_context, feature_default_value=feature_default_value, rules=rules_list, ): self._logger.debug(f"feature's calculated value is True, feature_name={feature_name}") ret_list.append(feature_name) return ret_list
Methods
def get_all_enabled_feature_toggles(self, *, rules_context: Union[Dict[str, Any], NoneType] = None) ‑> List[str]
-
Get all enabled feature toggles while also taking into account rule_context (when a feature has defined rules)
Parameters
rules_context
:Optional[Dict[str, Any]]
- dict of attributes that you would like to match the rules
against, can be
{'tenant_id: 'X', 'username':' 'Y', 'region': 'Z'}
etc.
Returns
List[str]
- a list of all features name that are enabled by also taking into account rule_context (when a feature has defined rules)
Expand source code
def get_all_enabled_feature_toggles(self, *, rules_context: Optional[Dict[str, Any]] = None) -> List[str]: """Get all enabled feature toggles while also taking into account rule_context (when a feature has defined rules) Parameters ---------- rules_context: Optional[Dict[str, Any]] dict of attributes that you would like to match the rules against, can be `{'tenant_id: 'X', 'username':' 'Y', 'region': 'Z'}` etc. Returns ---------- List[str] a list of all features name that are enabled by also taking into account rule_context (when a feature has defined rules) """ if rules_context is None: rules_context = {} try: toggles_dict: Dict[str, Any] = self.get_configuration() except ConfigurationError: logger.error("unable to get feature toggles JSON") return [] ret_list = [] features: Dict[str, Any] = toggles_dict.get(schema.FEATURES_KEY, {}) for feature_name, feature_dict_def in features.items(): rules_list = feature_dict_def.get(schema.RULES_KEY, []) feature_default_value = feature_dict_def.get(schema.FEATURE_DEFAULT_VAL_KEY) if feature_default_value and not rules_list: self._logger.debug( f"feature is enabled by default and has no defined rules, feature_name={feature_name}" ) ret_list.append(feature_name) elif self._handle_rules( feature_name=feature_name, rules_context=rules_context, feature_default_value=feature_default_value, rules=rules_list, ): self._logger.debug(f"feature's calculated value is True, feature_name={feature_name}") ret_list.append(feature_name) return ret_list
def get_configuration(self) ‑> Dict[str, Any]
-
Get configuration string from AWs AppConfig and returned 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_configuration(self) -> Dict[str, Any]: """Get configuration string from AWs AppConfig and returned the parsed JSON dictionary Raises ------ ConfigurationError Any validation error or appconfig error that can occur Returns ------ Dict[str, Any] parsed JSON dictionary """ # parse result conf as JSON, keep in cache for self.max_age seconds config = self._schema_fetcher.get_json_configuration() # validate schema self._schema_validator.validate_json_schema(config) return config
def get_feature_toggle(self, *, feature_name: str, rules_context: Union[Dict[str, Any], NoneType] = None, value_if_missing: bool) ‑> bool
-
Get a feature toggle boolean value. Value is calculated according to a set of rules and conditions.
See below for explanation.
Parameters
feature_name
:str
- feature name that you wish to fetch
rules_context
:Optional[Dict[str, Any]]
- dict of attributes that you would like to match the rules against, can be {'tenant_id: 'X', 'username':' 'Y', 'region': 'Z'} etc.
value_if_missing
:bool
- this will be the returned value in case the feature toggle doesn't exist in the schema or there has been an error while fetching the configuration from appconfig
Returns
bool
- calculated feature toggle value. several possibilities: 1. if the feature doesn't appear in the schema or there has been an error fetching the configuration -> error/warning log would appear and value_if_missing is returned 2. feature exists and has no rules or no rules have matched -> return feature_default_value of the defined feature 3. feature exists and a rule matches -> rule_default_value of rule is returned
Expand source code
def get_feature_toggle( self, *, feature_name: str, rules_context: Optional[Dict[str, Any]] = None, value_if_missing: bool ) -> bool: """Get a feature toggle boolean value. Value is calculated according to a set of rules and conditions. See below for explanation. Parameters ---------- feature_name: str feature name that you wish to fetch rules_context: Optional[Dict[str, Any]] dict of attributes that you would like to match the rules against, can be {'tenant_id: 'X', 'username':' 'Y', 'region': 'Z'} etc. value_if_missing: bool this will be the returned value in case the feature toggle doesn't exist in the schema or there has been an error while fetching the configuration from appconfig Returns ------ bool calculated feature toggle value. several possibilities: 1. if the feature doesn't appear in the schema or there has been an error fetching the configuration -> error/warning log would appear and value_if_missing is returned 2. feature exists and has no rules or no rules have matched -> return feature_default_value of the defined feature 3. feature exists and a rule matches -> rule_default_value of rule is returned """ if rules_context is None: rules_context = {} try: toggles_dict: Dict[str, Any] = self.get_configuration() except ConfigurationError: logger.error("unable to get feature toggles JSON, returning provided value_if_missing value") return value_if_missing feature: Dict[str, Dict] = toggles_dict.get(schema.FEATURES_KEY, {}).get(feature_name, None) if feature is None: logger.warning( f"feature does not appear in configuration, using provided value_if_missing, " f"feature_name={feature_name}, value_if_missing={value_if_missing}" ) return value_if_missing rules_list = feature.get(schema.RULES_KEY) feature_default_value = feature.get(schema.FEATURE_DEFAULT_VAL_KEY) if not rules_list: # not rules but has a value logger.debug( f"no rules found, returning feature default value, feature_name={feature_name}, " f"default_value={feature_default_value}" ) return bool(feature_default_value) # look for first rule match logger.debug( f"looking for rule match, feature_name={feature_name}, feature_default_value={feature_default_value}" ) return self._handle_rules( feature_name=feature_name, rules_context=rules_context, feature_default_value=bool(feature_default_value), rules=cast(List, rules_list), )