Home - News ( United States | United Kingdom | Italy | Germany ) - Football scores

Showing content from https://docs.powertools.aws.dev/lambda/python/latest/utilities/feature_flags/ below:

Feature flags - Powertools for AWS Lambda (Python)

The feature flags utility provides a simple rule engine to define when one or multiple features should be enabled depending on the input.

When using AppConfigStore, we currently only support AppConfig using freeform configuration profile .

• Define simple feature flags to dynamically decide when to enable a feature
• Fetch one or all feature flags enabled for a given application context
• Support for static feature flags to simply turn on/off a feature without rules
• Support for time based feature flags
• Bring your own Feature Flags Store Provider

Feature flags are used to modify behaviour without changing the application's code. These flags can be static or dynamic.

Static flags. Indicates something is simply on or off, for example TRACER_ENABLED=True.

Dynamic flags. Indicates something can have varying states, for example enable a list of premium features for customer X not Y.

You can use Parameters utility for static flags while this utility can do both static and dynamic feature flags.

Be mindful that feature flags can increase the complexity of your application over time; use them sparingly.

If you want to learn more about feature flags, their variations and trade-offs, check these articles:

• Feature Toggles (aka Feature Flags) - Pete Hodgson
• AWS Lambda Feature Toggles Made Simple - Ran Isenberg
• Feature Flags Getting Started - CloudBees

AWS AppConfig requires two API calls to fetch configuration for the first time. You can improve latency by consolidating your feature settings in a single Configuration.

When using the default store AppConfigStore, your Lambda function IAM Role must have appconfig:GetLatestConfiguration and appconfig:StartConfigurationSession IAM permissions before using this feature.

By default, this utility provides AWS AppConfig as a configuration store.

The following sample infrastructure will be used throughout this documentation:

template.yamlCDK

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

AWSTemplateFormatVersion: "2010-09-09"
Description: Lambda Powertools for Python Feature flags sample template
Resources:
  FeatureStoreApp:
    Type: AWS::AppConfig::Application
    Properties:
      Description: "AppConfig Application for feature toggles"
      Name: product-catalogue

  FeatureStoreDevEnv:
    Type: AWS::AppConfig::Environment
    Properties:
      ApplicationId: !Ref FeatureStoreApp
      Description: "Development Environment for the App Config Store"
      Name: dev

  FeatureStoreConfigProfile:
    Type: AWS::AppConfig::ConfigurationProfile
    Properties:
      ApplicationId: !Ref FeatureStoreApp
      Name: features
      LocationUri: "hosted"

  HostedConfigVersion:
    Type: AWS::AppConfig::HostedConfigurationVersion
    Properties:
      ApplicationId: !Ref FeatureStoreApp
      ConfigurationProfileId: !Ref FeatureStoreConfigProfile
      Description: 'A sample hosted configuration version'
      Content: |
        {
              "premium_features": {
                "default": false,
                "rules": {
                  "customer tier equals premium": {
                    "when_match": true,
                    "conditions": [
                      {
                        "action": "EQUALS",
                        "key": "tier",
                        "value": "premium"
                      }
                    ]
                  }
                }
              },
              "ten_percent_off_campaign": {
                "default": false
              }
          }
      ContentType: 'application/json'

  ConfigDeployment:
    Type: AWS::AppConfig::Deployment
    Properties:
      ApplicationId: !Ref FeatureStoreApp
      ConfigurationProfileId: !Ref FeatureStoreConfigProfile
      ConfigurationVersion: !Ref HostedConfigVersion
      DeploymentStrategyId: "AppConfig.AllAtOnce"
      EnvironmentId: !Ref FeatureStoreDevEnv

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

import json

import aws_cdk.aws_appconfig as appconfig
from aws_cdk import core


