Module aws_lambda_powertools.utilities.feature_flags.schema

Expand source code
import logging
from enum import Enum
from typing import Any, Dict, List, Optional, Union

from ... import Logger
from .base import BaseValidator
from .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"


class RuleAction(str, Enum):
    EQUALS = "EQUALS"
    NOT_EQUALS = "NOT_EQUALS"
    KEY_GREATER_THAN_VALUE = "KEY_GREATER_THAN_VALUE"
    KEY_GREATER_THAN_OR_EQUAL_VALUE = "KEY_GREATER_THAN_OR_EQUAL_VALUE"
    KEY_LESS_THAN_VALUE = "KEY_LESS_THAN_VALUE"
    KEY_LESS_THAN_OR_EQUAL_VALUE = "KEY_LESS_THAN_OR_EQUAL_VALUE"
    STARTSWITH = "STARTSWITH"
    ENDSWITH = "ENDSWITH"
    IN = "IN"
    NOT_IN = "NOT_IN"
    KEY_IN_VALUE = "KEY_IN_VALUE"
    KEY_NOT_IN_VALUE = "KEY_NOT_IN_VALUE"
    VALUE_IN_KEY = "VALUE_IN_KEY"
    VALUE_NOT_IN_KEY = "VALUE_NOT_IN_KEY"


class SchemaValidator(BaseValidator):
    """Validates feature flag schema configuration

    Raises
    ------
    SchemaValidationError
        When schema doesn't conform with feature flag schema

    Schema
    ------

    **Feature object**

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

    `JSONType` being any JSON primitive value: `Union[str, int, float, bool, None, Dict[str, Any], List[Any]]`

    ```json
    {
        "my_feature": {
            "default": true,
            "rules": {}
        },
        "my_non_boolean_feature": {
            "default": {"group": "read-only"},
            "boolean_type": false,
            "rules": {}
        }
    }
    ```

    **Rules object**

    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

    ```json
    {
        "my_feature": {
            "default": true,
            "rules": {
                "tenant id equals 345345435": {
                    "when_match": false,
                    "conditions": []
                }
            }
        },
        "my_non_boolean_feature": {
            "default": {"group": "read-only"},
            "boolean_type": false,
            "rules": {
                "tenant id equals 345345435": {
                    "when_match": {"group": "admin"},
                    "conditions": []
                }
            }
        }
    }
    ```

    **Conditions object**

    A list of dictionaries containing conditions for a given rule.
    The value MUST contain the following members:

    * **action**: `str`. Operation to perform to match a key and value.
    The value MUST be either EQUALS, STARTSWITH, ENDSWITH,
    KEY_IN_VALUE KEY_NOT_IN_VALUE VALUE_IN_KEY VALUE_NOT_IN_KEY

    * **key**: `str`. Key in given context to perform operation
    * **value**: `Any`. Value in given context that should match action operation.

    ```json
    {
        "my_feature": {
            "default": true,
            "rules": {
                "tenant id equals 345345435": {
                    "when_match": false,
                    "conditions": [
                        {
                            "action": "EQUALS",
                            "key": "tenant_id",
                            "value": "345345435",
                        }
                    ]
                }
            }
        }
    }
    ```
    """

    def __init__(self, schema: Dict[str, Any], logger: Optional[Union[logging.Logger, Logger]] = None):
        self.schema = schema
        self.logger = logger or logging.getLogger(__name__)

    def validate(self) -> None:
        self.logger.debug("Validating schema")
        if not isinstance(self.schema, dict):
            raise SchemaValidationError(f"Features must be a dictionary, schema={str(self.schema)}")

        features = FeaturesValidator(schema=self.schema)
        features.validate()


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):
        self.schema = schema
        self.logger = logger or logging.getLogger(__name__)

    def validate(self):
        for name, feature in self.schema.items():
            self.logger.debug(f"Attempting to validate feature '{name}'")
            boolean_feature: bool = self.validate_feature(name, feature)
            rules = RulesValidator(feature=feature, boolean_feature=boolean_feature)
            rules.validate()

    # returns True in case the feature is a regular feature flag with a  boolean default value
    @staticmethod
    def validate_feature(name, feature) -> bool:
        if not feature or not isinstance(feature, dict):
            raise SchemaValidationError(f"Feature must be a non-empty dictionary, feature={name}")

        default_value: Any = feature.get(FEATURE_DEFAULT_VAL_KEY)
        boolean_feature: bool = feature.get(FEATURE_DEFAULT_VAL_TYPE_KEY, True)
        # if feature is boolean_feature, default_value must be a boolean type.
        # default_value must exist
        # Maintenance: Revisit before going GA. We might to simplify customers on-boarding by not requiring it
        # for non-boolean flags.
        if default_value is None or (not isinstance(default_value, bool) and boolean_feature):
            raise SchemaValidationError(f"feature 'default' boolean key must be present, feature={name}")
        return boolean_feature


