open-feature
/
python-sdk
mirror of https://github.com/open-feature/python-sdk.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refine typing.Any type hints 

Signed-off-by: gruebel <anton.gruebel@gmail.com>
This commit is contained in:
gruebel 2025-06-01 00:10:52 +02:00
parent 87e448593d
commit e19133d58e
8 changed files with 254 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

169
openfeature/client.py
View File

@ -1,6 +1,6 @@
import logging
import typing
from collections.abc import Awaitable
from collections.abc import Awaitable, Sequence
from dataclasses import dataclass
from openfeature import _event_support
@ -19,6 +19,7 @@ from openfeature.flag_evaluation import (
FlagEvaluationOptions,
FlagResolutionDetails,
FlagType,
FlagValueType,
Reason,
)
from openfeature.hook import Hook, HookContext, HookHints, get_hooks
@ -342,10 +343,12 @@ class OpenFeatureClient:
def get_object_value(
self,
flag_key: str,
default_value: typing.Union[dict, list],
default_value: typing.Union[
Sequence[FlagValueType], typing.Mapping[str, FlagValueType]
],
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> typing.Union[dict, list]:
) -> typing.Union[Sequence[FlagValueType], typing.Mapping[str, FlagValueType]]:
return self.get_object_details(
flag_key,
default_value,
@ -356,10 +359,12 @@ class OpenFeatureClient:
async def get_object_value_async(
self,
flag_key: str,
default_value: typing.Union[dict, list],
default_value: typing.Union[
Sequence[FlagValueType], typing.Mapping[str, FlagValueType]
],
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> typing.Union[dict, list]:
) -> typing.Union[Sequence[FlagValueType], typing.Mapping[str, FlagValueType]]:
details = await self.get_object_details_async(
flag_key,
default_value,
@ -371,10 +376,14 @@ class OpenFeatureClient:
def get_object_details(
self,
flag_key: str,
default_value: typing.Union[dict, list],
default_value: typing.Union[
Sequence[FlagValueType], typing.Mapping[str, FlagValueType]
],
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[typing.Union[dict, list]]:
) -> FlagEvaluationDetails[
typing.Union[Sequence[FlagValueType], typing.Mapping[str, FlagValueType]]
]:
return self.evaluate_flag_details(
FlagType.OBJECT,
flag_key,
@ -386,10 +395,14 @@ class OpenFeatureClient:
async def get_object_details_async(
self,
flag_key: str,
default_value: typing.Union[dict, list],
default_value: typing.Union[
Sequence[FlagValueType], typing.Mapping[str, FlagValueType]
],
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[typing.Union[dict, list]]:
) -> FlagEvaluationDetails[
typing.Union[Sequence[FlagValueType], typing.Mapping[str, FlagValueType]]
]:
return await self.evaluate_flag_details_async(
FlagType.OBJECT,
flag_key,
@ -402,7 +415,7 @@ class OpenFeatureClient:
self,
flag_type: FlagType,
flag_key: str,
default_value: typing.Any,
default_value: FlagValueType,
evaluation_context: typing.Optional[EvaluationContext],
flag_evaluation_options: typing.Optional[FlagEvaluationOptions],
) -> tuple[
@ -479,14 +492,74 @@ class OpenFeatureClient:
)
return merged_context
@typing.overload
async def evaluate_flag_details_async(
self,
flag_type: FlagType,
flag_key: str,
default_value: typing.Any,
default_value: bool,
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[typing.Any]:
) -> FlagEvaluationDetails[bool]: ...
@typing.overload
async def evaluate_flag_details_async(
self,
flag_type: FlagType,
flag_key: str,
default_value: int,
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[int]: ...
@typing.overload
async def evaluate_flag_details_async(
self,
flag_type: FlagType,
flag_key: str,
default_value: float,
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[float]: ...
@typing.overload
async def evaluate_flag_details_async(
self,
flag_type: FlagType,
flag_key: str,
default_value: str,
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[str]: ...
@typing.overload
async def evaluate_flag_details_async(
self,
flag_type: FlagType,
flag_key: str,
default_value: Sequence["FlagValueType"],
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[Sequence["FlagValueType"]]: ...
@typing.overload
async def evaluate_flag_details_async(
self,
flag_type: FlagType,
flag_key: str,
default_value: typing.Mapping[str, "FlagValueType"],
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[typing.Mapping[str, "FlagValueType"]]: ...
async def evaluate_flag_details_async(
self,
flag_type: FlagType,
flag_key: str,
default_value: FlagValueType,
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[FlagValueType]:
"""
Evaluate the flag requested by the user from the clients provider.
@ -595,14 +668,74 @@ class OpenFeatureClient:
hook_hints,
)
@typing.overload
def evaluate_flag_details(
self,
flag_type: FlagType,
flag_key: str,
default_value: typing.Any,
default_value: bool,
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[typing.Any]:
) -> FlagEvaluationDetails[bool]: ...
@typing.overload
def evaluate_flag_details(
self,
flag_type: FlagType,
flag_key: str,
default_value: int,
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[int]: ...
@typing.overload
def evaluate_flag_details(
self,
flag_type: FlagType,
flag_key: str,
default_value: float,
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[float]: ...
@typing.overload
def evaluate_flag_details(
self,
flag_type: FlagType,
flag_key: str,
default_value: str,
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[str]: ...
@typing.overload
def evaluate_flag_details(
self,
flag_type: FlagType,
flag_key: str,
default_value: Sequence["FlagValueType"],
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[Sequence["FlagValueType"]]: ...
@typing.overload
def evaluate_flag_details(
self,
flag_type: FlagType,
flag_key: str,
default_value: typing.Mapping[str, "FlagValueType"],
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[typing.Mapping[str, "FlagValueType"]]: ...
def evaluate_flag_details(
self,
flag_type: FlagType,
flag_key: str,
default_value: FlagValueType,
evaluation_context: typing.Optional[EvaluationContext] = None,
flag_evaluation_options: typing.Optional[FlagEvaluationOptions] = None,
) -> FlagEvaluationDetails[FlagValueType]:
"""
Evaluate the flag requested by the user from the clients provider.
@ -718,9 +851,9 @@ class OpenFeatureClient:
provider: FeatureProvider,
flag_type: FlagType,
flag_key: str,
default_value: typing.Any,
default_value: FlagValueType,
evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagEvaluationDetails[typing.Any]:
) -> FlagEvaluationDetails[FlagValueType]:
get_details_callables_async: typing.Mapping[
FlagType, ResolveDetailsCallableAsync
] = {
@ -765,9 +898,9 @@ class OpenFeatureClient:
provider: FeatureProvider,
flag_type: FlagType,
flag_key: str,
default_value: typing.Any,
default_value: FlagValueType,
evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagEvaluationDetails[typing.Any]:
) -> FlagEvaluationDetails[FlagValueType]:
"""
Encapsulated method to create a FlagEvaluationDetail from a specific provider.

18
openfeature/evaluation_context/__init__.py
View File

@ -1,17 +1,33 @@
from __future__ import annotations
import typing
from collections.abc import Sequence
from dataclasses import dataclass, field
from datetime import datetime
from openfeature.exception import GeneralError
__all__ = ["EvaluationContext", "get_evaluation_context", "set_evaluation_context"]
# https://openfeature.dev/specification/sections/evaluation-context#requirement-312
EvaluationContextAttributes = typing.Mapping[
str,
typing.Union[
bool,
int,
float,
str,
datetime,
Sequence["EvaluationContextAttributes"],
typing.Mapping[str, "EvaluationContextAttributes"],
],
]
@dataclass
class EvaluationContext:
targeting_key: typing.Optional[str] = None
attributes: dict = field(default_factory=dict)
attributes: EvaluationContextAttributes = field(default_factory=dict)
def merge(self, ctx2: EvaluationContext) -> EvaluationContext:
if not (self and ctx2):

11
openfeature/flag_evaluation.py
View File

@ -1,6 +1,7 @@
from __future__ import annotations
import typing
from collections.abc import Sequence
from dataclasses import dataclass, field
from openfeature._backports.strenum import StrEnum
@ -41,7 +42,15 @@ class Reason(StrEnum):
UNKNOWN = "UNKNOWN"
FlagMetadata = typing.Mapping[str, typing.Any]
FlagMetadata = typing.Mapping[str, typing.Union[bool, int, float, str]]
FlagValueType = typing.Union[
bool,
int,
float,
str,
Sequence["FlagValueType"],
typing.Mapping[str, "FlagValueType"],
]
T_co = typing.TypeVar("T_co", covariant=True)

13
openfeature/hook/__init__.py
View File

@ -1,12 +1,13 @@
from __future__ import annotations
import typing
from collections.abc import Sequence
from datetime import datetime
from enum import Enum
from typing import TYPE_CHECKING
from openfeature.evaluation_context import EvaluationContext
from openfeature.flag_evaluation import FlagEvaluationDetails, FlagType
from openfeature.flag_evaluation import FlagEvaluationDetails, FlagType, FlagValueType
if TYPE_CHECKING:
from openfeature.client import ClientMetadata
@ -37,7 +38,7 @@ class HookContext:
self,
flag_key: str,
flag_type: FlagType,
default_value: typing.Any,
default_value: FlagValueType,
evaluation_context: EvaluationContext,
client_metadata: typing.Optional[ClientMetadata] = None,
provider_metadata: typing.Optional[Metadata] = None,
@ -70,8 +71,8 @@ HookHints = typing.Mapping[
float,
str,
datetime,
list[typing.Any],
dict[str, typing.Any],
Sequence["HookHints"],
typing.Mapping[str, "HookHints"],
],
]
@ -94,7 +95,7 @@ class Hook:
def after(
self,
hook_context: HookContext,
details: FlagEvaluationDetails[typing.Any],
details: FlagEvaluationDetails[FlagValueType],
hints: HookHints,
) -> None:
"""
@ -122,7 +123,7 @@ class Hook:
def finally_after(
self,
hook_context: HookContext,
details: FlagEvaluationDetails[typing.Any],
details: FlagEvaluationDetails[FlagValueType],
hints: HookHints,
) -> None:
"""

36
openfeature/provider/__init__.py
View File

@ -2,6 +2,7 @@ from __future__ import annotations
import typing
from abc import abstractmethod
from collections.abc import Sequence
from enum import Enum
from openfeature.evaluation_context import EvaluationContext
@ -11,6 +12,9 @@ from openfeature.hook import Hook
from .metadata import Metadata
if typing.TYPE_CHECKING:
from openfeature.flag_evaluation import FlagValueType
__all__ = ["AbstractProvider", "FeatureProvider", "Metadata", "ProviderStatus"]
@ -99,16 +103,24 @@ class FeatureProvider(typing.Protocol): # pragma: no cover
def resolve_object_details(
self,
flag_key: str,
default_value: typing.Union[dict, list],
default_value: typing.Union[
Sequence[FlagValueType], typing.Mapping[str, FlagValueType]
],
evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[typing.Union[dict, list]]: ...
) -> FlagResolutionDetails[
typing.Union[Sequence[FlagValueType], typing.Mapping[str, FlagValueType]]
]: ...
async def resolve_object_details_async(
self,
flag_key: str,
default_value: typing.Union[dict, list],
default_value: typing.Union[
Sequence[FlagValueType], typing.Mapping[str, FlagValueType]
],
evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[typing.Union[dict, list]]: ...
) -> FlagResolutionDetails[
typing.Union[Sequence[FlagValueType], typing.Mapping[str, FlagValueType]]
]: ...
class AbstractProvider(FeatureProvider):
@ -213,17 +225,25 @@ class AbstractProvider(FeatureProvider):
def resolve_object_details(
self,
flag_key: str,
default_value: typing.Union[dict, list],
default_value: typing.Union[
Sequence[FlagValueType], typing.Mapping[str, FlagValueType]
],
evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[typing.Union[dict, list]]:
) -> FlagResolutionDetails[
typing.Union[Sequence[FlagValueType], typing.Mapping[str, FlagValueType]]
]:
pass
async def resolve_object_details_async(
self,
flag_key: str,
default_value: typing.Union[dict, list],
default_value: typing.Union[
Sequence[FlagValueType], typing.Mapping[str, FlagValueType]
],
evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[typing.Union[dict, list]]:
) -> FlagResolutionDetails[
typing.Union[Sequence[FlagValueType], typing.Mapping[str, FlagValueType]]
]:
return self.resolve_object_details(flag_key, default_value, evaluation_context)
def emit_provider_ready(self, details: ProviderEventDetails) -> None:

28
openfeature/provider/in_memory_provider.py
View File

@ -1,13 +1,19 @@
from __future__ import annotations
import typing
from collections.abc import Sequence
from dataclasses import dataclass, field
from openfeature._backports.strenum import StrEnum
from openfeature.evaluation_context import EvaluationContext
from openfeature.exception import ErrorCode
from openfeature.flag_evaluation import FlagMetadata, FlagResolutionDetails, Reason
from openfeature.hook import Hook
from openfeature.flag_evaluation import FlagResolutionDetails, Reason
from openfeature.provider import AbstractProvider, Metadata
if typing.TYPE_CHECKING:
from openfeature.flag_evaluation import FlagMetadata, FlagValueType
from openfeature.hook import Hook
PASSED_IN_DEFAULT = "Passed in default"
@ -31,7 +37,7 @@ class InMemoryFlag(typing.Generic[T_co]):
state: State = State.ENABLED
context_evaluator: typing.Optional[
typing.Callable[
["InMemoryFlag[T_co]", EvaluationContext], FlagResolutionDetails[T_co]
[InMemoryFlag[T_co], EvaluationContext], FlagResolutionDetails[T_co]
]
] = None
@ -135,17 +141,25 @@ class InMemoryProvider(AbstractProvider):
def resolve_object_details(
self,
flag_key: str,
default_value: typing.Union[dict, list],
default_value: typing.Union[
Sequence[FlagValueType], typing.Mapping[str, FlagValueType]
],
evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[typing.Union[dict, list]]:
) -> FlagResolutionDetails[
typing.Union[Sequence[FlagValueType], typing.Mapping[str, FlagValueType]]
]:
return self._resolve(flag_key, default_value, evaluation_context)
async def resolve_object_details_async(
self,
flag_key: str,
default_value: typing.Union[dict, list],
default_value: typing.Union[
Sequence[FlagValueType], typing.Mapping[str, FlagValueType]
],
evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[typing.Union[dict, list]]:
) -> FlagResolutionDetails[
typing.Union[Sequence[FlagValueType], typing.Mapping[str, FlagValueType]]
]:
return await self._resolve_async(flag_key, default_value, evaluation_context)
def _resolve(

23
openfeature/provider/no_op_provider.py
View File

@ -1,11 +1,18 @@
import typing
from __future__ import annotations
import typing
from collections.abc import Sequence
from openfeature.evaluation_context import EvaluationContext
from openfeature.flag_evaluation import FlagResolutionDetails, Reason
from openfeature.hook import Hook
from openfeature.provider import AbstractProvider, Metadata
from openfeature.provider import AbstractProvider
from openfeature.provider.no_op_metadata import NoOpMetadata
if typing.TYPE_CHECKING:
from openfeature.evaluation_context import EvaluationContext
from openfeature.flag_evaluation import FlagValueType
from openfeature.hook import Hook
from openfeature.provider import Metadata
PASSED_IN_DEFAULT = "Passed in default"
@ -67,9 +74,13 @@ class NoOpProvider(AbstractProvider):
def resolve_object_details(
self,
flag_key: str,
default_value: typing.Union[dict, list],
default_value: typing.Union[
Sequence[FlagValueType], typing.Mapping[str, FlagValueType]
],
evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[typing.Union[dict, list]]:
) -> FlagResolutionDetails[
typing.Union[Sequence[FlagValueType], typing.Mapping[str, FlagValueType]]
]:
return FlagResolutionDetails(
value=default_value,
reason=Reason.DEFAULT,

6
openfeature/telemetry/__init__.py
View File

@ -49,13 +49,13 @@ def create_evaluation_event(
TelemetryFlagMetadata.CONTEXT_ID, hook_context.evaluation_context.targeting_key
)
if context_id:
attributes[TelemetryAttribute.CONTEXT_ID] = context_id
attributes[TelemetryAttribute.CONTEXT_ID] = typing.cast("str", context_id)
if set_id := details.flag_metadata.get(TelemetryFlagMetadata.FLAG_SET_ID):
attributes[TelemetryAttribute.SET_ID] = set_id
attributes[TelemetryAttribute.SET_ID] = typing.cast("str", set_id)
if version := details.flag_metadata.get(TelemetryFlagMetadata.VERSION):
attributes[TelemetryAttribute.VERSION] = version
attributes[TelemetryAttribute.VERSION] = typing.cast("str", version)
if metadata := hook_context.provider_metadata:
attributes[TelemetryAttribute.PROVIDER_NAME] = metadata.name