class SampleFeatureFlagStore(core.Construct):
    def __init__(self, scope: core.Construct, id_: str) -> None:
        super().__init__(scope, id_)

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

        self.config_app = appconfig.CfnApplication(
            self,
            id="app",
            name="product-catalogue",
        )
        self.config_env = appconfig.CfnEnvironment(
            self,
            id="env",
            application_id=self.config_app.ref,
            name="dev-env",
        )
        self.config_profile = appconfig.CfnConfigurationProfile(
            self,
            id="profile",
            application_id=self.config_app.ref,
            location_uri="hosted",
            name="features",
        )
        self.hosted_cfg_version = appconfig.CfnHostedConfigurationVersion(
            self,
            "version",
            application_id=self.config_app.ref,
            configuration_profile_id=self.config_profile.ref,
            content=json.dumps(features_config),
            content_type="application/json",
        )
        self.app_config_deployment = appconfig.CfnDeployment(
            self,
            id="deploy",
            application_id=self.config_app.ref,
            configuration_profile_id=self.config_profile.ref,
            configuration_version=self.hosted_cfg_version.ref,
            deployment_strategy_id="AppConfig.AllAtOnce",
            environment_id=self.config_env.ref,
        )

To get started, you'd need to initialize AppConfigStore and FeatureFlags. Then call FeatureFlags evaluate method to fetch, validate, and evaluate your feature.

The evaluate method supports two optional parameters:

• context: Value to be evaluated against each rule defined for the given feature
• default: Sentinel value to use in case we experience any issues with our store, or feature doesn't exist

getting_started_single_feature_flag.pygetting_started_single_feature_flag_payload.jsongetting_started_single_feature_flag_features.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

from typing import Any

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):
    """
    This feature flag is enabled under the following conditions:
    - The request payload contains a field 'tier' with the value 'premium'.

    Rule condition to be evaluated:
        "conditions": [
            {
                "action": "EQUALS",
                "key": "tier",
                "value": "premium"
            }
        ]
    """

    # 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: Any = feature_flags.evaluate(name="premium_features", context=ctx, default=False)
    if has_premium_features:
        # enable premium features
        ...

{
    "username": "lessa",
    "tier": "premium",
    "basked_id": "random_id"
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

{
    "premium_features": {
        "default": false,
        "rules": {
            "customer tier equals premium": {
                "when_match": true,
                "conditions": [
                    {
                        "action": "EQUALS",
                        "key": "tier",
                        "value": "premium"
                    }
                ]
            }
        }
    },
    "ten_percent_off_campaign": {
        "default": false
    }
}

We have a static flag named ten_percent_off_campaign. Meaning, there are no conditional rules, it's either ON or OFF for all customers.

In this case, we could omit the context parameter and simply evaluate whether we should apply the 10% discount.

getting_started_static_flag.pygetting_started_static_flag_payload.jsongetting_started_static_flag_features.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from typing import Any

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):
    """
    This feature flag is enabled by default for all requests.
    """

    apply_discount: Any = feature_flags.evaluate(name="ten_percent_off_campaign", default=False)

    price: Any = event.get("price")

    if apply_discount:
        # apply 10% discount to product
        price = price * 0.9

    return {"price": price}

{
    "product": "laptop",
    "price": 1000
}

{
    "ten_percent_off_campaign": {
        "default": true
    }
}

As you might have noticed, each evaluate call means an API call to the Store and the more features you have the more costly this becomes.

You can use get_enabled_features method for scenarios where you need a list of all enabled features according to the input context.

getting_all_enabled_features.pygetting_all_enabled_features_payload.jsongetting_all_enabled_features_features.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

from __future__ import annotations

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

app = APIGatewayRestResolver()

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

feature_flags = FeatureFlags(store=app_config)


@app.get("/products")
def list_products():
    # getting fields from request
    # https://docs.powertools.aws.dev/lambda/python/latest/core/event_handler/api_gateway/#accessing-request-details
    json_body = app.current_event.json_body
    headers = app.current_event.headers

    ctx = {**headers, **json_body}

    # getting price from payload
    price: float = float(json_body.get("price"))
    percent_discount: int = 0

    # all_features is evaluated to ["premium_features", "geo_customer_campaign", "ten_percent_off_campaign"]
    all_features: list[str] = feature_flags.get_enabled_features(context=ctx)

    if "geo_customer_campaign" in all_features:
        # apply 20% discounts for customers in NL
        percent_discount += 20

    if "ten_percent_off_campaign" in all_features:
        # apply additional 10% for all customers
        percent_discount += 10

    price = price * (100 - percent_discount) / 100

    return {"price": price}


