Feature flags

CLASS	DESCRIPTION
`FeatureFlags`

FeatureFlags ¶

FeatureFlags(
    store: StoreProvider,
    logger: Logger | Logger | None = None,
)

It uses the provided store to fetch feature flag rules before evaluating them.

Examples:

from aws_lambda_powertools.utilities.feature_flags import FeatureFlags, AppConfigStore

app_config = AppConfigStore(
    environment="test",
    application="powertools",
    name="test_conf_name",
    max_age=300,
    envelope="features"
)

feature_flags: FeatureFlags = FeatureFlags(store=app_config)

PARAMETER	DESCRIPTION
`store`	Store to use to fetch feature flag schema configuration. TYPE: `StoreProvider`
`logger`	Used to log messages. If None is supplied, one will be created. TYPE: `Logger \| Logger \| None` DEFAULT: `None`

METHOD	DESCRIPTION
`evaluate`	Evaluate whether a feature flag should be enabled according to stored schema and input context
`get_configuration`	Get validated feature flag schema from configured store.
`get_enabled_features`	Get all enabled feature flags while also taking into account context
`validation_exception_handler`	Registers function to handle unexpected validation exceptions when evaluating flags.

Source code in aws_lambda_powertools/utilities/feature_flags/feature_flags.py

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.

    Examples
    --------

    ```python
    from aws_lambda_powertools.utilities.feature_flags import FeatureFlags, AppConfigStore

    app_config = AppConfigStore(
        environment="test",
        application="powertools",
        name="test_conf_name",
        max_age=300,
        envelope="features"
    )

    feature_flags: FeatureFlags = FeatureFlags(store=app_config)
    ```

    Parameters
    ----------
    store: StoreProvider
        Store to use to fetch feature flag schema configuration.
    logger: A logging object
        Used to log messages. If None is supplied, one will be created.
    """
    self.store = store
    self.logger = logger or logging.getLogger(__name__)
    self._exception_handlers: dict[Exception, Callable] = {}

evaluate ¶

evaluate(
    *,
    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

Feature exists and a rule matches, returns when_match value
Feature exists but has either no rules or no match, return feature default value
Feature doesn't exist in stored schema, encountered an error when fetching -> return default value provided

┌────────────────────────┐ ┌────────────────────────┐ ┌────────────────────────┐ │ Feature flags │──────▶ Get Configuration ├───────▶ Evaluate rules │ └────────────────────────┘ │ │ │ │ │┌──────────────────────┐│ │┌──────────────────────┐│ ││ Fetch schema ││ ││ Match rule ││ │└───────────┬──────────┘│ │└───────────┬──────────┘│ │ │ │ │ │ │ │┌───────────▼──────────┐│ │┌───────────▼──────────┐│ ││ Cache schema ││ ││ Match condition ││ │└───────────┬──────────┘│ │└───────────┬──────────┘│ │ │ │ │ │ │ │┌───────────▼──────────┐│ │┌───────────▼──────────┐│ ││ Validate schema ││ ││ Match action ││ │└──────────────────────┘│ │└──────────────────────┘│ └────────────────────────┘ └────────────────────────┘

PARAMETER	DESCRIPTION
`name`	feature name to evaluate TYPE: `str`
`context`	Attributes that should be evaluated against the stored schema. for example: `{"tenant_id": "X", "username": "Y", "region": "Z"}` TYPE: `dict[str, Any] \| None` DEFAULT: `None`
`default`	default value if feature flag doesn't exist in the schema, or there has been an error when fetching the configuration from the store Can be boolean or any JSON values for non-boolean features. TYPE: `JSONType`

Example

from aws_lambda_powertools.utilities.feature_flags import AppConfigStore, FeatureFlags
from aws_lambda_powertools.utilities.typing import LambdaContext

app_config = AppConfigStore(environment="dev", application="product-catalogue", name="features")

feature_flags = FeatureFlags(store=app_config)


def lambda_handler(event: dict, context: LambdaContext):
    # Get customer's tier from incoming request
    ctx = {"tier": event.get("tier", "standard")}

    # Evaluate whether customer's tier has access to premium features
    # based on `has_premium_features` rules
    has_premium_features: bool = feature_flags.evaluate(name="premium_features", context=ctx, default=False)
    if has_premium_features:
        # enable premium features
        ...

RETURNS	DESCRIPTION
`JSONType`	whether feature should be enabled (bool flags) or JSON value when non-bool feature matches

RAISES	DESCRIPTION
`SchemaValidationError`	When schema doesn't conform with feature flag schema

Source code in aws_lambda_powertools/utilities/feature_flags/feature_flags.py

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**

    1. Feature exists and a rule matches, returns when_match value
    2. Feature exists but has either no rules or no match, return feature default value
    3. Feature doesn't exist in stored schema, encountered an error when fetching -> return default value provided

    ┌────────────────────────┐      ┌────────────────────────┐       ┌────────────────────────┐
    │     Feature flags      │──────▶   Get Configuration    ├───────▶     Evaluate rules     │
    └────────────────────────┘      │                        │       │                        │
                                    │┌──────────────────────┐│       │┌──────────────────────┐│
                                    ││     Fetch schema     ││       ││      Match rule      ││
                                    │└───────────┬──────────┘│       │└───────────┬──────────┘│
                                    │            │           │       │            │           │
                                    │┌───────────▼──────────┐│       │┌───────────▼──────────┐│
                                    ││     Cache schema     ││       ││   Match condition    ││
                                    │└───────────┬──────────┘│       │└───────────┬──────────┘│
                                    │            │           │       │            │           │
                                    │┌───────────▼──────────┐│       │┌───────────▼──────────┐│
                                    ││   Validate schema    ││       ││     Match action     ││
                                    │└──────────────────────┘│       │└──────────────────────┘│
                                    └────────────────────────┘       └────────────────────────┘

    Parameters
    ----------
    name: str
        feature name to evaluate
    context: dict[str, Any] | None
        Attributes that should be evaluated against the stored schema.

        for example: `{"tenant_id": "X", "username": "Y", "region": "Z"}`
    default: JSONType
        default value if feature flag doesn't exist in the schema,
        or there has been an error when fetching the configuration from the store
        Can be boolean or any JSON values for non-boolean features.


    Example
    --------

    ```python
    from aws_lambda_powertools.utilities.feature_flags import AppConfigStore, FeatureFlags
    from aws_lambda_powertools.utilities.typing import LambdaContext

    app_config = AppConfigStore(environment="dev", application="product-catalogue", name="features")

    feature_flags = FeatureFlags(store=app_config)


    def lambda_handler(event: dict, context: LambdaContext):
        # Get customer's tier from incoming request
        ctx = {"tier": event.get("tier", "standard")}

        # Evaluate whether customer's tier has access to premium features
        # based on `has_premium_features` rules
        has_premium_features: bool = feature_flags.evaluate(name="premium_features", context=ctx, default=False)
        if has_premium_features:
            # enable premium features
            ...
    ```

    Returns
    ------
    JSONType
        whether feature should be enabled (bool flags) or JSON value when non-bool feature matches

    Raises
    ------
    SchemaValidationError
        When schema doesn't conform with feature flag schema
    """
    if context is None:
        context = {}

    try:
        features = self.get_configuration()
    except ConfigurationStoreError as err:
        self.logger.debug(f"Failed to fetch feature flags from store, returning default provided, reason={err}")
        return default

    feature = features.get(name)
    if feature is None:
        self.logger.debug(f"Feature not found; returning default provided, name={name}, default={default}")
        return default

    rules = feature.get(schema.RULES_KEY)
    feat_default = feature.get(schema.FEATURE_DEFAULT_VAL_KEY)
    # 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]
    boolean_feature = feature.get(
        schema.FEATURE_DEFAULT_VAL_TYPE_KEY,
        True,
    )  # backwards compatibility, assume feature flag
    if not rules:
        self.logger.debug(
            f"no rules found, returning feature default, name={name}, default={str(feat_default)}, boolean_feature={boolean_feature}",  # noqa: E501
        )
        # Maintenance: Revisit before going GA. We might to simplify customers on-boarding by not requiring it
        # for non-boolean flags.
        return bool(feat_default) if boolean_feature else feat_default

    self.logger.debug(
        f"looking for rule match, name={name}, default={str(feat_default)}, boolean_feature={boolean_feature}",  # noqa: E501
    )
    return self._evaluate_rules(
        feature_name=name,
        context=context,
        feat_default=feat_default,
        rules=rules,
        boolean_feature=boolean_feature,
    )

