Type Checking

.github/workflows/validate.yml

@@ -27,5 +27,8 @@ jobs:
     - name: Test with pytest
       run: |
         pytest tests --doctest-modules --cov-report=xml --cov-report=html
+    - name: Run mypy
+      run: |
+        mypy featuremanagement
     - name: cspell-action
       uses: streetsidesoftware/[email protected]

featuremanagement/_defaultfilters.py

@@ -44,7 +44,7 @@ class TimeWindowFilter(FeatureFilter):
     Feature Filter that determines if the current time is within the time window.
     """
 
-    def evaluate(self, context: Mapping, **kwargs: Dict[str, Any]) -> bool:
+    def evaluate(self, context: Mapping[Any, Any], **kwargs: Any) -> bool:
         """
         Determine if the feature flag is enabled for the given context.
 
@@ -87,7 +87,7 @@ def _is_targeted(context_id: str, rollout_percentage: int) -> bool:
         return percentage < rollout_percentage
 
     def _target_group(
-        self, target_user: Optional[str], target_group: str, group: Mapping, feature_flag_name: str
+        self, target_user: Optional[str], target_group: str, group: Mapping[str, Any], feature_flag_name: str
     ) -> bool:
         group_rollout_percentage = group.get(ROLLOUT_PERCENTAGE_KEY, 0)
         if not target_user:
@@ -96,7 +96,7 @@ def _target_group(
 
         return self._is_targeted(audience_context_id, group_rollout_percentage)
 
-    def evaluate(self, context: Mapping, **kwargs: Dict[str, Any]) -> bool:
+    def evaluate(self, context: Mapping[Any, Any], **kwargs: Any) -> bool:
         """
         Determine if the feature flag is enabled for the given context.
 
@@ -156,11 +156,11 @@ def evaluate(self, context: Mapping, **kwargs: Dict[str, Any]) -> bool:
         return self._is_targeted(context_id, default_rollout_percentage)
 
     @staticmethod
-    def _validate(groups: List, default_rollout_percentage: int) -> None:
+    def _validate(groups: List[Dict[str, Any]], default_rollout_percentage: int) -> None:
         # Validate the audience settings
         if default_rollout_percentage < 0 or default_rollout_percentage > 100:
             raise TargetingException("DefaultRolloutPercentage must be between 0 and 100")
 
         for group in groups:
-            if group.get(ROLLOUT_PERCENTAGE_KEY) < 0 or group.get(ROLLOUT_PERCENTAGE_KEY) > 100:
+            if group.get(ROLLOUT_PERCENTAGE_KEY, 0) < 0 or group.get(ROLLOUT_PERCENTAGE_KEY, 100) > 100:
                 raise TargetingException("RolloutPercentage must be between 0 and 100")

featuremanagement/_featurefilters.py

@@ -4,16 +4,19 @@
 # license information.
 # -------------------------------------------------------------------------
 from abc import ABC, abstractmethod
-from typing import Mapping, Callable, Dict, Any
+from typing import Mapping, Callable, Any, Optional
+from typing_extensions import Self
 
 
 class FeatureFilter(ABC):
     """
     Parent class for all feature filters.
     """
 
+    _alias: Optional[str] = None
+
     @abstractmethod
-    def evaluate(self, context: Mapping, **kwargs: Dict[str, Any]) -> bool:
+    def evaluate(self, context: Mapping[Any, Any], **kwargs: Any) -> bool:
         """
         Determine if the feature flag is enabled for the given context.
 
@@ -28,12 +31,12 @@ def name(self) -> str:
         :return: Name of the filter, or alias if it exists.
         :rtype: str
         """
-        if hasattr(self, "_alias"):
-            return self._alias  # type: ignore
+        if hasattr(self, "_alias") and self._alias:
+            return self._alias
         return self.__class__.__name__
 
     @staticmethod
-    def alias(alias: str) -> Callable:
+    def alias(alias: str) -> Callable[..., Any]:
         """
         Decorator to set the alias for the filter.
 
@@ -42,7 +45,7 @@ def alias(alias: str) -> Callable:
         :rtype: Callable
         """
 
-        def wrapper(cls) -> Any:  # type: ignore
+        def wrapper(cls: Self) -> Any:
             cls._alias = alias  # pylint: disable=protected-access
             return cls
 

featuremanagement/_featuremanager.py

@@ -28,17 +28,20 @@ class FeatureManager(FeatureManagerBase):
     evaluated.
     """
 
-    def __init__(self, configuration: Mapping, **kwargs: Dict[str, Any]):
+    def __init__(self, configuration: Mapping[str, Any], **kwargs: Any):
         super().__init__(configuration, **kwargs)
-        filters = [TimeWindowFilter(), TargetingFilter()] + cast(List, kwargs.pop(PROVIDED_FEATURE_FILTERS, []))
+        self._filters: Dict[str, FeatureFilter] = {}
+        filters = [TimeWindowFilter(), TargetingFilter()] + cast(
+            List[FeatureFilter], kwargs.pop(PROVIDED_FEATURE_FILTERS, [])
+        )
 
         for feature_filter in filters:
             if not isinstance(feature_filter, FeatureFilter):
                 raise ValueError("Custom filter must be a subclass of FeatureFilter")
             self._filters[feature_filter.name] = feature_filter
 
     @overload  # type: ignore
-    def is_enabled(self, feature_flag_id: str, user_id: str, **kwargs: Dict[str, Any]) -> bool:
+    def is_enabled(self, feature_flag_id: str, user_id: str, **kwargs: Any) -> bool:
         """
         Determine if the feature flag is enabled for the given context.
 
@@ -48,7 +51,7 @@ def is_enabled(self, feature_flag_id: str, user_id: str, **kwargs: Dict[str, Any
         :rtype: bool
         """
 
-    def is_enabled(self, feature_flag_id: str, *args: Any, **kwargs: Dict[str, Any]) -> bool:  # type: ignore
+    def is_enabled(self, feature_flag_id: str, *args: Any, **kwargs: Any) -> bool:
         """
         Determine if the feature flag is enabled for the given context.
 
@@ -70,7 +73,7 @@ def is_enabled(self, feature_flag_id: str, *args: Any, **kwargs: Dict[str, Any])
         return result.enabled
 
     @overload  # type: ignore
-    def get_variant(self, feature_flag_id: str, user_id: str, **kwargs: Dict[str, Any]) -> Optional[Variant]:
+    def get_variant(self, feature_flag_id: str, user_id: str, **kwargs: Any) -> Optional[Variant]:
         """
         Determine the variant for the given context.
 
@@ -80,9 +83,7 @@ def get_variant(self, feature_flag_id: str, user_id: str, **kwargs: Dict[str, An
         :rtype: Variant
         """
 
-    def get_variant(  # type: ignore
-        self, feature_flag_id: str, *args: Any, **kwargs: Dict[str, Any]
-    ) -> Optional[Variant]:
+    def get_variant(self, feature_flag_id: str, *args: Any, **kwargs: Any) -> Optional[Variant]:
         """
         Determine the variant for the given context.
 
@@ -105,7 +106,7 @@ def get_variant(  # type: ignore
         return result.variant
 
     def _check_feature_filters(
-        self, evaluation_event: EvaluationEvent, targeting_context: TargetingContext, **kwargs: Dict
+        self, evaluation_event: EvaluationEvent, targeting_context: TargetingContext, **kwargs: Any
     ) -> None:
         feature_flag = evaluation_event.feature
         if not feature_flag:
@@ -123,8 +124,8 @@ def _check_feature_filters(
 
         for feature_filter in feature_filters:
             filter_name = feature_filter[FEATURE_FILTER_NAME]
-            kwargs["user"] = targeting_context.user_id  # type: ignore
-            kwargs["groups"] = targeting_context.groups  # type: ignore
+            kwargs["user"] = targeting_context.user_id
+            kwargs["groups"] = targeting_context.groups
             if filter_name not in self._filters:
                 raise ValueError(f"Feature flag {feature_flag.name} has unknown filter {filter_name}")
             if feature_conditions.requirement_type == REQUIREMENT_TYPE_ALL:
@@ -136,7 +137,7 @@ def _check_feature_filters(
                 break
 
     def _check_feature(
-        self, feature_flag_id: str, targeting_context: TargetingContext, **kwargs: Dict[str, Any]
+        self, feature_flag_id: str, targeting_context: TargetingContext, **kwargs: Any
     ) -> EvaluationEvent:
         """
         Determine if the feature flag is enabled for the given context.

featuremanagement/_featuremanagerbase.py

@@ -3,10 +3,9 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # -------------------------------------------------------------------------
-from collections.abc import Mapping
 import hashlib
 from abc import ABC
-from typing import List, Optional, Dict, Tuple, Any
+from typing import List, Optional, Dict, Tuple, Any, Mapping
 from ._models import FeatureFlag, Variant, VariantAssignmentReason, TargetingContext, EvaluationEvent, VariantReference
 
 
@@ -21,7 +20,7 @@
 FEATURE_FILTER_PARAMETERS = "parameters"
 
 
-def _get_feature_flag(configuration: Mapping, feature_flag_name: str) -> Optional[FeatureFlag]:
+def _get_feature_flag(configuration: Mapping[str, Any], feature_flag_name: str) -> Optional[FeatureFlag]:
     """
     Gets the FeatureFlag json from the configuration, if it exists it gets converted to a FeatureFlag object.
 
@@ -44,7 +43,7 @@ def _get_feature_flag(configuration: Mapping, feature_flag_name: str) -> Optiona
     return None
 
 
-def _list_feature_flag_names(configuration: Mapping) -> List[str]:
+def _list_feature_flag_names(configuration: Mapping[str, Any]) -> List[str]:
     """
     List of all feature flag names.
 
@@ -70,12 +69,11 @@ class FeatureManagerBase(ABC):
     Base class for Feature Manager. This class is responsible for all shared logic between the sync and async.
     """
 
-    def __init__(self, configuration: Mapping, **kwargs: Dict[str, Any]):
-        self._filters: Dict = {}
+    def __init__(self, configuration: Mapping[str, Any], **kwargs: Any):
         if configuration is None or not isinstance(configuration, Mapping):
             raise AttributeError("Configuration must be a non-empty dictionary")
         self._configuration = configuration
-        self._cache: Dict = {}
+        self._cache: Dict[str, Optional[FeatureFlag]] = {}
         self._copy = configuration.get(FEATURE_MANAGEMENT_KEY)
         self._on_feature_evaluated = kwargs.pop("on_feature_evaluated", None)
 
@@ -214,7 +212,7 @@ def _variant_name_to_variant(self, feature_flag: FeatureFlag, variant_name: Opti
         for variant_reference in feature_flag.variants:
             if variant_reference.name == variant_name:
                 configuration = variant_reference.configuration_value
-                if not configuration:
+                if not configuration and variant_reference.configuration_reference:
                     configuration = self._configuration.get(variant_reference.configuration_reference)
                 return Variant(variant_reference.name, configuration)
         return None

featuremanagement/_models/_allocation.py

@@ -3,7 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # -------------------------------------------------------------------------
-from typing import cast, List, Optional, Mapping, Dict, Any
+from typing import cast, List, Optional, Mapping, Dict, Any, Union
 from dataclasses import dataclass
 from typing_extensions import Self
 from ._constants import DEFAULT_WHEN_ENABLED, DEFAULT_WHEN_DISABLED, USER, GROUP, PERCENTILE, SEED
@@ -16,7 +16,7 @@ class UserAllocation:
     """
 
     variant: str
-    users: list
+    users: List[str]
 
 
 @dataclass
@@ -26,7 +26,7 @@ class GroupAllocation:
     """
 
     variant: str
-    groups: list
+    groups: List[str]
 
 
 class PercentileAllocation:
@@ -35,12 +35,12 @@ class PercentileAllocation:
     """
 
     def __init__(self) -> None:
-        self._variant = None
+        self._variant: Optional[str] = None
         self._percentile_from: int = 0
         self._percentile_to: Optional[int] = None
 
     @classmethod
-    def convert_from_json(cls, json: Mapping) -> Self:
+    def convert_from_json(cls, json: Mapping[str, Union[str, int]]) -> Self:
         """
         Convert a JSON object to PercentileAllocation.
 
@@ -51,9 +51,21 @@ def convert_from_json(cls, json: Mapping) -> Self:
         if not json:
             raise ValueError("Percentile allocation is not valid.")
         user_allocation = cls()
-        user_allocation._variant = json.get("variant")
-        user_allocation._percentile_from = json.get("from", 0)
-        user_allocation._percentile_to = json.get("to")
+
+        variant = json.get("variant")
+        if not variant or not isinstance(variant, str):
+            raise ValueError("Percentile allocation does not have a valid assigned variant.")
+        user_allocation._variant = variant
+
+        percentile_from = json.get("from", 0)
+        if not isinstance(percentile_from, int):
+            raise ValueError("Percentile allocation does not have a valid starting percentile.")
+        user_allocation._percentile_from = percentile_from
+
+        percentile_to = json.get("to")
+        if not percentile_to or not isinstance(percentile_to, int):
+            raise ValueError("Percentile allocation does not have a valid ending percentile.")
+        user_allocation._percentile_to = percentile_to
         return user_allocation
 
     @property
@@ -101,7 +113,7 @@ def __init__(self, feature_name: str) -> None:
         self._seed = "allocation\n" + feature_name
 
     @classmethod
-    def convert_from_json(cls, json: Dict, feature_name: str) -> Optional[Self]:
+    def convert_from_json(cls, json: Dict[str, Any], feature_name: str) -> Optional[Self]:
         """
         Convert a JSON object to Allocation.
 

featuremanagement/_models/_feature_conditions.py

@@ -4,7 +4,7 @@
 # license information.
 # -------------------------------------------------------------------------
 from collections.abc import Mapping
-from typing import List
+from typing import Any, Dict, List
 from typing_extensions import Self
 from ._constants import (
     FEATURE_FLAG_CLIENT_FILTERS,
@@ -22,7 +22,7 @@ class FeatureConditions:
 
     def __init__(self) -> None:
         self._requirement_type = REQUIREMENT_TYPE_ANY
-        self._client_filters: List[dict] = []
+        self._client_filters: List[Dict[str, Any]] = []
 
     @classmethod
     def convert_from_json(cls, feature_name: str, json_value: str) -> Self:
@@ -55,7 +55,7 @@ def requirement_type(self) -> str:
         return self._requirement_type
 
     @property
-    def client_filters(self) -> List[dict]:
+    def client_filters(self) -> List[Dict[str, Any]]:
         """
         Get the client filters for the feature flag.
 

featuremanagement/_models/_feature_flag.py

@@ -32,7 +32,7 @@ def __init__(self) -> None:
         self._telemetry: Telemetry = Telemetry()
 
     @classmethod
-    def convert_from_json(cls, json_value: Mapping) -> Self:
+    def convert_from_json(cls, json_value: Mapping[str, Any]) -> Self:
         """
         Convert a JSON object to FeatureFlag.
 
@@ -56,7 +56,7 @@ def convert_from_json(cls, json_value: Mapping) -> Self:
             json_value.get(FEATURE_FLAG_ALLOCATION, None), feature_flag._id
         )
         if FEATURE_FLAG_VARIANTS in json_value:
-            variants: List[Mapping] = json_value.get(FEATURE_FLAG_VARIANTS, [])
+            variants: List[Mapping[str, Any]] = json_value.get(FEATURE_FLAG_VARIANTS, [])
             feature_flag._variants = []
             for variant in variants:
                 if variant:

featuremanagement/_models/_telemetry.py

@@ -3,6 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # -------------------------------------------------------------------------
+from typing import Dict
 from dataclasses import dataclass, field
 
 
@@ -13,4 +14,4 @@ class Telemetry:
     """
 
     enabled: bool = False
-    metadata: dict = field(default_factory=dict)
+    metadata: Dict[str, str] = field(default_factory=dict)

featuremanagement/_models/_variant_reference.py

@@ -4,7 +4,7 @@
 # license information.
 # -------------------------------------------------------------------------
 from dataclasses import dataclass
-from typing import Optional, Mapping
+from typing import Optional, Mapping, Any
 from typing_extensions import Self
 from ._constants import VARIANT_REFERENCE_NAME, CONFIGURATION_VALUE, CONFIGURATION_REFERENCE, STATUS_OVERRIDE
 
@@ -22,7 +22,7 @@ def __init__(self) -> None:
         self._status_override = None
 
     @classmethod
-    def convert_from_json(cls, json: Mapping) -> Self:
+    def convert_from_json(cls, json: Mapping[str, Any]) -> Self:
         """
         Convert a JSON object to VariantReference.