def lambda_handler(event: dict, context: LambdaContext):
    return app.resolve(event, context)

{
    "body": "{\"username\": \"lessa\", \"tier\": \"premium\", \"basked_id\": \"random_id\", \"price\": 1000}",
    "resource": "/products",
    "path": "/products",
    "httpMethod": "GET",
    "isBase64Encoded": false,
    "headers": {
        "CloudFront-Viewer-Country": "NL"
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

{
    "premium_features": {
      "default": false,
      "rules": {
        "customer tier equals premium": {
          "when_match": true,
          "conditions": [
            {
              "action": "EQUALS",
              "key": "tier",
              "value": "premium"
            }
          ]
        }
      }
    },
    "ten_percent_off_campaign": {
      "default": true
    },
    "geo_customer_campaign": {
      "default": false,
      "rules": {
        "customer in temporary discount geo": {
          "when_match": true,
          "conditions": [
            {
              "action": "KEY_IN_VALUE",
              "key": "CloudFront-Viewer-Country",
              "value": [
                "NL",
                "IE",
                "UK",
                "PL",
                "PT"
              ]
            }
          ]
        }
      }
    }
  }

Feature flags can also return enabled features based on time or datetime ranges. This allows you to have features that are only enabled on certain days of the week, certain time intervals or between certain calendar dates.

Use cases:

• Enable maintenance mode during a weekend
• Disable support/chat feature after working hours
• Launch a new feature on a specific date and time

You can also have features enabled only at certain times of the day for premium tier customers

timebased_feature.pytimebased_feature_event.jsontimebased_features.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

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):
    """
    This feature flag is enabled under the following conditions:
    - The request payload contains a field 'tier' with the value 'premium'.
    - If the current day is either Saturday or Sunday in America/New_York timezone.

    Rule condition to be evaluated:
        "conditions": [
          {
            "action": "EQUALS",
            "key": "tier",
            "value": "premium"
          },
          {
            "action": "SCHEDULE_BETWEEN_DAYS_OF_WEEK",
            "key": "CURRENT_DAY_OF_WEEK",
            "value": {
              "DAYS": [
                "SATURDAY",
                "SUNDAY"
              ],
              "TIMEZONE": "America/New_York"
            }
          }
        ]
    """

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

    # Checking if the weekend premum discount is enable
    weekend_premium_discount = feature_flags.evaluate(name="weekend_premium_discount", default=False, context=ctx)

    if weekend_premium_discount:
        # Enable special discount on weekend for premium users:
        return {"message": "The weekend premium discount is enabled."}

    return {"message": "The weekend premium discount is not enabled."}

{
  "username": "rubefons",
  "tier": "premium",
  "basked_id": "random_id"
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

{
  "weekend_premium_discount": {
    "default": false,
    "rules": {
      "customer tier equals premium and its time for a discount": {
        "when_match": true,
        "conditions": [
          {
            "action": "EQUALS",
            "key": "tier",
            "value": "premium"
          },
          {
            "action": "SCHEDULE_BETWEEN_DAYS_OF_WEEK",
            "key": "CURRENT_DAY_OF_WEEK",
            "value": {
              "DAYS": [
                "SATURDAY",
                "SUNDAY"
              ],
              "TIMEZONE": "America/New_York"
            }
          }
        ]
      }
    }
  }
}

You can also have features enabled only at certain times of the day.

timebased_happyhour_feature.pytimebased_happyhour_features.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

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):
    """
    This feature flag is enabled under the following conditions:
    - Every day between 17:00 to 19:00 in Europe/Copenhagen timezone

    Rule condition to be evaluated:
        "conditions": [
          {
            "action": "SCHEDULE_BETWEEN_TIME_RANGE",
            "key": "CURRENT_TIME",
            "value": {
              "START": "17:00",
              "END": "19:00",
              "TIMEZONE": "Europe/Copenhagen"
            }
          }
        ]
    """

    # Checking if the happy hour discount is enable
    is_happy_hour = feature_flags.evaluate(name="happy_hour", default=False)

    if is_happy_hour:
        # Enable special discount on happy hour:
        return {"message": "The happy hour discount is enabled."}

    return {"message": "The happy hour discount is not enabled."}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

{
  "happy_hour": {
    "default": false,
    "rules": {
      "is happy hour": {
        "when_match": true,
        "conditions": [
          {
            "action": "SCHEDULE_BETWEEN_TIME_RANGE",
            "key": "CURRENT_TIME",
            "value": {
              "START": "17:00",
              "END": "19:00",
              "TIMEZONE": "Europe/Copenhagen"
            }
          }
        ]
      }
    }
  }
}

You can also have features enabled only at specific days, for example: enable christmas sale discount during specific dates.

datetime_feature.pydatetime_features.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

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):
    """
    This feature flag is enabled under the following conditions:
    - Start date: December 25th, 2022 at 12:00:00 PM EST
    - End date: December 31st, 2022 at 11:59:59 PM EST
    - Timezone: America/New_York

    Rule condition to be evaluated:
        "conditions": [
          {
            "action": "SCHEDULE_BETWEEN_DATETIME_RANGE",
            "key": "CURRENT_DATETIME",
            "value": {
              "START": "2022-12-25T12:00:00",
              "END": "2022-12-31T23:59:59",
              "TIMEZONE": "America/New_York"
            }
          }
        ]
    """

    # Checking if the Christmas discount is enable
    xmas_discount = feature_flags.evaluate(name="christmas_discount", default=False)

    if xmas_discount:
        # Enable special discount on christmas:
        return {"message": "The Christmas discount is enabled."}

    return {"message": "The Christmas discount is not enabled."}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