class RulesValidator(BaseValidator):
    """Validates each rule and calls ConditionsValidator to validate each rule's conditions"""

    def __init__(
        self, feature: Dict[str, Any], boolean_feature: bool, logger: Optional[Union[logging.Logger, Logger]] = None
    ):
        self.feature = feature
        self.feature_name = next(iter(self.feature))
        self.rules: Optional[Dict] = self.feature.get(RULES_KEY)
        self.logger = logger or logging.getLogger(__name__)
        self.boolean_feature = boolean_feature

    def validate(self):
        if not self.rules:
            self.logger.debug("Rules are empty, ignoring validation")
            return

        if not isinstance(self.rules, dict):
            raise SchemaValidationError(f"Feature rules must be a dictionary, feature={self.feature_name}")

        for rule_name, rule in self.rules.items():
            self.logger.debug(f"Attempting to validate rule '{rule_name}'")
            self.validate_rule(
                rule=rule, rule_name=rule_name, feature_name=self.feature_name, boolean_feature=self.boolean_feature
            )
            conditions = ConditionsValidator(rule=rule, rule_name=rule_name)
            conditions.validate()

    @staticmethod
    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}")

        RulesValidator.validate_rule_name(rule_name=rule_name, feature_name=feature_name)
        RulesValidator.validate_rule_default_value(rule=rule, rule_name=rule_name, boolean_feature=boolean_feature)

    @staticmethod
    def validate_rule_name(rule_name: str, feature_name: str):
        if not rule_name or not isinstance(rule_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):
        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, {})
        self.rule_name = rule_name
        self.logger = logger or logging.getLogger(__name__)

    def validate(self):
        if not self.conditions or not isinstance(self.conditions, list):
            raise SchemaValidationError(f"Invalid condition, rule={self.rule_name}")

        for condition in self.conditions:
            # Condition can contain PII data; do not log condition value
            self.logger.debug(f"Attempting to validate condition for '{self.rule_name}'")
            self.validate_condition(rule_name=self.rule_name, condition=condition)

    @staticmethod
    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}")

        ConditionsValidator.validate_condition_action(condition=condition, rule_name=rule_name)
        ConditionsValidator.validate_condition_key(condition=condition, rule_name=rule_name)
        ConditionsValidator.validate_condition_value(condition=condition, rule_name=rule_name)

    @staticmethod
    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]
            raise SchemaValidationError(
                f"'action' value must be either {allowed_values}, rule_name={rule_name}, action={action}"
            )

    @staticmethod
    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}")

    @staticmethod
    def validate_condition_value(condition: Dict[str, Any], rule_name: str):
        value = condition.get(CONDITION_VALUE, "")
        if not value:
            raise SchemaValidationError(f"'value' key must not be empty, rule={rule_name}")

Classes

class ConditionsValidator (rule: Dict[str, Any], rule_name: str, logger: Union[logging.Logger, Logger, None] = None)

Helper class that provides a standard way to create an ABC using inheritance.

Expand source code
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, {})
        self.rule_name = rule_name
        self.logger = logger or logging.getLogger(__name__)

    def validate(self):
        if not self.conditions or not isinstance(self.conditions, list):
            raise SchemaValidationError(f"Invalid condition, rule={self.rule_name}")

        for condition in self.conditions:
            # Condition can contain PII data; do not log condition value
            self.logger.debug(f"Attempting to validate condition for '{self.rule_name}'")
            self.validate_condition(rule_name=self.rule_name, condition=condition)

    @staticmethod
    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}")

        ConditionsValidator.validate_condition_action(condition=condition, rule_name=rule_name)
        ConditionsValidator.validate_condition_key(condition=condition, rule_name=rule_name)
        ConditionsValidator.validate_condition_value(condition=condition, rule_name=rule_name)

    @staticmethod
    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]
            raise SchemaValidationError(
                f"'action' value must be either {allowed_values}, rule_name={rule_name}, action={action}"
            )

    @staticmethod
    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}")

    @staticmethod
    def validate_condition_value(condition: Dict[str, Any], rule_name: str):
        value = condition.get(CONDITION_VALUE, "")
        if not value:
            raise SchemaValidationError(f"'value' key must not be empty, rule={rule_name}")

