76 lines
2.4 KiB
Python
76 lines
2.4 KiB
Python
import typing
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
|
|
from openfeature.exception import ErrorCode
|
|
from openfeature.flag_evaluation import FlagEvaluationDetails, Reason
|
|
from openfeature.hook import HookContext
|
|
from openfeature.telemetry.attributes import TelemetryAttribute
|
|
from openfeature.telemetry.body import TelemetryBodyField
|
|
from openfeature.telemetry.metadata import TelemetryFlagMetadata
|
|
|
|
__all__ = [
|
|
"EvaluationEvent",
|
|
"TelemetryAttribute",
|
|
"TelemetryBodyField",
|
|
"TelemetryFlagMetadata",
|
|
"create_evaluation_event",
|
|
]
|
|
|
|
FLAG_EVALUATION_EVENT_NAME = "feature_flag.evaluation"
|
|
|
|
T_co = typing.TypeVar("T_co", covariant=True)
|
|
|
|
|
|
@dataclass
|
|
class EvaluationEvent(typing.Generic[T_co]):
|
|
name: str
|
|
attributes: Mapping[TelemetryAttribute, typing.Union[str, T_co]]
|
|
body: Mapping[TelemetryBodyField, T_co]
|
|
|
|
|
|
def create_evaluation_event(
|
|
hook_context: HookContext, details: FlagEvaluationDetails[T_co]
|
|
) -> EvaluationEvent[T_co]:
|
|
attributes = {
|
|
TelemetryAttribute.KEY: details.flag_key,
|
|
TelemetryAttribute.EVALUATION_REASON: (
|
|
details.reason or Reason.UNKNOWN
|
|
).lower(),
|
|
}
|
|
body = {}
|
|
|
|
if variant := details.variant:
|
|
attributes[TelemetryAttribute.VARIANT] = variant
|
|
else:
|
|
body[TelemetryBodyField.VALUE] = details.value
|
|
|
|
context_id = details.flag_metadata.get(
|
|
TelemetryFlagMetadata.CONTEXT_ID, hook_context.evaluation_context.targeting_key
|
|
)
|
|
if 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] = typing.cast("str", set_id)
|
|
|
|
if version := details.flag_metadata.get(TelemetryFlagMetadata.VERSION):
|
|
attributes[TelemetryAttribute.VERSION] = typing.cast("str", version)
|
|
|
|
if metadata := hook_context.provider_metadata:
|
|
attributes[TelemetryAttribute.PROVIDER_NAME] = metadata.name
|
|
|
|
if details.reason == Reason.ERROR:
|
|
attributes[TelemetryAttribute.ERROR_TYPE] = (
|
|
details.error_code or ErrorCode.GENERAL
|
|
).lower()
|
|
|
|
if err_msg := details.error_message:
|
|
attributes[TelemetryAttribute.EVALUATION_ERROR_MESSAGE] = err_msg
|
|
|
|
return EvaluationEvent(
|
|
name=FLAG_EVALUATION_EVENT_NAME,
|
|
attributes=attributes,
|
|
body=body,
|
|
)
|