get_configuration ¶

get_configuration() -> 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.

RAISES	DESCRIPTION
`ConfigurationStoreError`	Any propagated error from store
`SchemaValidationError`	When schema doesn't conform with feature flag schema

RETURNS	DESCRIPTION
`dict[str, dict]`	parsed JSON dictionary

Example

{
    "premium_features": {
        "default": False,
        "rules": {
            "customer tier equals premium": {
                "when_match": True,
                "conditions": [
                    {
                        "action": "EQUALS",
                        "key": "tier",
                        "value": "premium",
                    }
                ],
            }
        },
    },
    "feature_two": {
        "default": False
    }
}

Source code in aws_lambda_powertools/utilities/feature_flags/feature_flags.py

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.

    Raises
    ------
    ConfigurationStoreError
        Any propagated error from store
    SchemaValidationError
        When schema doesn't conform with feature flag schema

    Returns
    ------
    dict[str, dict]
        parsed JSON dictionary

    Example
    -------

    ```python
    {
        "premium_features": {
            "default": False,
            "rules": {
                "customer tier equals premium": {
                    "when_match": True,
                    "conditions": [
                        {
                            "action": "EQUALS",
                            "key": "tier",
                            "value": "premium",
                        }
                    ],
                }
            },
        },
        "feature_two": {
            "default": False
        }
    }
    ```
    """
    # 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()
    validator = schema.SchemaValidator(schema=config, logger=self.logger)
    validator.validate()

    return config

get_enabled_features ¶

get_enabled_features(
    *, 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)

PARAMETER	DESCRIPTION
`context`	dict of attributes that you would like to match the rules against, can be `{'tenant_id: 'X', 'username':' 'Y', 'region': 'Z'}` etc. TYPE: `dict[str, Any] \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`list[str]`	list of all feature names that either matches context or have True as default

Example

["premium_features", "my_feature_two", "always_true_feature"]

RAISES	DESCRIPTION
`SchemaValidationError`	When schema doesn't conform with feature flag schema

Source code in aws_lambda_powertools/utilities/feature_flags/feature_flags.py

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: 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 of all feature names that either matches context or have True as default

    Example
    -------

    ```python
    ["premium_features", "my_feature_two", "always_true_feature"]
    ```

    Raises
    ------
    SchemaValidationError
        When schema doesn't conform with feature flag schema
    """
    if context is None:
        context = {}

    features_enabled: list[str] = []

    try:
        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

    self.logger.debug("Evaluating all features")
    for name, feature in features.items():
        rules = feature.get(schema.RULES_KEY, {})
        feature_default_value = feature.get(schema.FEATURE_DEFAULT_VAL_KEY)
        boolean_feature = feature.get(
            schema.FEATURE_DEFAULT_VAL_TYPE_KEY,
            True,
        )  # backwards compatibility, assume feature flag

        if feature_default_value and not rules:
            self.logger.debug(f"feature is enabled by default and has no defined rules, name={name}")
            features_enabled.append(name)
        elif self._evaluate_rules(
            feature_name=name,
            context=context,
            feat_default=feature_default_value,
            rules=rules,
            boolean_feature=boolean_feature,
        ):
            self.logger.debug(f"feature's calculated value is True, name={name}")
            features_enabled.append(name)

    return features_enabled