Ancestors

Static methods

def validate_condition(rule_name: str, condition: Dict[str, str]) ‑> None
Expand source code
@staticmethod
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}")

    ConditionsValidator.validate_condition_action(condition=condition, rule_name=rule_name)
    ConditionsValidator.validate_condition_key(condition=condition, rule_name=rule_name)
    ConditionsValidator.validate_condition_value(condition=condition, rule_name=rule_name)
def validate_condition_action(condition: Dict[str, Any], rule_name: str)
Expand source code
@staticmethod
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]
        raise SchemaValidationError(
            f"'action' value must be either {allowed_values}, rule_name={rule_name}, action={action}"
        )
def validate_condition_key(condition: Dict[str, Any], rule_name: str)
Expand source code
@staticmethod
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}")
def validate_condition_value(condition: Dict[str, Any], rule_name: str)
Expand source code
@staticmethod
def validate_condition_value(condition: Dict[str, Any], rule_name: str):
    value = condition.get(CONDITION_VALUE, "")
    if not value:
        raise SchemaValidationError(f"'value' key must not be empty, rule={rule_name}")

Methods

def validate(self)
Expand source code
def validate(self):
    if not self.conditions or not isinstance(self.conditions, list):
        raise SchemaValidationError(f"Invalid condition, rule={self.rule_name}")

    for condition in self.conditions:
        # Condition can contain PII data; do not log condition value
        self.logger.debug(f"Attempting to validate condition for '{self.rule_name}'")
        self.validate_condition(rule_name=self.rule_name, condition=condition)
class FeaturesValidator (schema: Dict[~KT, ~VT], logger: Union[logging.Logger, Logger, None] = None)

Validates each feature and calls RulesValidator to validate its rules

Expand source code
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):
        self.schema = schema
        self.logger = logger or logging.getLogger(__name__)

    def validate(self):
        for name, feature in self.schema.items():
            self.logger.debug(f"Attempting to validate feature '{name}'")
            boolean_feature: bool = self.validate_feature(name, feature)
            rules = RulesValidator(feature=feature, boolean_feature=boolean_feature)
            rules.validate()

    # returns True in case the feature is a regular feature flag with a  boolean default value
    @staticmethod
    def validate_feature(name, feature) -> bool:
        if not feature or not isinstance(feature, dict):
            raise SchemaValidationError(f"Feature must be a non-empty dictionary, feature={name}")

        default_value: Any = feature.get(FEATURE_DEFAULT_VAL_KEY)
        boolean_feature: bool = feature.get(FEATURE_DEFAULT_VAL_TYPE_KEY, True)
        # if feature is boolean_feature, default_value must be a boolean type.
        # default_value must exist
        # Maintenance: Revisit before going GA. We might to simplify customers on-boarding by not requiring it
        # for non-boolean flags.
        if default_value is None or (not isinstance(default_value, bool) and boolean_feature):
            raise SchemaValidationError(f"feature 'default' boolean key must be present, feature={name}")
        return boolean_feature

Ancestors

Static methods

def validate_feature(name, feature) ‑> bool
Expand source code
@staticmethod
def validate_feature(name, feature) -> bool:
    if not feature or not isinstance(feature, dict):
        raise SchemaValidationError(f"Feature must be a non-empty dictionary, feature={name}")

    default_value: Any = feature.get(FEATURE_DEFAULT_VAL_KEY)
    boolean_feature: bool = feature.get(FEATURE_DEFAULT_VAL_TYPE_KEY, True)
    # if feature is boolean_feature, default_value must be a boolean type.
    # default_value must exist
    # Maintenance: Revisit before going GA. We might to simplify customers on-boarding by not requiring it
    # for non-boolean flags.
    if default_value is None or (not isinstance(default_value, bool) and boolean_feature):
        raise SchemaValidationError(f"feature 'default' boolean key must be present, feature={name}")
    return boolean_feature

Methods

def validate(self)
Expand source code
def validate(self):
    for name, feature in self.schema.items():
        self.logger.debug(f"Attempting to validate feature '{name}'")
        boolean_feature: bool = self.validate_feature(name, feature)
        rules = RulesValidator(feature=feature, boolean_feature=boolean_feature)
        rules.validate()