{
  "christmas_discount": {
    "default": false,
    "rules": {
      "enable discount during christmas": {
        "when_match": true,
        "conditions": [
          {
            "action": "SCHEDULE_BETWEEN_DATETIME_RANGE",
            "key": "CURRENT_DATETIME",
            "value": {
              "START": "2022-12-25T12:00:00",
              "END": "2022-12-31T23:59:59",
              "TIMEZONE": "America/New_York"
            }
          }
        ]
      }
    }
  }
}

You can use any IANA time zone (as originally specified in PEP 615) as part of your rules definition. Powertools for AWS Lambda (Python) takes care of converting and calculate the correct timestamps for you.

When using SCHEDULE_BETWEEN_DATETIME_RANGE, use timestamps without timezone information, and specify the timezone manually. This way, you'll avoid hitting problems with day light savings.

Feature flags can also be used to run experiments on a segment of users based on modulo range conditions on context variables. This allows you to have features that are only enabled for a certain segment of users, comparing across multiple variants of the same experiment.

Use cases:

• Enable an experiment for a percentage of users
• Scale up an experiment incrementally in production - canary release
• Run multiple experiments or variants simultaneously by assigning a spectrum segment to each experiment variant.

The modulo range condition takes three values - BASE, START and END.

The condition evaluates START <= CONTEXT_VALUE % BASE <= END.

modulo_range_feature.pymodulo_range_feature_event.jsonmodulo_range_features.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

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):
    """
    This feature flag is enabled under the following conditions:
    - The request payload contains a field 'tier' with the value 'standard'.
    - If the user_id belongs to the spectrum 0-19 modulo 100, (20% users) on whom we want to run the sale experiment.

    Rule condition to be evaluated:
        "conditions": [
          {
            "action": "EQUALS",
            "key": "tier",
            "value": "standard"
          },
          {
            "action": "MODULO_RANGE",
            "key": "user_id",
            "value": {
              "BASE": 100,
              "START": 0,
              "END": 19
            }
          }
        ]
    """

    # Get customer's tier and identifier from incoming request
    ctx = {"tier": event.get("tier", "standard"), "user_id": event.get("user_id", 0)}

    # Checking if the sale_experiment is enable
    sale_experiment = feature_flags.evaluate(name="sale_experiment", default=False, context=ctx)

    if sale_experiment:
        # Enable special discount for sale experiment segment users:
        return {"message": "The sale experiment is enabled."}

    return {"message": "The sale experiment is not enabled."}