validation_exception_handler ¶

validation_exception_handler(
    exc_class: Exception | list[Exception],
)

Registers function to handle unexpected validation exceptions when evaluating flags.

It does not override the function of a default flag value in case of network and IAM permissions. For example, you won't be able to catch ConfigurationStoreError exception.

PARAMETER	DESCRIPTION
`exc_class`	One or more exceptions to catch TYPE: `Exception \| list[Exception]`

Example

feature_flags = FeatureFlags(store=app_config)

@feature_flags.validation_exception_handler(Exception)  # any exception
def catch_exception(exc):
    raise TypeError("re-raised") from exc

Source code in aws_lambda_powertools/utilities/feature_flags/feature_flags.py

def validation_exception_handler(self, exc_class: Exception | list[Exception]):
    """Registers function to handle unexpected validation exceptions when evaluating flags.

    It does not override the function of a default flag value in case of network and IAM permissions.
    For example, you won't be able to catch ConfigurationStoreError exception.

    Parameters
    ----------
    exc_class : Exception | list[Exception]
        One or more exceptions to catch

    Example
    -------

    ```python
    feature_flags = FeatureFlags(store=app_config)

    @feature_flags.validation_exception_handler(Exception)  # any exception
    def catch_exception(exc):
        raise TypeError("re-raised") from exc
    ```
    """

    def register_exception_handler(func: Callable[P, T]) -> Callable[P, T]:
        if isinstance(exc_class, list):
            for exp in exc_class:
                self._exception_handlers[exp] = func
        else:
            self._exception_handlers[exc_class] = func

        return func

    return register_exception_handler