class RuleAction (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code
class RuleAction(str, Enum):
    EQUALS = "EQUALS"
    NOT_EQUALS = "NOT_EQUALS"
    KEY_GREATER_THAN_VALUE = "KEY_GREATER_THAN_VALUE"
    KEY_GREATER_THAN_OR_EQUAL_VALUE = "KEY_GREATER_THAN_OR_EQUAL_VALUE"
    KEY_LESS_THAN_VALUE = "KEY_LESS_THAN_VALUE"
    KEY_LESS_THAN_OR_EQUAL_VALUE = "KEY_LESS_THAN_OR_EQUAL_VALUE"
    STARTSWITH = "STARTSWITH"
    ENDSWITH = "ENDSWITH"
    IN = "IN"
    NOT_IN = "NOT_IN"
    KEY_IN_VALUE = "KEY_IN_VALUE"
    KEY_NOT_IN_VALUE = "KEY_NOT_IN_VALUE"
    VALUE_IN_KEY = "VALUE_IN_KEY"
    VALUE_NOT_IN_KEY = "VALUE_NOT_IN_KEY"

Ancestors

  • builtins.str
  • enum.Enum

Class variables

var ENDSWITH
var EQUALS
var IN
var KEY_GREATER_THAN_OR_EQUAL_VALUE
var KEY_GREATER_THAN_VALUE
var KEY_IN_VALUE
var KEY_LESS_THAN_OR_EQUAL_VALUE
var KEY_LESS_THAN_VALUE
var KEY_NOT_IN_VALUE
var NOT_EQUALS
var NOT_IN
var STARTSWITH
var VALUE_IN_KEY
var VALUE_NOT_IN_KEY
class RulesValidator (feature: Dict[str, Any], boolean_feature: bool, logger: Union[logging.Logger, Logger, None] = None)

Validates each rule and calls ConditionsValidator to validate each rule's conditions

Expand source code
class RulesValidator(BaseValidator):
    """Validates each rule and calls ConditionsValidator to validate each rule's conditions"""

    def __init__(
        self, feature: Dict[str, Any], boolean_feature: bool, logger: Optional[Union[logging.Logger, Logger]] = None
    ):
        self.feature = feature
        self.feature_name = next(iter(self.feature))
        self.rules: Optional[Dict] = self.feature.get(RULES_KEY)
        self.logger = logger or logging.getLogger(__name__)
        self.boolean_feature = boolean_feature

    def validate(self):
        if not self.rules:
            self.logger.debug("Rules are empty, ignoring validation")
            return

        if not isinstance(self.rules, dict):
            raise SchemaValidationError(f"Feature rules must be a dictionary, feature={self.feature_name}")

        for rule_name, rule in self.rules.items():
            self.logger.debug(f"Attempting to validate rule '{rule_name}'")
            self.validate_rule(
                rule=rule, rule_name=rule_name, feature_name=self.feature_name, boolean_feature=self.boolean_feature
            )
            conditions = ConditionsValidator(rule=rule, rule_name=rule_name)
            conditions.validate()

    @staticmethod
    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}")

        RulesValidator.validate_rule_name(rule_name=rule_name, feature_name=feature_name)
        RulesValidator.validate_rule_default_value(rule=rule, rule_name=rule_name, boolean_feature=boolean_feature)

    @staticmethod
    def validate_rule_name(rule_name: str, feature_name: str):
        if not rule_name or not isinstance(rule_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):
        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}")

Ancestors

Static methods

def validate_rule(rule: Dict[~KT, ~VT], rule_name: str, feature_name: str, boolean_feature: bool = True)
Expand source code
@staticmethod
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}")

    RulesValidator.validate_rule_name(rule_name=rule_name, feature_name=feature_name)
    RulesValidator.validate_rule_default_value(rule=rule, rule_name=rule_name, boolean_feature=boolean_feature)
def validate_rule_default_value(rule: Dict[~KT, ~VT], rule_name: str, boolean_feature: bool)
Expand source code
@staticmethod
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}")
def validate_rule_name(rule_name: str, feature_name: str)
Expand source code
@staticmethod
def validate_rule_name(rule_name: str, feature_name: str):
    if not rule_name or not isinstance(rule_name, str):
        raise SchemaValidationError(f"Rule name key must have a non-empty string, feature={feature_name}")

Methods

def validate(self)
Expand source code
def validate(self):
    if not self.rules:
        self.logger.debug("Rules are empty, ignoring validation")
        return

    if not isinstance(self.rules, dict):
        raise SchemaValidationError(f"Feature rules must be a dictionary, feature={self.feature_name}")

    for rule_name, rule in self.rules.items():
        self.logger.debug(f"Attempting to validate rule '{rule_name}'")
        self.validate_rule(
            rule=rule, rule_name=rule_name, feature_name=self.feature_name, boolean_feature=self.boolean_feature
        )
        conditions = ConditionsValidator(rule=rule, rule_name=rule_name)
        conditions.validate()
