python-sdk/openfeature/api.py at e4aa328d77c564cc5636f65aa54151f84a96b598 · open-feature/python-sdk · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
import typing

from openfeature import _event_support
from openfeature.client import OpenFeatureClient
from openfeature.evaluation_context import EvaluationContext
from openfeature.event import (
    EventHandler,
    ProviderEvent,
)
from openfeature.exception import GeneralError
from openfeature.hook import Hook
from openfeature.provider import FeatureProvider
from openfeature.provider._registry import provider_registry
from openfeature.provider.metadata import Metadata
from openfeature.transaction_context import TransactionContextPropagator
from openfeature.transaction_context.no_op_transaction_context_propagator import (
    NoOpTransactionContextPropagator,
)

__all__ = [
    "add_handler",
    "add_hooks",
    "clear_hooks",
    "clear_providers",
    "get_client",
    "get_evaluation_context",
    "get_hooks",
    "get_provider_metadata",
    "get_transaction_context",
    "remove_handler",
    "set_evaluation_context",
    "set_provider",
    "set_transaction_context",
    "set_transaction_context_propagator",
    "shutdown",
]

_evaluation_context = EvaluationContext()
_evaluation_transaction_context_propagator: TransactionContextPropagator = (
    NoOpTransactionContextPropagator()
)

_hooks: list[Hook] = []


def get_client(
    domain: typing.Optional[str] = None, version: typing.Optional[str] = None
) -> OpenFeatureClient:
    return OpenFeatureClient(domain=domain, version=version)


def set_provider(
    provider: FeatureProvider, domain: typing.Optional[str] = None
) -> None:
    if domain is None:
        provider_registry.set_default_provider(provider)
    else:
        provider_registry.set_provider(domain, provider)


def clear_providers() -> None:
    provider_registry.clear_providers()
    _event_support.clear()


def get_provider_metadata(domain: typing.Optional[str] = None) -> Metadata:
    return provider_registry.get_provider(domain).get_metadata()


def get_evaluation_context() -> EvaluationContext:
    return _evaluation_context


def set_evaluation_context(evaluation_context: EvaluationContext) -> None:
    global _evaluation_context
    if evaluation_context is None:
        raise GeneralError(error_message="No api level evaluation context")
    _evaluation_context = evaluation_context


def set_transaction_context_propagator(
    transaction_context_propagator: TransactionContextPropagator,
) -> None:
    global _evaluation_transaction_context_propagator
    _evaluation_transaction_context_propagator = transaction_context_propagator


def get_transaction_context() -> EvaluationContext:
    return _evaluation_transaction_context_propagator.get_transaction_context()


def set_transaction_context(evaluation_context: EvaluationContext) -> None:
    global _evaluation_transaction_context_propagator
    _evaluation_transaction_context_propagator.set_transaction_context(
        evaluation_context
    )


def add_hooks(hooks: list[Hook]) -> None:
    global _hooks
    _hooks = _hooks + hooks


def clear_hooks() -> None:
    global _hooks
    _hooks = []


def get_hooks() -> list[Hook]:
    return _hooks


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


def add_handler(event: ProviderEvent, handler: EventHandler) -> None:
    _event_support.add_global_handler(event, handler)


def remove_handler(event: ProviderEvent, handler: EventHandler) -> None:
    _event_support.remove_global_handler(event, handler)