feat: strict typing

- add mypy, fix typing errors
- add absolufy-imports, ditch relative imports
- move segment evaluation logic to `flag_engine.segments.evaluation` module
- migrate `IdentityFeaturesList` to a Pydantic model

.github/workflows/pull-request.yml

@@ -40,14 +40,22 @@ jobs:
             - name: Check Formatting
               run: black --check .
 
+            - name: Check Imports
+              run: |
+                  git ls-files | grep '\.py$' | xargs absolufy-imports
+                  isort . --check
+
             - name: Check flake8 linting
               run: flake8 .
 
+            - name: Check Typing
+              run: mypy --strict .
+
             - name: Run Tests
               run: pytest -p no:warnings
 
             - name: Check Coverage
               uses: 5monkeys/cobertura-action@v13
               with:
-                minimum_coverage: 100
-                fail_below_threshold: true
+                  minimum_coverage: 100
+                  fail_below_threshold: true

.pre-commit-config.yaml

@@ -1,6 +1,16 @@
 repos:
+    - repo: https://github.com/pre-commit/mirrors-mypy
+      rev: v1.3.0
+      hooks:
+          - id: mypy
+            args: [--strict]
+            additional_dependencies: [pydantic, pytest, pytest_mock, types-pytest-lazy-fixture, types-setuptools]
+    - repo: https://github.com/MarcoGorelli/absolufy-imports
+      rev: v0.3.1
+      hooks:
+          - id: absolufy-imports
     - repo: https://github.com/PyCQA/isort
-      rev: 5.9.3
+      rev: 5.12.0
       hooks:
           - id: isort
             name: isort (python)

flag_engine/engine.py

@@ -8,7 +8,9 @@
 from flag_engine.utils.exceptions import FeatureStateNotFound
 
 
-def get_environment_feature_states(environment: EnvironmentModel):
+def get_environment_feature_states(
+    environment: EnvironmentModel,
+) -> typing.List[FeatureStateModel]:
     """
     Get a list of feature states for a given environment
 
@@ -19,7 +21,9 @@ def get_environment_feature_states(environment: EnvironmentModel):
     return environment.feature_states
 
 
-def get_environment_feature_state(environment: EnvironmentModel, feature_name: str):
+def get_environment_feature_state(
+    environment: EnvironmentModel, feature_name: str
+) -> FeatureStateModel:
     """
     Get a specific feature state for a given feature_name in a given environment
 
