import threading
from typing import Any, List, Optional, Union
from ldclient import LDClient, Config
from ldclient.interfaces import DataSourceStatus, FlagChange, DataSourceState
from openfeature.evaluation_context import EvaluationContext
from openfeature.exception import ErrorCode, ProviderFatalError
from openfeature.flag_evaluation import FlagResolutionDetails, FlagType, Reason
from openfeature.hook import Hook
from openfeature.provider.metadata import Metadata
from openfeature.provider import AbstractProvider
from openfeature.event import ProviderEventDetails
from ld_openfeature.impl.context_converter import EvaluationContextConverter
from ld_openfeature.impl.details_converter import ResolutionDetailsConverter
[docs]class LaunchDarklyProvider(AbstractProvider):
def __init__(self, config: Config):
self.__client = LDClient(config)
self.__context_converter = EvaluationContextConverter()
self.__details_converter = ResolutionDetailsConverter()
def __handle_data_source_status(self, status: DataSourceStatus):
state = status.state
if state == DataSourceState.INITIALIZING:
return
elif state == DataSourceState.VALID:
self.emit_provider_ready(ProviderEventDetails())
elif state == DataSourceState.OFF:
error_message = self.__get_message(status,
"the provider has encountered a permanent error or has been shutdown")
self.emit_provider_error(ProviderEventDetails(error_code=ErrorCode.PROVIDER_FATAL,
message=error_message))
elif state == DataSourceState.INTERRUPTED:
error_message = self.__get_message(status, "encountered an unknown error")
self.emit_provider_stale(ProviderEventDetails(message=error_message))
# For now treat an unknown state as no change.
def __handle_flag_change(self, change: FlagChange):
self.emit_provider_configuration_changed(ProviderEventDetails(flags_changed=[change.key]))
pass
[docs] def initialize(self, evaluation_context: EvaluationContext):
ready_event = threading.Event()
def ready_handler(status: DataSourceStatus):
if status.state == DataSourceState.VALID:
ready_event.set()
elif status.state == DataSourceState.OFF:
ready_event.set()
# We listen just to handle the ready event. We do not emit events because the client emits them for us.
self.__client.data_source_status_provider.add_listener(ready_handler)
# Check for conditions that may have happened before we added the listener.
if self.__client.data_source_status_provider.status.state == DataSourceState.OFF:
ready_event.set()
if self.__client.is_initialized():
ready_event.set()
ready_event.wait()
self.__client.data_source_status_provider.remove_listener(ready_handler)
if not self.__client.is_initialized():
raise ProviderFatalError(error_message="launchdarkly client initialization failed")
# Listen to new status events and emit them.
self.__client.data_source_status_provider.add_listener(self.__handle_data_source_status)
self.__client.flag_tracker.add_listener(self.__handle_flag_change)
[docs] def shutdown(self):
self.__client.data_source_status_provider.remove_listener(self.__handle_data_source_status)
self.__client.flag_tracker.remove_listener(self.__handle_flag_change)
self.__client.close()
[docs] def get_provider_hooks(self) -> List[Hook]:
return []
[docs] def resolve_boolean_details(
self,
flag_key: str,
default_value: bool,
evaluation_context: Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[bool]:
"""Resolves the flag value for the provided flag key as a boolean"""
return self.__resolve_value(FlagType(FlagType.BOOLEAN), flag_key, default_value, evaluation_context)
[docs] def resolve_string_details(
self,
flag_key: str,
default_value: str,
evaluation_context: Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[str]:
"""Resolves the flag value for the provided flag key as a string"""
return self.__resolve_value(FlagType(FlagType.STRING), flag_key, default_value, evaluation_context)
[docs] def resolve_integer_details(
self,
flag_key: str,
default_value: int,
evaluation_context: Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[int]:
"""Resolves the flag value for the provided flag key as a integer"""
return self.__resolve_value(FlagType(FlagType.INTEGER), flag_key, default_value, evaluation_context)
[docs] def resolve_float_details(
self,
flag_key: str,
default_value: float,
evaluation_context: Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[float]:
"""Resolves the flag value for the provided flag key as a float"""
return self.__resolve_value(FlagType(FlagType.FLOAT), flag_key, default_value, evaluation_context)
[docs] def resolve_object_details(
self,
flag_key: str,
default_value: Union[dict, list],
evaluation_context: Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[Union[dict, list]]:
"""Resolves the flag value for the provided flag key as a list or dictionary"""
return self.__resolve_value(FlagType(FlagType.OBJECT), flag_key, default_value, evaluation_context)
def __resolve_value(self, flag_type: FlagType, flag_key: str, default_value: Any,
evaluation_context: Optional[EvaluationContext] = None) -> FlagResolutionDetails:
if evaluation_context is None:
return FlagResolutionDetails(
value=default_value,
reason=Reason(Reason.ERROR),
error_code=ErrorCode.TARGETING_KEY_MISSING
)
ld_context = self.__context_converter.to_ld_context(evaluation_context)
result = self.__client.variation_detail(flag_key, ld_context, default_value)
if flag_type == FlagType.BOOLEAN and not isinstance(result.value, bool):
return self.__mismatched_type_details(default_value)
elif flag_type == FlagType.STRING and not isinstance(result.value, str):
return self.__mismatched_type_details(default_value)
elif flag_type == FlagType.INTEGER and isinstance(result.value, bool):
# Python treats boolean values as instances of int
return self.__mismatched_type_details(default_value)
elif flag_type == FlagType.FLOAT and isinstance(result.value, bool):
# Python treats boolean values as instances of int
return self.__mismatched_type_details(default_value)
elif flag_type == FlagType.INTEGER and not isinstance(result.value, int):
return self.__mismatched_type_details(default_value)
elif flag_type == FlagType.FLOAT and not isinstance(result.value, float) and not isinstance(result.value, int):
return self.__mismatched_type_details(default_value)
elif flag_type == FlagType.OBJECT and not isinstance(result.value, dict) and not isinstance(result.value, list):
return self.__mismatched_type_details(default_value)
return self.__details_converter.to_resolution_details(result)
@staticmethod
def __mismatched_type_details(default_value: Any) -> FlagResolutionDetails:
return FlagResolutionDetails(
value=default_value,
reason=Reason(Reason.ERROR),
error_code=ErrorCode.TYPE_MISMATCH
)
@staticmethod
def __get_message(status: DataSourceStatus, default: str):
if status.error and status.error.message:
return status.error.message
return default