refactor(feature_flags): add from __future__ import annotations

aws_lambda_powertools/utilities/feature_flags/appconfig.py

@@ -1,10 +1,9 @@
+from __future__ import annotations
+
 import logging
 import traceback
-from typing import Any, Dict, Optional, Union, cast
-
-from botocore.config import Config
+from typing import TYPE_CHECKING, Any, cast
 
-from aws_lambda_powertools.logging import Logger
 from aws_lambda_powertools.utilities import jmespath_utils
 from aws_lambda_powertools.utilities.feature_flags.base import StoreProvider
 from aws_lambda_powertools.utilities.feature_flags.exceptions import ConfigurationStoreError, StoreClientError
@@ -14,6 +13,11 @@
     TransformParameterError,
 )
 
+if TYPE_CHECKING:
+    from botocore.config import Config
+
+    from aws_lambda_powertools.logging import Logger
+
 
 class AppConfigStore(StoreProvider):
     def __init__(
@@ -22,10 +26,10 @@ def __init__(
         application: str,
         name: str,
         max_age: int = 5,
-        sdk_config: Optional[Config] = None,
-        envelope: Optional[str] = "",
-        jmespath_options: Optional[Dict] = None,
-        logger: Optional[Union[logging.Logger, Logger]] = None,
+        sdk_config: Config | None = None,
+        envelope: str | None = "",
+        jmespath_options: dict | None = None,
+        logger: logging.Logger | Logger | None = None,
     ):
         """This class fetches JSON schemas from AWS AppConfig
 
@@ -39,11 +43,11 @@ def __init__(
             AppConfig configuration name e.g. `my_conf`
         max_age: int
             cache expiration time in seconds, or how often to call AppConfig to fetch latest configuration
-        sdk_config: Optional[Config]
+        sdk_config: Config | None
             Botocore Config object to pass during client initialization
-        envelope : Optional[str]
+        envelope : str | None
             JMESPath expression to pluck feature flags data from config
-        jmespath_options : Optional[Dict]
+        jmespath_options : dict | None
             Alternative JMESPath options to be included when filtering expr
         logger: A logging object
             Used to log messages. If None is supplied, one will be created.
@@ -60,7 +64,7 @@ def __init__(
         self._conf_store = AppConfigProvider(environment=environment, application=application, boto_config=sdk_config)
 
     @property
-    def get_raw_configuration(self) -> Dict[str, Any]:
+    def get_raw_configuration(self) -> dict[str, Any]:
         """Fetch feature schema configuration from AWS AppConfig"""
         try:
             # parse result conf as JSON, keep in cache for self.max_age seconds
@@ -82,7 +86,7 @@ def get_raw_configuration(self) -> Dict[str, Any]:
                 raise StoreClientError(err_msg) from exc
             raise ConfigurationStoreError("Unable to get AWS AppConfig configuration file") from exc
 
-    def get_configuration(self) -> Dict[str, Any]:
+    def get_configuration(self) -> dict[str, Any]:
         """Fetch feature schema configuration from AWS AppConfig
 
         If envelope is set, it'll extract and return feature flags from configuration,
@@ -95,7 +99,7 @@ def get_configuration(self) -> Dict[str, Any]:
 
         Returns
         -------
-        Dict[str, Any]
+        dict[str, Any]
             parsed JSON dictionary
         """
         config = self.get_raw_configuration

aws_lambda_powertools/utilities/feature_flags/base.py

@@ -1,16 +1,18 @@
+from __future__ import annotations
+
 from abc import ABC, abstractmethod
-from typing import Any, Dict
+from typing import Any
 
 
 class StoreProvider(ABC):
     @property
     @abstractmethod
-    def get_raw_configuration(self) -> Dict[str, Any]:
+    def get_raw_configuration(self) -> dict[str, Any]:
         """Get configuration from any store and return the parsed JSON dictionary"""
         raise NotImplementedError()  # pragma: no cover
 
     @abstractmethod
-    def get_configuration(self) -> Dict[str, Any]:
+    def get_configuration(self) -> dict[str, Any]:
         """Get configuration from any store and return the parsed JSON dictionary
 
         If envelope is set, it'll extract and return feature flags from configuration,
@@ -23,7 +25,7 @@ def get_configuration(self) -> Dict[str, Any]:
 
         Returns
         -------
-        Dict[str, Any]
+        dict[str, Any]
             parsed JSON dictionary
 
             **Example**

aws_lambda_powertools/utilities/feature_flags/comparators.py

@@ -1,14 +1,15 @@
 from __future__ import annotations
 
 from datetime import datetime, tzinfo
-from typing import Any, Dict, Optional
+from typing import Any
 
 from dateutil.tz import gettz
 
-from aws_lambda_powertools.utilities.feature_flags.schema import HOUR_MIN_SEPARATOR, ModuloRangeValues, TimeValues
+from aws_lambda_powertools.utilities.feature_flags.constants import HOUR_MIN_SEPARATOR
+from aws_lambda_powertools.utilities.feature_flags.schema import ModuloRangeValues, TimeValues
 
 
-def _get_now_from_timezone(timezone: Optional[tzinfo]) -> datetime:
+def _get_now_from_timezone(timezone: tzinfo | None) -> datetime:
     """
     Returns now in the specified timezone. Defaults to UTC if not present.
     At this stage, we already validated that the passed timezone string is valid, so we assume that
@@ -18,7 +19,7 @@ def _get_now_from_timezone(timezone: Optional[tzinfo]) -> datetime:
     return datetime.now(timezone)
 
 
-def compare_days_of_week(context_value: Any, condition_value: Dict) -> bool:
+def compare_days_of_week(context_value: Any, condition_value: dict) -> bool:
     timezone_name = condition_value.get(TimeValues.TIMEZONE.value, "UTC")
 
     # %A = Weekday as locale’s full name.
@@ -28,7 +29,7 @@ def compare_days_of_week(context_value: Any, condition_value: Dict) -> bool:
     return current_day in days
 
 
-def compare_datetime_range(context_value: Any, condition_value: Dict) -> bool:
+def compare_datetime_range(context_value: Any, condition_value: dict) -> bool:
     timezone_name = condition_value.get(TimeValues.TIMEZONE.value, "UTC")
     timezone = gettz(timezone_name)
     current_time: datetime = _get_now_from_timezone(timezone)
@@ -44,7 +45,7 @@ def compare_datetime_range(context_value: Any, condition_value: Dict) -> bool:
     return start_date <= current_time <= end_date
 
 
-def compare_time_range(context_value: Any, condition_value: Dict) -> bool:
+def compare_time_range(context_value: Any, condition_value: dict) -> bool:
     timezone_name = condition_value.get(TimeValues.TIMEZONE.value, "UTC")
     current_time: datetime = _get_now_from_timezone(gettz(timezone_name))
 
@@ -75,7 +76,7 @@ def compare_time_range(context_value: Any, condition_value: Dict) -> bool:
         return start_time <= current_time <= end_time
 
 
-def compare_modulo_range(context_value: int, condition_value: Dict) -> bool:
+def compare_modulo_range(context_value: int, condition_value: dict) -> bool:
     """
     Returns for a given context 'a' and modulo condition 'b' -> b.start <= a % b.base <= b.end
     """

aws_lambda_powertools/utilities/feature_flags/constants.py

@@ -0,0 +1,13 @@
+import re
+
+RULES_KEY = "rules"
+FEATURE_DEFAULT_VAL_KEY = "default"
+CONDITIONS_KEY = "conditions"
+RULE_MATCH_VALUE = "when_match"
+CONDITION_KEY = "key"
+CONDITION_VALUE = "value"
+CONDITION_ACTION = "action"
+FEATURE_DEFAULT_VAL_TYPE_KEY = "boolean_type"
+TIME_RANGE_FORMAT = "%H:%M"  # hour:min 24 hours clock
+TIME_RANGE_PATTERN = re.compile(r"2[0-3]:[0-5]\d|[0-1]\d:[0-5]\d")  # 24 hour clock
+HOUR_MIN_SEPARATOR = ":"

aws_lambda_powertools/utilities/feature_flags/exceptions.py

@@ -7,7 +7,7 @@ class SchemaValidationError(Exception):
 
 
 class StoreClientError(Exception):
-    """When a store raises an exception that should be propagated to the client to fix
+    """When a store raises an exception that should be propagated to the client
 
     For example, Access Denied errors when the client doesn't permissions to fetch config
     """

aws_lambda_powertools/utilities/feature_flags/feature_flags.py

@@ -1,13 +1,9 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Callable, Dict, List, Optional, TypeVar, Union, cast
+from typing import TYPE_CHECKING, Any, Callable, cast
 
-from typing_extensions import ParamSpec
-
-from aws_lambda_powertools.logging import Logger
 from aws_lambda_powertools.utilities.feature_flags import schema
-from aws_lambda_powertools.utilities.feature_flags.base import StoreProvider
 from aws_lambda_powertools.utilities.feature_flags.comparators import (
     compare_all_in_list,
     compare_any_in_list,
@@ -18,10 +14,13 @@
     compare_time_range,
 )
 from aws_lambda_powertools.utilities.feature_flags.exceptions import ConfigurationStoreError
-from aws_lambda_powertools.utilities.feature_flags.types import JSONType
+from aws_lambda_powertools.utilities.feature_flags.types import P, T
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.logging import Logger
+    from aws_lambda_powertools.utilities.feature_flags.base import StoreProvider
+    from aws_lambda_powertools.utilities.feature_flags.types import JSONType
 
-T = TypeVar("T")
-P = ParamSpec("P")
 
 RULE_ACTION_MAPPING = {
     schema.RuleAction.EQUALS.value: lambda a, b: a == b,
@@ -49,7 +48,7 @@
 
 
 class FeatureFlags:
-    def __init__(self, store: StoreProvider, logger: Optional[Union[logging.Logger, Logger]] = None):
+    def __init__(self, store: StoreProvider, logger: logging.Logger | Logger | None = None):
         """Evaluates whether feature flags should be enabled based on a given context.
 
         It uses the provided store to fetch feature flag rules before evaluating them.
@@ -100,12 +99,12 @@ def _evaluate_conditions(
         self,
         rule_name: str,
         feature_name: str,
-        rule: Dict[str, Any],
-        context: Dict[str, Any],
+        rule: dict[str, Any],
+        context: dict[str, Any],
     ) -> bool:
         """Evaluates whether context matches conditions, return False otherwise"""
         rule_match_value = rule.get(schema.RULE_MATCH_VALUE)
-        conditions = cast(List[Dict], rule.get(schema.CONDITIONS_KEY))
+        conditions = cast(list[dict], rule.get(schema.CONDITIONS_KEY))
 
         if not conditions:
             self.logger.debug(
@@ -141,9 +140,9 @@ def _evaluate_rules(
         self,
         *,
         feature_name: str,
-        context: Dict[str, Any],
+        context: dict[str, Any],
         feat_default: Any,
-        rules: Dict[str, Any],
+        rules: dict[str, Any],
         boolean_feature: bool,
     ) -> bool:
         """Evaluates whether context matches rules and conditions, otherwise return feature default"""
@@ -164,7 +163,7 @@ def _evaluate_rules(
         )
         return feat_default
 
-    def get_configuration(self) -> Dict:
+    def get_configuration(self) -> dict:
         """Get validated feature flag schema from configured store.
 
         Largely used to aid testing, since it's called by `evaluate` and `get_enabled_features` methods.
@@ -178,7 +177,7 @@ def get_configuration(self) -> Dict:
 
         Returns
         ------
-        Dict[str, Dict]
+        dict[str, dict]
             parsed JSON dictionary
 
             **Example**
@@ -208,13 +207,13 @@ def get_configuration(self) -> Dict:
         """
         # parse result conf as JSON, keep in cache for max age defined in store
         self.logger.debug(f"Fetching schema from registered store, store={self.store}")
-        config: Dict = self.store.get_configuration()
+        config: dict = self.store.get_configuration()
         validator = schema.SchemaValidator(schema=config, logger=self.logger)
         validator.validate()
 
         return config
 
-    def evaluate(self, *, name: str, context: Optional[Dict[str, Any]] = None, default: JSONType) -> JSONType:
+    def evaluate(self, *, name: str, context: dict[str, Any] | None = None, default: JSONType) -> JSONType:
         """Evaluate whether a feature flag should be enabled according to stored schema and input context
 
         **Logic when evaluating a feature flag**
@@ -243,7 +242,7 @@ def evaluate(self, *, name: str, context: Optional[Dict[str, Any]] = None, defau
         ----------
         name: str
             feature name to evaluate
-        context: Optional[Dict[str, Any]]
+        context: dict[str, Any] | None
             Attributes that should be evaluated against the stored schema.
 
             for example: `{"tenant_id": "X", "username": "Y", "region": "Z"}`
@@ -306,7 +305,7 @@ def lambda_handler(event: dict, context: LambdaContext):
         # Maintenance: Revisit before going GA. We might to simplify customers on-boarding by not requiring it
         # for non-boolean flags. It'll need minor implementation changes, docs changes, and maybe refactor
         # get_enabled_features. We can minimize breaking change, despite Beta label, by having a new
-        # method `get_matching_features` returning Dict[feature_name, feature_value]
+        # method `get_matching_features` returning dict[feature_name, feature_value]
         boolean_feature = feature.get(
             schema.FEATURE_DEFAULT_VAL_TYPE_KEY,
             True,
@@ -330,19 +329,19 @@ def lambda_handler(event: dict, context: LambdaContext):
             boolean_feature=boolean_feature,
         )
 
-    def get_enabled_features(self, *, context: Optional[Dict[str, Any]] = None) -> List[str]:
+    def get_enabled_features(self, *, context: dict[str, Any] | None = None) -> list[str]:
         """Get all enabled feature flags while also taking into account context
         (when a feature has defined rules)
 
         Parameters
         ----------
-        context: Optional[Dict[str, Any]]
+        context: dict[str, Any] | None
             dict of attributes that you would like to match the rules
             against, can be `{'tenant_id: 'X', 'username':' 'Y', 'region': 'Z'}` etc.
 
         Returns
         ----------
-        List[str]
+        list[str]
             list of all feature names that either matches context or have True as default
 
             **Example**
@@ -359,10 +358,10 @@ def get_enabled_features(self, *, context: Optional[Dict[str, Any]] = None) -> L
         if context is None:
             context = {}
 
-        features_enabled: List[str] = []
+        features_enabled: list[str] = []
 
         try:
-            features: Dict[str, Any] = self.get_configuration()
+            features: dict[str, Any] = self.get_configuration()
         except ConfigurationStoreError as err:
             self.logger.debug(f"Failed to fetch feature flags from store, returning empty list, reason={err}")
             return features_enabled