diff --git a/aws_lambda_powertools/utilities/feature_flags/appconfig.py b/aws_lambda_powertools/utilities/feature_flags/appconfig.py index aa705c477c9..e0668da8390 100644 --- a/aws_lambda_powertools/utilities/feature_flags/appconfig.py +++ b/aws_lambda_powertools/utilities/feature_flags/appconfig.py @@ -1,10 +1,9 @@ +from __future__ import annotations + import logging import traceback -from typing import Any, Dict, Optional, Union, cast - -from botocore.config import Config +from typing import TYPE_CHECKING, Any, cast -from aws_lambda_powertools.logging import Logger from aws_lambda_powertools.utilities import jmespath_utils from aws_lambda_powertools.utilities.feature_flags.base import StoreProvider from aws_lambda_powertools.utilities.feature_flags.exceptions import ConfigurationStoreError, StoreClientError @@ -14,6 +13,11 @@ TransformParameterError, ) +if TYPE_CHECKING: + from botocore.config import Config + + from aws_lambda_powertools.logging import Logger + class AppConfigStore(StoreProvider): def __init__( @@ -22,10 +26,10 @@ def __init__( application: str, name: str, max_age: int = 5, - sdk_config: Optional[Config] = None, - envelope: Optional[str] = "", - jmespath_options: Optional[Dict] = None, - logger: Optional[Union[logging.Logger, Logger]] = None, + sdk_config: Config | None = None, + envelope: str | None = "", + jmespath_options: dict | None = None, + logger: logging.Logger | Logger | None = None, ): """This class fetches JSON schemas from AWS AppConfig @@ -39,11 +43,11 @@ def __init__( 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] + sdk_config: Config | None Botocore Config object to pass during client initialization - envelope : Optional[str] + envelope : str | None JMESPath expression to pluck feature flags data from config - jmespath_options : Optional[Dict] + 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. @@ -60,7 +64,7 @@ def __init__( self._conf_store = AppConfigProvider(environment=environment, application=application, boto_config=sdk_config) @property - def get_raw_configuration(self) -> Dict[str, Any]: + 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 @@ -82,7 +86,7 @@ def get_raw_configuration(self) -> Dict[str, Any]: raise StoreClientError(err_msg) from exc raise ConfigurationStoreError("Unable to get AWS AppConfig configuration file") from exc - def get_configuration(self) -> Dict[str, Any]: + 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, @@ -95,7 +99,7 @@ def get_configuration(self) -> Dict[str, Any]: Returns ------- - Dict[str, Any] + dict[str, Any] parsed JSON dictionary """ config = self.get_raw_configuration diff --git a/aws_lambda_powertools/utilities/feature_flags/base.py b/aws_lambda_powertools/utilities/feature_flags/base.py index e323f32d8b1..cd2d65fa211 100644 --- a/aws_lambda_powertools/utilities/feature_flags/base.py +++ b/aws_lambda_powertools/utilities/feature_flags/base.py @@ -1,16 +1,18 @@ +from __future__ import annotations + from abc import ABC, abstractmethod -from typing import Any, Dict +from typing import Any class StoreProvider(ABC): @property @abstractmethod - def get_raw_configuration(self) -> Dict[str, Any]: + def get_raw_configuration(self) -> dict[str, Any]: """Get configuration from any store and return the parsed JSON dictionary""" raise NotImplementedError() # pragma: no cover @abstractmethod - def get_configuration(self) -> Dict[str, Any]: + def get_configuration(self) -> dict[str, Any]: """Get configuration from any store and return the parsed JSON dictionary If envelope is set, it'll extract and return feature flags from configuration, @@ -23,7 +25,7 @@ def get_configuration(self) -> Dict[str, Any]: Returns ------- - Dict[str, Any] + dict[str, Any] parsed JSON dictionary **Example** diff --git a/aws_lambda_powertools/utilities/feature_flags/comparators.py b/aws_lambda_powertools/utilities/feature_flags/comparators.py index 035566cad4c..47354f26e73 100644 --- a/aws_lambda_powertools/utilities/feature_flags/comparators.py +++ b/aws_lambda_powertools/utilities/feature_flags/comparators.py @@ -1,14 +1,15 @@ from __future__ import annotations from datetime import datetime, tzinfo -from typing import Any, Dict, Optional +from typing import Any from dateutil.tz import gettz -from aws_lambda_powertools.utilities.feature_flags.schema import HOUR_MIN_SEPARATOR, ModuloRangeValues, TimeValues +from aws_lambda_powertools.utilities.feature_flags.constants import HOUR_MIN_SEPARATOR +from aws_lambda_powertools.utilities.feature_flags.schema import ModuloRangeValues, TimeValues -def _get_now_from_timezone(timezone: Optional[tzinfo]) -> datetime: +def _get_now_from_timezone(timezone: tzinfo | None) -> datetime: """ Returns now in the specified timezone. Defaults to UTC if not present. At this stage, we already validated that the passed timezone string is valid, so we assume that @@ -18,7 +19,7 @@ def _get_now_from_timezone(timezone: Optional[tzinfo]) -> datetime: return datetime.now(timezone) -def compare_days_of_week(context_value: Any, condition_value: Dict) -> bool: +def compare_days_of_week(context_value: Any, condition_value: dict) -> bool: timezone_name = condition_value.get(TimeValues.TIMEZONE.value, "UTC") # %A = Weekday as locale’s full name. @@ -28,7 +29,7 @@ def compare_days_of_week(context_value: Any, condition_value: Dict) -> bool: return current_day in days -def compare_datetime_range(context_value: Any, condition_value: Dict) -> bool: +def compare_datetime_range(context_value: Any, condition_value: dict) -> bool: timezone_name = condition_value.get(TimeValues.TIMEZONE.value, "UTC") timezone = gettz(timezone_name) current_time: datetime = _get_now_from_timezone(timezone) @@ -44,7 +45,7 @@ def compare_datetime_range(context_value: Any, condition_value: Dict) -> bool: return start_date <= current_time <= end_date -def compare_time_range(context_value: Any, condition_value: Dict) -> bool: +def compare_time_range(context_value: Any, condition_value: dict) -> bool: timezone_name = condition_value.get(TimeValues.TIMEZONE.value, "UTC") current_time: datetime = _get_now_from_timezone(gettz(timezone_name)) @@ -75,7 +76,7 @@ def compare_time_range(context_value: Any, condition_value: Dict) -> bool: return start_time <= current_time <= end_time -def compare_modulo_range(context_value: int, condition_value: Dict) -> bool: +def compare_modulo_range(context_value: int, condition_value: dict) -> bool: """ Returns for a given context 'a' and modulo condition 'b' -> b.start <= a % b.base <= b.end """ diff --git a/aws_lambda_powertools/utilities/feature_flags/constants.py b/aws_lambda_powertools/utilities/feature_flags/constants.py new file mode 100644 index 00000000000..dcb3a7419e5 --- /dev/null +++ b/aws_lambda_powertools/utilities/feature_flags/constants.py @@ -0,0 +1,13 @@ +import re + +RULES_KEY = "rules" +FEATURE_DEFAULT_VAL_KEY = "default" +CONDITIONS_KEY = "conditions" +RULE_MATCH_VALUE = "when_match" +CONDITION_KEY = "key" +CONDITION_VALUE = "value" +CONDITION_ACTION = "action" +FEATURE_DEFAULT_VAL_TYPE_KEY = "boolean_type" +TIME_RANGE_FORMAT = "%H:%M" # hour:min 24 hours clock +TIME_RANGE_PATTERN = re.compile(r"2[0-3]:[0-5]\d|[0-1]\d:[0-5]\d") # 24 hour clock +HOUR_MIN_SEPARATOR = ":" diff --git a/aws_lambda_powertools/utilities/feature_flags/exceptions.py b/aws_lambda_powertools/utilities/feature_flags/exceptions.py index eaea6c61cca..33e305270aa 100644 --- a/aws_lambda_powertools/utilities/feature_flags/exceptions.py +++ b/aws_lambda_powertools/utilities/feature_flags/exceptions.py @@ -7,7 +7,7 @@ class SchemaValidationError(Exception): class StoreClientError(Exception): - """When a store raises an exception that should be propagated to the client to fix + """When a store raises an exception that should be propagated to the client For example, Access Denied errors when the client doesn't permissions to fetch config """ diff --git a/aws_lambda_powertools/utilities/feature_flags/feature_flags.py b/aws_lambda_powertools/utilities/feature_flags/feature_flags.py index 2b93887138c..5ef21aff2f3 100644 --- a/aws_lambda_powertools/utilities/feature_flags/feature_flags.py +++ b/aws_lambda_powertools/utilities/feature_flags/feature_flags.py @@ -1,13 +1,9 @@ from __future__ import annotations import logging -from typing import Any, Callable, Dict, List, Optional, TypeVar, Union, cast +from typing import TYPE_CHECKING, Any, Callable, List, cast -from typing_extensions import ParamSpec - -from aws_lambda_powertools.logging import Logger from aws_lambda_powertools.utilities.feature_flags import schema -from aws_lambda_powertools.utilities.feature_flags.base import StoreProvider from aws_lambda_powertools.utilities.feature_flags.comparators import ( compare_all_in_list, compare_any_in_list, @@ -18,10 +14,13 @@ compare_time_range, ) from aws_lambda_powertools.utilities.feature_flags.exceptions import ConfigurationStoreError -from aws_lambda_powertools.utilities.feature_flags.types import JSONType +from aws_lambda_powertools.utilities.feature_flags.types import P, T + +if TYPE_CHECKING: + from aws_lambda_powertools.logging import Logger + from aws_lambda_powertools.utilities.feature_flags.base import StoreProvider + from aws_lambda_powertools.utilities.feature_flags.types import JSONType -T = TypeVar("T") -P = ParamSpec("P") RULE_ACTION_MAPPING = { schema.RuleAction.EQUALS.value: lambda a, b: a == b, @@ -49,7 +48,7 @@ class FeatureFlags: - def __init__(self, store: StoreProvider, logger: Optional[Union[logging.Logger, Logger]] = None): + def __init__(self, store: StoreProvider, logger: logging.Logger | Logger | None = None): """Evaluates whether feature flags should be enabled based on a given context. It uses the provided store to fetch feature flag rules before evaluating them. @@ -100,12 +99,12 @@ def _evaluate_conditions( self, rule_name: str, feature_name: str, - rule: Dict[str, Any], - context: Dict[str, Any], + rule: dict[str, Any], + context: dict[str, Any], ) -> bool: """Evaluates whether context matches conditions, return False otherwise""" rule_match_value = rule.get(schema.RULE_MATCH_VALUE) - conditions = cast(List[Dict], rule.get(schema.CONDITIONS_KEY)) + conditions = cast(List[dict], rule.get(schema.CONDITIONS_KEY)) if not conditions: self.logger.debug( @@ -141,9 +140,9 @@ def _evaluate_rules( self, *, feature_name: str, - context: Dict[str, Any], + context: dict[str, Any], feat_default: Any, - rules: Dict[str, Any], + rules: dict[str, Any], boolean_feature: bool, ) -> bool: """Evaluates whether context matches rules and conditions, otherwise return feature default""" @@ -164,7 +163,7 @@ def _evaluate_rules( ) return feat_default - def get_configuration(self) -> Dict: + def get_configuration(self) -> dict: """Get validated feature flag schema from configured store. Largely used to aid testing, since it's called by `evaluate` and `get_enabled_features` methods. @@ -178,7 +177,7 @@ def get_configuration(self) -> Dict: Returns ------ - Dict[str, Dict] + dict[str, dict] parsed JSON dictionary **Example** @@ -208,13 +207,13 @@ def get_configuration(self) -> Dict: """ # parse result conf as JSON, keep in cache for max age defined in store self.logger.debug(f"Fetching schema from registered store, store={self.store}") - config: Dict = self.store.get_configuration() + config: dict = self.store.get_configuration() validator = schema.SchemaValidator(schema=config, logger=self.logger) validator.validate() return config - def evaluate(self, *, name: str, context: Optional[Dict[str, Any]] = None, default: JSONType) -> JSONType: + def evaluate(self, *, name: str, context: dict[str, Any] | None = None, default: JSONType) -> JSONType: """Evaluate whether a feature flag should be enabled according to stored schema and input context **Logic when evaluating a feature flag** @@ -243,7 +242,7 @@ def evaluate(self, *, name: str, context: Optional[Dict[str, Any]] = None, defau ---------- name: str feature name to evaluate - context: Optional[Dict[str, Any]] + context: dict[str, Any] | None Attributes that should be evaluated against the stored schema. for example: `{"tenant_id": "X", "username": "Y", "region": "Z"}` @@ -306,7 +305,7 @@ def lambda_handler(event: dict, context: LambdaContext): # Maintenance: Revisit before going GA. We might to simplify customers on-boarding by not requiring it # for non-boolean flags. It'll need minor implementation changes, docs changes, and maybe refactor # get_enabled_features. We can minimize breaking change, despite Beta label, by having a new - # method `get_matching_features` returning Dict[feature_name, feature_value] + # method `get_matching_features` returning dict[feature_name, feature_value] boolean_feature = feature.get( schema.FEATURE_DEFAULT_VAL_TYPE_KEY, True, @@ -330,19 +329,19 @@ def lambda_handler(event: dict, context: LambdaContext): boolean_feature=boolean_feature, ) - def get_enabled_features(self, *, context: Optional[Dict[str, Any]] = None) -> List[str]: + def get_enabled_features(self, *, context: dict[str, Any] | None = None) -> list[str]: """Get all enabled feature flags while also taking into account context (when a feature has defined rules) Parameters ---------- - context: Optional[Dict[str, Any]] + context: dict[str, Any] | None 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] + list[str] list of all feature names that either matches context or have True as default **Example** @@ -359,10 +358,10 @@ def get_enabled_features(self, *, context: Optional[Dict[str, Any]] = None) -> L if context is None: context = {} - features_enabled: List[str] = [] + features_enabled: list[str] = [] try: - features: Dict[str, Any] = self.get_configuration() + features: dict[str, Any] = self.get_configuration() except ConfigurationStoreError as err: self.logger.debug(f"Failed to fetch feature flags from store, returning empty list, reason={err}") return features_enabled diff --git a/aws_lambda_powertools/utilities/feature_flags/schema.py b/aws_lambda_powertools/utilities/feature_flags/schema.py index 0fdb5e47810..a8739d5eb05 100644 --- a/aws_lambda_powertools/utilities/feature_flags/schema.py +++ b/aws_lambda_powertools/utilities/feature_flags/schema.py @@ -1,29 +1,31 @@ from __future__ import annotations import logging -import re from datetime import datetime from enum import Enum from functools import lru_cache -from typing import Any, Dict, List, Optional, Union +from typing import TYPE_CHECKING, Any from dateutil import tz -from aws_lambda_powertools.logging import Logger from aws_lambda_powertools.utilities.feature_flags.base import BaseValidator from aws_lambda_powertools.utilities.feature_flags.exceptions import SchemaValidationError -RULES_KEY = "rules" -FEATURE_DEFAULT_VAL_KEY = "default" -CONDITIONS_KEY = "conditions" -RULE_MATCH_VALUE = "when_match" -CONDITION_KEY = "key" -CONDITION_VALUE = "value" -CONDITION_ACTION = "action" -FEATURE_DEFAULT_VAL_TYPE_KEY = "boolean_type" -TIME_RANGE_FORMAT = "%H:%M" # hour:min 24 hours clock -TIME_RANGE_PATTERN = re.compile(r"2[0-3]:[0-5]\d|[0-1]\d:[0-5]\d") # 24 hour clock -HOUR_MIN_SEPARATOR = ":" +if TYPE_CHECKING: + from aws_lambda_powertools.logging import Logger + +from aws_lambda_powertools.utilities.feature_flags.constants import ( + CONDITION_ACTION, + CONDITION_KEY, + CONDITION_VALUE, + CONDITIONS_KEY, + FEATURE_DEFAULT_VAL_KEY, + FEATURE_DEFAULT_VAL_TYPE_KEY, + RULE_MATCH_VALUE, + RULES_KEY, + TIME_RANGE_FORMAT, + TIME_RANGE_PATTERN, +) LOGGER: logging.Logger | Logger = logging.getLogger(__name__) @@ -111,11 +113,11 @@ class SchemaValidator(BaseValidator): A dictionary containing default value and rules for matching. The value MUST be an object and MIGHT contain the following members: - * **default**: `Union[bool, JSONType]`. Defines default feature value. This MUST be present + * **default**: `bool | JSONType`. Defines default feature value. This MUST be present * **boolean_type**: bool. Defines whether feature has non-boolean value (`JSONType`). This MIGHT be present - * **rules**: `Dict[str, Dict]`. Rules object. This MIGHT be present + * **rules**: `dict[str, dict]`. Rules object. This MIGHT be present - `JSONType` being any JSON primitive value: `Union[str, int, float, bool, None, Dict[str, Any], List[Any]]` + `JSONType` being any JSON primitive value: `str | int | float | bool | None | dict[str, Any] | list[Any]` ```json { @@ -136,8 +138,8 @@ class SchemaValidator(BaseValidator): A dictionary with each rule and their conditions that a feature might have. The value MIGHT be present, and when defined it MUST contain the following members: - * **when_match**: `Union[bool, JSONType]`. Defines value to return when context matches conditions - * **conditions**: `List[Dict]`. Conditions object. This MUST be present + * **when_match**: `bool | JSONType`. Defines value to return when context matches conditions + * **conditions**: `list[dict]`. Conditions object. This MUST be present ```json { @@ -196,7 +198,7 @@ class SchemaValidator(BaseValidator): ``` """ - def __init__(self, schema: Dict[str, Any], logger: Optional[Union[logging.Logger, Logger]] = None): + def __init__(self, schema: dict[str, Any], logger: logging.Logger | Logger | None = None): self.schema = schema self.logger = logger or LOGGER @@ -222,7 +224,7 @@ def _link_global_logger(logger: logging.Logger | Logger): class FeaturesValidator(BaseValidator): """Validates each feature and calls RulesValidator to validate its rules""" - def __init__(self, schema: Dict, logger: Optional[Union[logging.Logger, Logger]] = None): + def __init__(self, schema: dict, logger: logging.Logger | Logger | None = None): self.schema = schema self.logger = logger or LOGGER @@ -255,13 +257,13 @@ class RulesValidator(BaseValidator): def __init__( self, - feature: Dict[str, Any], + feature: dict[str, Any], boolean_feature: bool, - logger: Optional[Union[logging.Logger, Logger]] = None, + logger: logging.Logger | Logger | None = None, ): self.feature = feature self.feature_name = next(iter(self.feature)) - self.rules: Optional[Dict] = self.feature.get(RULES_KEY) + self.rules: dict | None = self.feature.get(RULES_KEY) self.logger = logger or LOGGER self.boolean_feature = boolean_feature @@ -286,7 +288,7 @@ def validate(self): conditions.validate() @staticmethod - def validate_rule(rule: Dict, rule_name: str, feature_name: str, boolean_feature: bool = True): + def validate_rule(rule: dict, rule_name: str, feature_name: str, boolean_feature: bool = True): if not rule or not isinstance(rule, dict): raise SchemaValidationError(f"Feature rule must be a dictionary, feature={feature_name}") @@ -299,15 +301,15 @@ def validate_rule_name(rule_name: str, feature_name: str): raise SchemaValidationError(f"Rule name key must have a non-empty string, feature={feature_name}") @staticmethod - def validate_rule_default_value(rule: Dict, rule_name: str, boolean_feature: bool): + def validate_rule_default_value(rule: dict, rule_name: str, boolean_feature: bool): rule_default_value = rule.get(RULE_MATCH_VALUE) if boolean_feature and not isinstance(rule_default_value, bool): raise SchemaValidationError(f"'rule_default_value' key must have be bool, rule={rule_name}") class ConditionsValidator(BaseValidator): - def __init__(self, rule: Dict[str, Any], rule_name: str, logger: Optional[Union[logging.Logger, Logger]] = None): - self.conditions: List[Dict[str, Any]] = rule.get(CONDITIONS_KEY, {}) + def __init__(self, rule: dict[str, Any], rule_name: str, logger: logging.Logger | Logger | None = None): + self.conditions: list[dict[str, Any]] = rule.get(CONDITIONS_KEY, {}) self.rule_name = rule_name self.logger = logger or LOGGER @@ -322,7 +324,7 @@ def validate(self): self.validate_condition(rule_name=self.rule_name, condition=condition) @staticmethod - def validate_condition(rule_name: str, condition: Dict[str, str]) -> None: + def validate_condition(rule_name: str, condition: dict[str, str]) -> None: if not condition or not isinstance(condition, dict): raise SchemaValidationError(f"Feature rule condition must be a dictionary, rule={rule_name}") @@ -331,7 +333,7 @@ def validate_condition(rule_name: str, condition: Dict[str, str]) -> None: ConditionsValidator.validate_condition_value(condition=condition, rule_name=rule_name) @staticmethod - def validate_condition_action(condition: Dict[str, Any], rule_name: str): + def validate_condition_action(condition: dict[str, Any], rule_name: str): action = condition.get(CONDITION_ACTION, "") if action not in RuleAction.__members__: allowed_values = [_action.value for _action in RuleAction] @@ -340,7 +342,7 @@ def validate_condition_action(condition: Dict[str, Any], rule_name: str): ) @staticmethod - def validate_condition_key(condition: Dict[str, Any], rule_name: str): + def validate_condition_key(condition: dict[str, Any], rule_name: str): key = condition.get(CONDITION_KEY, "") if not key or not isinstance(key, str): raise SchemaValidationError(f"'key' value must be a non empty string, rule={rule_name}") @@ -367,7 +369,7 @@ def validate_condition_key(condition: Dict[str, Any], rule_name: str): custom_validator(key, rule_name) @staticmethod - def validate_condition_value(condition: Dict[str, Any], rule_name: str): + def validate_condition_value(condition: dict[str, Any], rule_name: str): value = condition.get(CONDITION_VALUE) if value is None: raise SchemaValidationError(f"'value' key must not be null, rule={rule_name}") @@ -427,7 +429,7 @@ def _validate_schedule_between_time_range_key(key: str, rule_name: str): ) @staticmethod - def _validate_schedule_between_time_range_value(value: Dict, rule_name: str): + def _validate_schedule_between_time_range_value(value: dict, rule_name: str): if not isinstance(value, dict): raise SchemaValidationError( f"{RuleAction.SCHEDULE_BETWEEN_TIME_RANGE.value} action must have a dictionary with 'START' and 'END' keys, rule={rule_name}", # noqa: E501 diff --git a/aws_lambda_powertools/utilities/feature_flags/types.py b/aws_lambda_powertools/utilities/feature_flags/types.py index 6df79e5d608..fa6c763ead9 100644 --- a/aws_lambda_powertools/utilities/feature_flags/types.py +++ b/aws_lambda_powertools/utilities/feature_flags/types.py @@ -1,4 +1,8 @@ -from typing import Any, Dict, List, Union +from typing import Any, Dict, List, TypeVar, Union + +from typing_extensions import ParamSpec # JSON primitives only, mypy doesn't support recursive tho JSONType = Union[str, int, float, bool, None, Dict[str, Any], List[Any]] +T = TypeVar("T") +P = ParamSpec("P")