class SchemaValidator (schema: Dict[str, Any], logger: Union[logging.Logger, Logger, None] = None)

Validates feature flag schema configuration

Raises

SchemaValidationError
When schema doesn't conform with feature flag schema

Schema

Feature object

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

JSONType being any JSON primitive value: Union[str, int, float, bool, None, Dict[str, Any], List[Any]]

{
    "my_feature": {
        "default": true,
        "rules": {}
    },
    "my_non_boolean_feature": {
        "default": {"group": "read-only"},
        "boolean_type": false,
        "rules": {}
    }
}

Rules object

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
{
    "my_feature": {
        "default": true,
        "rules": {
            "tenant id equals 345345435": {
                "when_match": false,
                "conditions": []
            }
        }
    },
    "my_non_boolean_feature": {
        "default": {"group": "read-only"},
        "boolean_type": false,
        "rules": {
            "tenant id equals 345345435": {
                "when_match": {"group": "admin"},
                "conditions": []
            }
        }
    }
}

Conditions object

A list of dictionaries containing conditions for a given rule. The value MUST contain the following members:

  • action: str. Operation to perform to match a key and value. The value MUST be either EQUALS, STARTSWITH, ENDSWITH, KEY_IN_VALUE KEY_NOT_IN_VALUE VALUE_IN_KEY VALUE_NOT_IN_KEY

  • key: str. Key in given context to perform operation

  • value: Any. Value in given context that should match action operation.
{
    "my_feature": {
        "default": true,
        "rules": {
            "tenant id equals 345345435": {
                "when_match": false,
                "conditions": [
                    {
                        "action": "EQUALS",
                        "key": "tenant_id",
                        "value": "345345435",
                    }
                ]
            }
        }
    }
}
Expand source code
class SchemaValidator(BaseValidator):
    """Validates feature flag schema configuration

    Raises
    ------
    SchemaValidationError
        When schema doesn't conform with feature flag schema

    Schema
    ------

    **Feature object**

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

    `JSONType` being any JSON primitive value: `Union[str, int, float, bool, None, Dict[str, Any], List[Any]]`

    ```json
    {
        "my_feature": {
            "default": true,
            "rules": {}
        },
        "my_non_boolean_feature": {
            "default": {"group": "read-only"},
            "boolean_type": false,
            "rules": {}
        }
    }
    ```

    **Rules object**

    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

    ```json
    {
        "my_feature": {
            "default": true,
            "rules": {
                "tenant id equals 345345435": {
                    "when_match": false,
                    "conditions": []
                }
            }
        },
        "my_non_boolean_feature": {
            "default": {"group": "read-only"},
            "boolean_type": false,
            "rules": {
                "tenant id equals 345345435": {
                    "when_match": {"group": "admin"},
                    "conditions": []
                }
            }
        }
    }
    ```

    **Conditions object**

    A list of dictionaries containing conditions for a given rule.
    The value MUST contain the following members:

    * **action**: `str`. Operation to perform to match a key and value.
    The value MUST be either EQUALS, STARTSWITH, ENDSWITH,
    KEY_IN_VALUE KEY_NOT_IN_VALUE VALUE_IN_KEY VALUE_NOT_IN_KEY

    * **key**: `str`. Key in given context to perform operation
    * **value**: `Any`. Value in given context that should match action operation.

    ```json
    {
        "my_feature": {
            "default": true,
            "rules": {
                "tenant id equals 345345435": {
                    "when_match": false,
                    "conditions": [
                        {
                            "action": "EQUALS",
                            "key": "tenant_id",
                            "value": "345345435",
                        }
                    ]
                }
            }
        }
    }
    ```
    """

    def __init__(self, schema: Dict[str, Any], logger: Optional[Union[logging.Logger, Logger]] = None):
        self.schema = schema
        self.logger = logger or logging.getLogger(__name__)

    def validate(self) -> None:
        self.logger.debug("Validating schema")
        if not isinstance(self.schema, dict):
            raise SchemaValidationError(f"Features must be a dictionary, schema={str(self.schema)}")

        features = FeaturesValidator(schema=self.schema)
        features.validate()

Ancestors

Methods

def validate(self) ‑> None
Expand source code
def validate(self) -> None:
    self.logger.debug("Validating schema")
    if not isinstance(self.schema, dict):
        raise SchemaValidationError(f"Features must be a dictionary, schema={str(self.schema)}")

    features = FeaturesValidator(schema=self.schema)
    features.validate()