{
  "user_id": 134532511,
  "tier": "standard",
  "basked_id": "random_id"
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

{
  "sale_experiment": {
    "default": false,
    "rules": {
      "experiment 1 segment - 20% users": {
        "when_match": true,
        "conditions": [
          {
            "action": "EQUALS",
            "key": "tier",
            "value": "standard"
          },
          {
            "action": "MODULO_RANGE",
            "key": "user_id",
            "value": {
              "BASE": 100,
              "START": 0,
              "END": 19
            }
          }
        ]
      }
    }
  }
}

You can run multiple experiments on your users with the spectrum of your choice.

modulo_range_multiple_feature.pymodulo_range_multiple_features.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

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):
    """
    This non-boolean feature flag returns the percentage discount depending on the sale experiment segment:
    - 10% standard discount if the user_id belongs to the spectrum 0-3 modulo 10, (40% users).
    - 15% experiment discount if the user_id belongs to the spectrum 4-6 modulo 10, (30% users).
    - 18% experiment discount if the user_id belongs to the spectrum 7-9 modulo 10, (30% users).

    Rule conditions to be evaluated:
    "rules": {
      "control group - standard 10% discount segment": {
        "when_match": 10,
        "conditions": [
          {
            "action": "MODULO_RANGE",
            "key": "user_id",
            "value": {
              "BASE": 10,
              "START": 0,
              "END": 3
            }
          }
        ]
      },
      "test experiment 1 - 15% discount segment": {
        "when_match": 15,
        "conditions": [
          {
            "action": "MODULO_RANGE",
            "key": "user_id",
            "value": {
              "BASE": 10,
              "START": 4,
              "END": 6
            }
          }
        ]
      },
      "test experiment 2 - 18% discount segment": {
        "when_match": 18,
        "conditions": [
          {
            "action": "MODULO_RANGE",
            "key": "user_id",
            "value": {
              "BASE": 10,
              "START": 7,
              "END": 9
            }
          }
        ]
      }
    }
    """

    # Get customer's tier and identifier from incoming request
    ctx = {"tier": event.get("tier", "standard"), "user_id": event.get("user_id", 0)}

    # Get sale discount percentage from feature flag.
    sale_experiment_discount = feature_flags.evaluate(name="sale_experiment_discount", default=0, context=ctx)

    return {"message": f" {sale_experiment_discount}% discount applied."}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

{
  "sale_experiment_discount": {
    "boolean_type": false,
    "default": 0,
    "rules": {
      "control group - standard 10% discount segment": {
        "when_match": 10,
        "conditions": [
          {
            "action": "MODULO_RANGE",
            "key": "user_id",
            "value": {
              "BASE": 10,
              "START": 0,
              "END": 3
            }
          }
        ]
      },
      "test experiment 1 - 15% discount segment": {
        "when_match": 15,
        "conditions": [
          {
            "action": "MODULO_RANGE",
            "key": "user_id",
            "value": {
              "BASE": 10,
              "START": 4,
              "END": 6
            }
          }
        ]
      },
      "test experiment 2 - 18% discount segment": {
        "when_match": 18,
        "conditions": [
          {
            "action": "MODULO_RANGE",
            "key": "user_id",
            "value": {
              "BASE": 10,
              "START": 7,
              "END": 9
            }
          }
        ]
      }
    }
  }
}

You might have a list of features to unlock for premium customers, unlock a specific set of features for admin users, etc.

Feature flags can return any JSON values when boolean_type parameter is set to false. These can be dictionaries, list, string, integers, etc.

beyond_boolean.pybeyond_boolean_payload.jsonbeyond_boolean_features.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

from typing import Any

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

app_config = AppConfigStore(environment="dev", application="comments", name="config")

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 `has_premium_features` based on customer's tier
    premium_features: Any = feature_flags.evaluate(name="premium_features", context=ctx, default=[])

    return {"Premium features enabled": premium_features}