@@ -38,7 +42,7 @@ def get_environment_feature_state(environment: EnvironmentModel, feature_name: s
 def get_identity_feature_states(
     environment: EnvironmentModel,
     identity: IdentityModel,
-    override_traits: typing.List[TraitModel] = None,
+    override_traits: typing.Optional[typing.List[TraitModel]] = None,
 ) -> typing.List[FeatureStateModel]:
     """
     Get a list of feature states for a given identity in a given environment.
@@ -63,8 +67,8 @@ def get_identity_feature_state(
     environment: EnvironmentModel,
     identity: IdentityModel,
     feature_name: str,
-    override_traits: typing.List[TraitModel] = None,
-):
+    override_traits: typing.Optional[typing.List[TraitModel]] = None,
+) -> FeatureStateModel:
     """
     Get a specific feature state for a given identity in a given environment.
 

flag_engine/environments/builders.py

@@ -1,13 +1,17 @@
+import typing
+
 from pydantic import parse_obj_as
 
 from flag_engine.environments.models import EnvironmentAPIKeyModel, EnvironmentModel
 
 
-def build_environment_model(environment_dict: dict) -> EnvironmentModel:
+def build_environment_model(
+    environment_dict: typing.Dict[str, typing.Any]
+) -> EnvironmentModel:
     return parse_obj_as(EnvironmentModel, environment_dict)
 
 
 def build_environment_api_key_model(
-    environment_key_dict: dict,
+    environment_key_dict: typing.Dict[str, typing.Any],
 ) -> EnvironmentAPIKeyModel:
     return parse_obj_as(EnvironmentAPIKeyModel, environment_key_dict)

flag_engine/environments/integrations/models.py

@@ -1,8 +1,9 @@
 from dataclasses import dataclass
+from typing import Optional
 
 
 @dataclass
 class IntegrationModel:
-    api_key: str = None
-    base_url: str = None
-    entity_selector: str = None
+    api_key: Optional[str] = None
+    base_url: Optional[str] = None
+    entity_selector: Optional[str] = None

flag_engine/environments/models.py

@@ -1,7 +1,8 @@
 import typing
 from datetime import datetime
 
-from pydantic.dataclasses import Field, dataclass
+from pydantic import Field
+from pydantic.dataclasses import dataclass
 
 from flag_engine.environments.integrations.models import IntegrationModel
 from flag_engine.features.models import FeatureStateModel
@@ -20,7 +21,7 @@ class EnvironmentAPIKeyModel:
     active: bool = True
 
     @property
-    def is_valid(self):
+    def is_valid(self) -> bool:
         return self.active and (
             not self.expires_at or self.expires_at > utcnow_with_tz()
         )
@@ -53,7 +54,7 @@ class EnvironmentModel:
     webhook_config: typing.Optional[WebhookModel] = None
     hide_disabled_flags: typing.Optional[bool] = None
 
-    _INTEGRATION_ATTS = [
+    _INTEGRATION_ATTRS = [
         "amplitude_config",
         "segment_config",
         "mixpanel_config",
@@ -77,9 +78,9 @@ def integrations_data(self) -> typing.Dict[str, typing.Dict[str, str]]:
         """
 
         integrations_data = {}
-        for integration_attr in self._INTEGRATION_ATTS:
-            integration_config: IntegrationModel = getattr(self, integration_attr, None)
-            if integration_config:
+        for integration_attr in self._INTEGRATION_ATTRS:
+            integration_config: typing.Optional[IntegrationModel]
+            if integration_config := getattr(self, integration_attr, None):
                 integrations_data[integration_attr] = {
                     "base_url": integration_config.base_url,
                     "api_key": integration_config.api_key,

flag_engine/features/models.py

@@ -14,48 +14,48 @@ class FeatureModel:
     name: str
     type: str
 
-    def __eq__(self, other):
-        return self.id == other.id
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, FeatureModel) and self.id == other.id
 
-    def __hash__(self):
+    def __hash__(self) -> int:
         return hash(self.id)
 
 
 @dataclass
 class MultivariateFeatureOptionModel:
     value: typing.Any
-    id: int = None
+    id: typing.Optional[int] = None
 
 
 @dataclass
 class MultivariateFeatureStateValueModel:
     multivariate_feature_option: MultivariateFeatureOptionModel
     percentage_allocation: float
-    id: int = None
+    id: typing.Optional[int] = None
     mv_fs_value_uuid: UUID4 = field(default_factory=uuid.uuid4)
 
 
 @dataclass
 class FeatureSegmentModel:
-    priority: int = None
+    priority: typing.Optional[int] = None
 
 
 @dataclass
 class FeatureStateModel:
     feature: FeatureModel
     enabled: bool
-    django_id: int = None
-    feature_segment: FeatureSegmentModel = None
+    django_id: typing.Optional[int] = None
+    feature_segment: typing.Optional[FeatureSegmentModel] = None
     featurestate_uuid: UUID4 = field(default_factory=uuid.uuid4)
     feature_state_value: typing.Any = field(default=None)
     multivariate_feature_state_values: typing.List[
         MultivariateFeatureStateValueModel
     ] = field(default_factory=list)
 
-    def set_value(self, value: typing.Any):
+    def set_value(self, value: typing.Any) -> None:
         self.feature_state_value = value
 
-    def get_value(self, identity_id: typing.Union[int, str] = None) -> typing.Any:
+    def get_value(self, identity_id: typing.Union[int, str, None] = None) -> typing.Any:
         """
         Get the value of the feature state.
 
@@ -82,18 +82,19 @@ def is_higher_segment_priority(self, other: "FeatureStateModel") -> bool:
 
         """
 
-        try:
-            return (
-                getattr(
-                    self.feature_segment,
-                    "priority",
-                    math.inf,
+        if other_feature_segment := other.feature_segment:
+            if (
+                other_feature_segment_priority := other_feature_segment.priority
+            ) is not None:
+                return (
+                    getattr(
+                        self.feature_segment,
+                        "priority",
+                        math.inf,
+                    )
+                    < other_feature_segment_priority
                 )
-                < other.feature_segment.priority
-            )
-
-        except (TypeError, AttributeError):
-            return False
+        return False
 
     def _get_multivariate_value(
         self, identity_id: typing.Union[int, str]
@@ -107,7 +108,7 @@ def _get_multivariate_value(
         # the percentage allocations of the multivariate options. This gives us a
         # way to ensure that the same value is returned every time we use the same
         # percentage value.
-        start_percentage = 0
+        start_percentage = 0.0
         for mv_value in sorted(
             self.multivariate_feature_state_values,
             key=lambda v: v.id or v.mv_fs_value_uuid,

flag_engine/identities/models.py

@@ -2,8 +2,8 @@
 import typing
 import uuid
 
-from pydantic import UUID4
-from pydantic.dataclasses import Field, dataclass
+from pydantic import UUID4, Field
+from pydantic.dataclasses import dataclass
 
 from flag_engine.identities.traits.models import TraitModel
 from flag_engine.utils.collections import IdentityFeaturesList

flag_engine/identities/traits/types.py

@@ -1,3 +1,3 @@
 from typing import Union
 
-TraitValue = Union[int, str, bool, float]
+TraitValue = Union[None, int, str, bool, float]

flag_engine/organisations/models.py

@@ -10,5 +10,5 @@ class OrganisationModel:
     persist_trait_data: bool
 
     @property
-    def unique_slug(self):
+    def unique_slug(self) -> str:
         return str(self.id) + "-" + self.name

flag_engine/segments/constants.py

@@ -1,39 +1,22 @@
-# Segment Rules
-ALL_RULE = "ALL"
-ANY_RULE = "ANY"
-NONE_RULE = "NONE"
+from flag_engine.segments.types import ConditionOperator, RuleType
 
-RULE_TYPES = [ALL_RULE, ANY_RULE, NONE_RULE]
+# Segment Rules
+ALL_RULE: RuleType = "ALL"
+ANY_RULE: RuleType = "ANY"
+NONE_RULE: RuleType = "NONE"
 
 # Segment Condition Operators
-EQUAL = "EQUAL"
-GREATER_THAN = "GREATER_THAN"
-LESS_THAN = "LESS_THAN"
-LESS_THAN_INCLUSIVE = "LESS_THAN_INCLUSIVE"
-CONTAINS = "CONTAINS"
-GREATER_THAN_INCLUSIVE = "GREATER_THAN_INCLUSIVE"
-NOT_CONTAINS = "NOT_CONTAINS"
-NOT_EQUAL = "NOT_EQUAL"
-REGEX = "REGEX"
-PERCENTAGE_SPLIT = "PERCENTAGE_SPLIT"
-MODULO = "MODULO"
-IS_SET = "IS_SET"
-IS_NOT_SET = "IS_NOT_SET"
-IN = "IN"
-
-CONDITION_OPERATORS = [
-    EQUAL,
-    GREATER_THAN,
-    LESS_THAN,
-    LESS_THAN_INCLUSIVE,
-    CONTAINS,
-    GREATER_THAN_INCLUSIVE,
-    NOT_CONTAINS,
-    NOT_EQUAL,
-    REGEX,
-    PERCENTAGE_SPLIT,
-    MODULO,
-    IS_SET,
-    IS_NOT_SET,
-    IN,
-]
+EQUAL: ConditionOperator = "EQUAL"
+GREATER_THAN: ConditionOperator = "GREATER_THAN"
+LESS_THAN: ConditionOperator = "LESS_THAN"
+LESS_THAN_INCLUSIVE: ConditionOperator = "LESS_THAN_INCLUSIVE"
+CONTAINS: ConditionOperator = "CONTAINS"
+GREATER_THAN_INCLUSIVE: ConditionOperator = "GREATER_THAN_INCLUSIVE"
+NOT_CONTAINS: ConditionOperator = "NOT_CONTAINS"
+NOT_EQUAL: ConditionOperator = "NOT_EQUAL"
+REGEX: ConditionOperator = "REGEX"
+PERCENTAGE_SPLIT: ConditionOperator = "PERCENTAGE_SPLIT"
+MODULO: ConditionOperator = "MODULO"
+IS_SET: ConditionOperator = "IS_SET"
+IS_NOT_SET: ConditionOperator = "IS_NOT_SET"
+IN: ConditionOperator = "IN"