Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: implement InMemoryProvider #157

Merged
merged 3 commits into from
Jul 25, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
123 changes: 123 additions & 0 deletions open_feature/provider/in_memory_provider.py
Original file line number Diff line number Diff line change
@@ -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
beeme1mr marked this conversation as resolved.
Show resolved Hide resolved
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):
federicobond marked this conversation as resolved.
Show resolved Hide resolved
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(
federicobond marked this conversation as resolved.
Show resolved Hide resolved
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)
156 changes: 156 additions & 0 deletions tests/provider/test_in_memory_provider.py
Original file line number Diff line number Diff line change
@@ -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"
beeme1mr marked this conversation as resolved.
Show resolved Hide resolved