{
    "username": "lessa",
    "tier": "premium",
    "basked_id": "random_id"
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

{
    "premium_features": {
      "boolean_type": false,
      "default": [],
      "rules": {
        "customer tier equals premium": {
          "when_match": [
            "no_ads",
            "no_limits",
            "chat"
          ],
          "conditions": [
            {
              "action": "EQUALS",
              "key": "tier",
              "value": "premium"
            }
          ]
        }
      }
    }
  }

By default, we cache configuration retrieved from the Store for 5 seconds for performance and reliability reasons.

You can override max_age parameter when instantiating the store.

getting_started_with_cache.pygetting_started_with_cache_payload.jsongetting_started_with_cache_features.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from typing import Any

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", max_age=300)

feature_flags = FeatureFlags(store=app_config)


def lambda_handler(event: dict, context: LambdaContext):
    """
    This feature flag is enabled by default for all requests.
    """

    apply_discount: Any = feature_flags.evaluate(name="ten_percent_off_campaign", default=False)

    price: Any = event.get("price")

    if apply_discount:
        # apply 10% discount to product
        price = price * 0.9

    return {"price": price}

{
    "product": "laptop",
    "price": 1000
}

{
    "ten_percent_off_campaign": {
        "default": true
    }
}

You might have application configuration in addition to feature flags in your store.

This means you don't need to make another call only to fetch app configuration.

You can access the configuration fetched from the store via get_raw_configuration property within the store instance.

getting_stored_features.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from aws_lambda_powertools.utilities.feature_flags import AppConfigStore, FeatureFlags

app_config = AppConfigStore(
    environment="dev",
    application="product-catalogue",
    name="configuration",
    envelope="feature_flags",
)

feature_flags = FeatureFlags(store=app_config)

config = app_config.get_raw_configuration
...

This utility expects a certain schema to be stored as JSON within AWS AppConfig.

A feature can simply have its name and a default value. This is either on or off, also known as a static flag.

minimal_schema.json

{
    "global_feature": {
        "default": true
    },
    "non_boolean_global_feature": {
        "default": {"group": "read-only"},
        "boolean_type": false
    }
}

If you need more control and want to provide context such as user group, permissions, location, etc., you need to add rules to your feature flag configuration.

When adding rules to a feature, they must contain:

1. A rule name as a key
2. when_match boolean or JSON value that should be used when conditions match
3. A list of conditions for evaluation

feature_with_rules.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

{
    "premium_feature": {
        "default": false,
        "rules": {
            "customer tier equals premium": {
                "when_match": true,
                "conditions": [
                    {
                        "action": "EQUALS",
                        "key": "tier",
                        "value": "premium"
                    }
                ]
            }
        }
    },
    "non_boolean_premium_feature": {
        "default": [],
        "rules": {
            "customer tier equals premium": {
                "when_match": ["remove_limits", "remove_ads"],
                "conditions": [
                    {
                        "action": "EQUALS",
                        "key": "tier",
                        "value": "premium"
                    }
                ]
            }
        }
    }
}

You can have multiple rules with different names. The rule engine will return the first result when_match of the matching rule configuration, or default value when none of the rules apply.

The conditions block is a list of conditions that contain action, key, and value keys:

conditions.json

{
    "conditions": [
        {
            "action": "EQUALS",
            "key": "tier",
            "value": "premium"
        }
    ]
}

The action configuration can have the following values, where the expressions a is the key and b is the value above:

The key and value will be compared to the input from the context parameter.

For time based keys, we provide a list of predefined keys. These will automatically get converted to the corresponding timestamp on each invocation of your Lambda function.

If not specified, the timezone used for calculations will be UTC.

For multiple conditions, we will evaluate the list of conditions as a logical AND, so all conditions needs to match to return when_match value.

Now that you've seen all properties of a feature flag schema, this flowchart describes how the rule engine decides what value to return.

There are scenarios where you might want to include feature flags as part of an existing application configuration.

For this to work, you need to use a JMESPath expression via the envelope parameter to extract that key as the feature flags configuration.

extracting_envelope.pyextracting_envelope_payload.jsonextracting_envelope_features.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from typing import Any

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="feature_flags",
    envelope="features",
)

