__init__.py

from __future__ import annotations

import typing
from abc import abstractmethod
from collections.abc import Callable, Mapping, Sequence
from enum import Enum

from openfeature.evaluation_context import EvaluationContext
from openfeature.event import ProviderEvent, ProviderEventDetails
from openfeature.flag_evaluation import FlagResolutionDetails
from openfeature.hook import Hook
from openfeature.track import TrackingEventDetails

from .metadata import Metadata

if typing.TYPE_CHECKING:
    from openfeature.flag_evaluation import FlagValueType

__all__ = ["AbstractProvider", "FeatureProvider", "Metadata", "ProviderStatus"]


class ProviderStatus(Enum):
    NOT_READY = "NOT_READY"
    READY = "READY"
    ERROR = "ERROR"
    STALE = "STALE"
    FATAL = "FATAL"


class FeatureProvider(typing.Protocol):  # pragma: no cover
    def attach(
        self,
        on_emit: Callable[[FeatureProvider, ProviderEvent, ProviderEventDetails], None],
    ) -> None: ...

    def detach(self) -> None: ...

    def initialize(self, evaluation_context: EvaluationContext) -> None: ...

    def shutdown(self) -> None: ...

    def get_metadata(self) -> Metadata: ...

    def get_provider_hooks(self) -> list[Hook]: ...

    def resolve_boolean_details(
        self,
        flag_key: str,
        default_value: bool,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[bool]: ...

    async def resolve_boolean_details_async(
        self,
        flag_key: str,
        default_value: bool,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[bool]: ...

    def resolve_string_details(
        self,
        flag_key: str,
        default_value: str,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[str]: ...

    async def resolve_string_details_async(
        self,
        flag_key: str,
        default_value: str,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[str]: ...

    def resolve_integer_details(
        self,
        flag_key: str,
        default_value: int,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[int]: ...

    async def resolve_integer_details_async(
        self,
        flag_key: str,
        default_value: int,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[int]: ...

    def resolve_float_details(
        self,
        flag_key: str,
        default_value: float,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[float]: ...

    async def resolve_float_details_async(
        self,
        flag_key: str,
        default_value: float,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[float]: ...

    def resolve_object_details(
        self,
        flag_key: str,
        default_value: Sequence[FlagValueType] | Mapping[str, FlagValueType],
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[
        Sequence[FlagValueType] | Mapping[str, FlagValueType]
    ]: ...

    async def resolve_object_details_async(
        self,
        flag_key: str,
        default_value: Sequence[FlagValueType] | Mapping[str, FlagValueType],
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[
        Sequence[FlagValueType] | Mapping[str, FlagValueType]
    ]: ...

    def track(self, tracking_event_name: str, evaluation_context: EvaluationContext | None = None, tracking_event_details: TrackingEventDetails | None = None) -> None: ...


class AbstractProvider(FeatureProvider):
    def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
        # this makes sure to invoke the parent of `FeatureProvider` -> `object`
        super(FeatureProvider, self).__init__(*args, **kwargs)

    def attach(
        self,
        on_emit: Callable[[FeatureProvider, ProviderEvent, ProviderEventDetails], None],
    ) -> None:
        self._on_emit = on_emit

    def detach(self) -> None:
        if hasattr(self, "_on_emit"):
            del self._on_emit

    def initialize(self, evaluation_context: EvaluationContext) -> None:
        pass

    def shutdown(self) -> None:
        pass

    def track(self, tracking_event_name: str, evaluation_context: EvaluationContext | None = None, tracking_event_details: TrackingEventDetails | None = None) -> None:
        pass

    @abstractmethod
    def get_metadata(self) -> Metadata:
        pass

    def get_provider_hooks(self) -> list[Hook]:
        return []

    @abstractmethod
    def resolve_boolean_details(
        self,
        flag_key: str,
        default_value: bool,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[bool]:
        pass

    async def resolve_boolean_details_async(
        self,
        flag_key: str,
        default_value: bool,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[bool]:
        return self.resolve_boolean_details(flag_key, default_value, evaluation_context)

    @abstractmethod
    def resolve_string_details(
        self,
        flag_key: str,
        default_value: str,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[str]:
        pass

    async def resolve_string_details_async(
        self,
        flag_key: str,
        default_value: str,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[str]:
        return self.resolve_string_details(flag_key, default_value, evaluation_context)

    @abstractmethod
    def resolve_integer_details(
        self,
        flag_key: str,
        default_value: int,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[int]:
        pass

    async def resolve_integer_details_async(
        self,
        flag_key: str,
        default_value: int,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[int]:
        return self.resolve_integer_details(flag_key, default_value, evaluation_context)

    @abstractmethod
    def resolve_float_details(
        self,
        flag_key: str,
        default_value: float,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[float]:
        pass

    async def resolve_float_details_async(
        self,
        flag_key: str,
        default_value: float,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[float]:
        return self.resolve_float_details(flag_key, default_value, evaluation_context)

    @abstractmethod
    def resolve_object_details(
        self,
        flag_key: str,
        default_value: Sequence[FlagValueType] | Mapping[str, FlagValueType],
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[Sequence[FlagValueType] | Mapping[str, FlagValueType]]:
        pass

    async def resolve_object_details_async(
        self,
        flag_key: str,
        default_value: Sequence[FlagValueType] | Mapping[str, FlagValueType],
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails[Sequence[FlagValueType] | Mapping[str, FlagValueType]]:
        return self.resolve_object_details(flag_key, default_value, evaluation_context)

    def emit_provider_ready(self, details: ProviderEventDetails) -> None:
        self.emit(ProviderEvent.PROVIDER_READY, details)

    def emit_provider_configuration_changed(
        self, details: ProviderEventDetails
    ) -> None:
        self.emit(ProviderEvent.PROVIDER_CONFIGURATION_CHANGED, details)

    def emit_provider_error(self, details: ProviderEventDetails) -> None:
        self.emit(ProviderEvent.PROVIDER_ERROR, details)

    def emit_provider_stale(self, details: ProviderEventDetails) -> None:
        self.emit(ProviderEvent.PROVIDER_STALE, details)

    def emit(self, event: ProviderEvent, details: ProviderEventDetails) -> None:
        if hasattr(self, "_on_emit"):
            self._on_emit(self, event, details)