feat: implement InMemoryProvider (#157)

Signed-off-by: Federico Bond <[email protected]>

Author: federicobond

open_feature/provider/in_memory_provider.py

@@ -0,0 +1,123 @@
+from dataclasses import dataclass
+import typing
+
+from open_feature.evaluation_context.evaluation_context import EvaluationContext
+from open_feature.exception.error_code import ErrorCode
+from open_feature.flag_evaluation.reason import Reason
+from open_feature.flag_evaluation.resolution_details import FlagResolutionDetails
+from open_feature.hooks.hook import Hook
+from open_feature.provider.metadata import Metadata
+from open_feature.provider.provider import AbstractProvider
+
+PASSED_IN_DEFAULT = "Passed in default"
+
+
+@dataclass
+class InMemoryMetadata(Metadata):
+    name: str = "In-Memory Provider"
+
+
+T = typing.TypeVar("T", covariant=True)
+
+
+@dataclass(frozen=True)
+class InMemoryFlag(typing.Generic[T]):
+    flag_key: str
+    default_variant: str
+    variants: typing.Dict[str, T]
+    reason: typing.Optional[Reason] = Reason.STATIC
+    error_code: typing.Optional[ErrorCode] = None
+    error_message: typing.Optional[str] = None
+    context_evaluator: typing.Optional[
+        typing.Callable[["InMemoryFlag", EvaluationContext], FlagResolutionDetails[T]]
+    ] = None
+
+    def resolve(
+        self, evaluation_context: typing.Optional[EvaluationContext]
+    ) -> FlagResolutionDetails[T]:
+        if self.context_evaluator:
+            return self.context_evaluator(
+                self, evaluation_context or EvaluationContext()
+            )
+
+        return FlagResolutionDetails(
+            value=self.variants[self.default_variant],
+            reason=self.reason,
+            variant=self.default_variant,
+            error_code=self.error_code,
+            error_message=self.error_message,
+        )
+
+
+FlagStorage = typing.Dict[str, InMemoryFlag]
+
+V = typing.TypeVar("V")
+
+
+class InMemoryProvider(AbstractProvider):
+    _flags: FlagStorage
+
+    def __init__(self, flags: FlagStorage):
+        self._flags = flags.copy()
+
+    def get_metadata(self) -> Metadata:
+        return InMemoryMetadata()
+
+    def get_provider_hooks(self) -> typing.List[Hook]:
+        return []
+
+    def resolve_boolean_details(
+        self,
+        flag_key: str,
+        default_value: bool,
+        evaluation_context: typing.Optional[EvaluationContext] = None,
+    ) -> FlagResolutionDetails[bool]:
+        return self._resolve(flag_key, default_value, evaluation_context)
+
+    def resolve_string_details(
+        self,
+        flag_key: str,
+        default_value: str,
+        evaluation_context: typing.Optional[EvaluationContext] = None,
+    ) -> FlagResolutionDetails[str]:
+        return self._resolve(flag_key, default_value, evaluation_context)
+
+    def resolve_integer_details(
+        self,
+        flag_key: str,
+        default_value: int,
+        evaluation_context: typing.Optional[EvaluationContext] = None,
+    ) -> FlagResolutionDetails[int]:
+        return self._resolve(flag_key, default_value, evaluation_context)
+
+    def resolve_float_details(
+        self,
+        flag_key: str,
+        default_value: float,
+        evaluation_context: typing.Optional[EvaluationContext] = None,
+    ) -> FlagResolutionDetails[float]:
+        return self._resolve(flag_key, default_value, evaluation_context)
+
+    def resolve_object_details(
+        self,
+        flag_key: str,
+        default_value: typing.Union[dict, list],
+        evaluation_context: typing.Optional[EvaluationContext] = None,
+    ) -> FlagResolutionDetails[typing.Union[dict, list]]:
+        return self._resolve(flag_key, default_value, evaluation_context)
+
+    def _resolve(
+        self,
+        flag_key: str,
+        default_value: V,
+        evaluation_context: typing.Optional[EvaluationContext],
+    ) -> FlagResolutionDetails[V]:
+        flag = self._flags.get(flag_key)
+        if flag is None:
+            return FlagResolutionDetails(
+                value=default_value,
+                reason=Reason.ERROR,
+                error_code=ErrorCode.FLAG_NOT_FOUND,
+                error_message=f"Flag '{flag_key}' not found",
+            )
+        return flag.resolve(evaluation_context)

tests/provider/test_in_memory_provider.py

@@ -0,0 +1,156 @@
+from numbers import Number
+
+from open_feature.exception.error_code import ErrorCode
+from open_feature.flag_evaluation.reason import Reason
+from open_feature.flag_evaluation.resolution_details import FlagResolutionDetails
+from open_feature.provider.in_memory_provider import InMemoryProvider, InMemoryFlag
+
+
+def test_should_return_in_memory_provider_metadata():
+    # Given
+    provider = InMemoryProvider({})
+    # When
+    metadata = provider.get_metadata()
+    # Then
+    assert metadata is not None
+    assert metadata.name == "In-Memory Provider"
+
+
+def test_should_handle_unknown_flags_correctly():
+    # Given
+    provider = InMemoryProvider({})
+    # When
+    flag = provider.resolve_boolean_details(flag_key="Key", default_value=True)
+    # Then
+    assert flag is not None
+    assert flag.value is True
+    assert isinstance(flag.value, bool)
+    assert flag.reason == Reason.ERROR
+    assert flag.error_code == ErrorCode.FLAG_NOT_FOUND
+    assert flag.error_message == "Flag 'Key' not found"
+
+
+def test_calls_context_evaluator_if_present():
+    # Given
+    def context_evaluator(flag: InMemoryFlag, evaluation_context: dict):
+        return FlagResolutionDetails(
+            value=False,
+            reason=Reason.TARGETING_MATCH,
+        )
+
+    provider = InMemoryProvider(
+        {
+            "Key": InMemoryFlag(
+                "Key",
+                "true",
+                {"true": True, "false": False},
+                context_evaluator=context_evaluator,
+            )
+        }
+    )
+    # When
+    flag = provider.resolve_boolean_details(flag_key="Key", default_value=False)
+    # Then
+    assert flag is not None
+    assert flag.value is False
+    assert isinstance(flag.value, bool)
+    assert flag.reason == Reason.TARGETING_MATCH
+
+
+def test_should_resolve_boolean_flag_from_in_memory():
+    # Given
+    provider = InMemoryProvider(
+        {"Key": InMemoryFlag("Key", "true", {"true": True, "false": False})}
+    )
+    # When
+    flag = provider.resolve_boolean_details(flag_key="Key", default_value=False)
+    # Then
+    assert flag is not None
+    assert flag.value is True
+    assert isinstance(flag.value, bool)
+    assert flag.variant == "true"
+
+
+def test_should_resolve_integer_flag_from_in_memory():
+    # Given
+    provider = InMemoryProvider(
+        {"Key": InMemoryFlag("Key", "hundred", {"zero": 0, "hundred": 100})}
+    )
+    # When
+    flag = provider.resolve_integer_details(flag_key="Key", default_value=0)
+    # Then
+    assert flag is not None
+    assert flag.value == 100
+    assert isinstance(flag.value, Number)
+    assert flag.variant == "hundred"
+
+
+def test_should_resolve_float_flag_from_in_memory():
+    # Given
+    provider = InMemoryProvider(
+        {"Key": InMemoryFlag("Key", "ten", {"zero": 0.0, "ten": 10.23})}
+    )
+    # When
+    flag = provider.resolve_float_details(flag_key="Key", default_value=0.0)
+    # Then
+    assert flag is not None
+    assert flag.value == 10.23
+    assert isinstance(flag.value, Number)
+    assert flag.variant == "ten"
+
+
+def test_should_resolve_string_flag_from_in_memory():
+    # Given
+    provider = InMemoryProvider(
+        {
+            "Key": InMemoryFlag(
+                "Key",
+                "stringVariant",
+                {"defaultVariant": "Default", "stringVariant": "String"},
+            )
+        }
+    )
+    # When
+    flag = provider.resolve_string_details(flag_key="Key", default_value="Default")
+    # Then
+    assert flag is not None
+    assert flag.value == "String"
+    assert isinstance(flag.value, str)
+    assert flag.variant == "stringVariant"
+
+
+def test_should_resolve_list_flag_from_in_memory():
+    # Given
+    provider = InMemoryProvider(
+        {
+            "Key": InMemoryFlag(
+                "Key", "twoItems", {"empty": [], "twoItems": ["item1", "item2"]}
+            )
+        }
+    )
+    # When
+    flag = provider.resolve_object_details(flag_key="Key", default_value=[])
+    # Then
+    assert flag is not None
+    assert flag.value == ["item1", "item2"]
+    assert isinstance(flag.value, list)
+    assert flag.variant == "twoItems"
+
+
+def test_should_resolve_object_flag_from_in_memory():
+    # Given
+    return_value = {
+        "String": "string",
+        "Number": 2,
+        "Boolean": True,
+    }
+    provider = InMemoryProvider(
+        {"Key": InMemoryFlag("Key", "obj", {"obj": return_value, "empty": {}})}
+    )
+    # When
+    flag = provider.resolve_object_details(flag_key="Key", default_value={})
+    # Then
+    assert flag is not None
+    assert flag.value == return_value
+    assert isinstance(flag.value, dict)
+    assert flag.variant == "obj"