feature_flags = FeatureFlags(store=app_config)


def lambda_handler(event: dict, context: LambdaContext):
    apply_discount: Any = feature_flags.evaluate(name="ten_percent_off_campaign", default=False)

    price: Any = event.get("price")

    if apply_discount:
        # apply 10% discount to product
        price = price * 0.9

    return {"price": price}

{
    "product": "laptop",
    "price": 1000
}

{
    "logging": {
          "level": "INFO",
          "sampling_rate": 0.1
     },
    "features": {
      "ten_percent_off_campaign": {
        "default": true
      }
    }
  }

AppConfig store provider fetches any JSON document from AWS AppConfig.

These are the available options for further customization.

appconfig_provider_options.pyappconfig_provider_options_payload.jsonappconfig_provider_options_features.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

from typing import Any

from botocore.config import Config
from jmespath.functions import Functions, signature

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

boto_config = Config(read_timeout=10, retries={"total_max_attempts": 2})


# Custom JMESPath functions
class CustomFunctions(Functions):
    @signature({"types": ["object"]})
    def _func_special_decoder(self, features):
        # You can add some logic here
        return features


custom_jmespath_options = {"custom_functions": CustomFunctions()}


app_config = AppConfigStore(
    environment="dev",
    application="product-catalogue",
    name="features",
    max_age=120,
    envelope="special_decoder(features)",  # using a custom function defined in CustomFunctions Class
    boto_config=boto_config,
    jmespath_options=custom_jmespath_options,
)

feature_flags = FeatureFlags(store=app_config)


def lambda_handler(event: dict, context: LambdaContext):
    apply_discount: Any = feature_flags.evaluate(name="ten_percent_off_campaign", default=False)

    price: Any = event.get("price")

    if apply_discount:
        # apply 10% discount to product
        price = price * 0.9

    return {"price": price}

{
    "product": "laptop",
    "price": 1000
}

{
    "logging": {
          "level": "INFO",
          "sampling_rate": 0.1
     },
    "features": {
      "ten_percent_off_campaign": {
        "default": true
      }
    }
  }

The boto_config , boto3_session, and boto3_client parameters enable you to pass in a custom botocore config object, boto3 session, or a boto3 client when constructing the AppConfig store provider.

custom_boto_session_feature_flags.pycustom_boto_config_feature_flags.pycustom_boto_client_feature_flags.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

from typing import Any

import boto3

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

boto3_session = boto3.session.Session()

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

feature_flags = FeatureFlags(store=app_config)


def lambda_handler(event: dict, context: LambdaContext):
    apply_discount: Any = feature_flags.evaluate(name="ten_percent_off_campaign", default=False)

    price: Any = event.get("price")

    if apply_discount:
        # apply 10% discount to product
        price = price * 0.9

    return {"price": price}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

from typing import Any

from botocore.config import Config

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

boto_config = Config(read_timeout=10, retries={"total_max_attempts": 2})

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

feature_flags = FeatureFlags(store=app_config)


def lambda_handler(event: dict, context: LambdaContext):
    apply_discount: Any = feature_flags.evaluate(name="ten_percent_off_campaign", default=False)

    price: Any = event.get("price")

    if apply_discount:
        # apply 10% discount to product
        price = price * 0.9

    return {"price": price}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

from typing import Any

import boto3

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

boto3_client = boto3.client("appconfigdata")

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

feature_flags = FeatureFlags(store=app_config)


def lambda_handler(event: dict, context: LambdaContext):
    apply_discount: Any = feature_flags.evaluate(name="ten_percent_off_campaign", default=False)

    price: Any = event.get("price")

    if apply_discount:
        # apply 10% discount to product
        price = price * 0.9

    return {"price": price}

You can create your own custom FeatureFlags store provider by inheriting the StoreProvider class, and implementing both get_raw_configuration() and get_configuration() methods to retrieve the configuration from your custom store.

• get_raw_configuration() – get the raw configuration from the store provider and return the parsed JSON dictionary
• get_configuration() – get the configuration from the store provider, parsing it as a JSON dictionary. If an envelope is set, extract the envelope data

Here are an example of implementing a custom store provider using Amazon S3, a popular object storage.

This is just one example of how you can create your own store provider. Before creating a custom store provider, carefully evaluate your requirements and consider factors such as performance, scalability, and ease of maintenance.

working_with_own_s3_store_provider.pycustom_s3_store_provider.pyworking_with_own_s3_store_provider_payload.jsonworking_with_own_s3_store_provider_features.json

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from typing import Any

from custom_s3_store_provider import S3StoreProvider

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

s3_config_store = S3StoreProvider("your-bucket-name", "working_with_own_s3_store_provider_features.json")

feature_flags = FeatureFlags(store=s3_config_store)


def lambda_handler(event: dict, context: LambdaContext):
    apply_discount: Any = feature_flags.evaluate(name="ten_percent_off_campaign", default=False)

    price: Any = event.get("price")

    if apply_discount:
        # apply 10% discount to product
        price = price * 0.9

    return {"price": price}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

import json
from typing import Any, Dict

import boto3
from botocore.exceptions import ClientError

from aws_lambda_powertools.utilities.feature_flags.base import StoreProvider
from aws_lambda_powertools.utilities.feature_flags.exceptions import (
    ConfigurationStoreError,
)


class S3StoreProvider(StoreProvider):
    def __init__(self, bucket_name: str, object_key: str):
        # Initialize the client to your custom store provider

        super().__init__()

        self.bucket_name = bucket_name
        self.object_key = object_key
        self.client = boto3.client("s3")

    def _get_s3_object(self) -> Dict[str, Any]:
        # Retrieve the object content
        try:
            response = self.client.get_object(Bucket=self.bucket_name, Key=self.object_key)
            return json.loads(response["Body"].read().decode())
        except ClientError as exc:
            raise ConfigurationStoreError("Unable to get S3 Store Provider configuration file") from exc

    def get_configuration(self) -> Dict[str, Any]:
        return self._get_s3_object()

    @property
    def get_raw_configuration(self) -> Dict[str, Any]:
        return self._get_s3_object()

{
    "product": "laptop",
    "price": 1000
}

{
    "ten_percent_off_campaign": {
        "default": true
    }
}

You can unit test your feature flags locally and independently without setting up AWS AppConfig.

AppConfigStore only fetches a JSON document with a specific schema. This allows you to mock the response and use it to verify the rule evaluation.

This excerpt relies on pytest and pytest-mock dependencies.

Testing your code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

from aws_lambda_powertools.utilities.feature_flags import (
    AppConfigStore,
    FeatureFlags,
    RuleAction,
)


def init_feature_flags(mocker, mock_schema, envelope="") -> FeatureFlags:
    """Mock AppConfig Store get_configuration method to use mock schema instead"""

    method_to_mock = "aws_lambda_powertools.utilities.feature_flags.AppConfigStore.get_configuration"
    mocked_get_conf = mocker.patch(method_to_mock)
    mocked_get_conf.return_value = mock_schema

    app_conf_store = AppConfigStore(
        environment="test_env",
        application="test_app",
        name="test_conf_name",
        envelope=envelope,
    )

    return FeatureFlags(store=app_conf_store)


def test_flags_condition_match(mocker):
    # GIVEN
    expected_value = True
    mocked_app_config_schema = {
        "my_feature": {
            "default": False,
            "rules": {
                "tenant id equals 12345": {
                    "when_match": expected_value,
                    "conditions": [
                        {
                            "action": RuleAction.EQUALS.value,
                            "key": "tenant_id",
                            "value": "12345",
                        },
                    ],
                },
            },
        },
    }

    # WHEN
    ctx = {"tenant_id": "12345", "username": "a"}
    feature_flags = init_feature_flags(mocker=mocker, mock_schema=mocked_app_config_schema)
    flag = feature_flags.evaluate(name="my_feature", context=ctx, default=False)

    # THEN
    assert flag == expected_value

RetroSearch is an open source project built by @garambo | Open a GitHub Issue

Search and Browse the WWW like it's 1997 | Search results from DuckDuckGo

HTML: 3.2 | Encoding: UTF-8 | Version: 0.7.4