134 lines
4.3 KiB
Python
134 lines
4.3 KiB
Python
import logging
|
|
import typing
|
|
from functools import reduce
|
|
|
|
from open_feature.evaluation_context.evaluation_context import EvaluationContext
|
|
from open_feature.flag_evaluation.flag_evaluation_details import FlagEvaluationDetails
|
|
from open_feature.flag_evaluation.flag_type import FlagType
|
|
from open_feature.hooks.hook import Hook
|
|
from open_feature.hooks.hook_context import HookContext
|
|
from open_feature.hooks.hook_type import HookType
|
|
|
|
|
|
def error_hooks(
|
|
flag_type: FlagType,
|
|
hook_context: HookContext,
|
|
exception: Exception,
|
|
hooks: typing.List[Hook],
|
|
hints: typing.Optional[typing.Mapping] = None,
|
|
):
|
|
kwargs = {"hook_context": hook_context, "exception": exception, "hints": hints}
|
|
_execute_hooks(
|
|
flag_type=flag_type, hooks=hooks, hook_method=HookType.ERROR, **kwargs
|
|
)
|
|
|
|
|
|
def after_all_hooks(
|
|
flag_type: FlagType,
|
|
hook_context: HookContext,
|
|
hooks: typing.List[Hook],
|
|
hints: typing.Optional[typing.Mapping] = None,
|
|
):
|
|
kwargs = {"hook_context": hook_context, "hints": hints}
|
|
_execute_hooks(
|
|
flag_type=flag_type, hooks=hooks, hook_method=HookType.FINALLY_AFTER, **kwargs
|
|
)
|
|
|
|
|
|
def after_hooks(
|
|
flag_type: FlagType,
|
|
hook_context: HookContext,
|
|
details: FlagEvaluationDetails,
|
|
hooks: typing.List[Hook],
|
|
hints: typing.Optional[typing.Mapping] = None,
|
|
):
|
|
kwargs = {"hook_context": hook_context, "details": details, "hints": hints}
|
|
_execute_hooks_unchecked(
|
|
flag_type=flag_type, hooks=hooks, hook_method=HookType.AFTER, **kwargs
|
|
)
|
|
|
|
|
|
def before_hooks(
|
|
flag_type: FlagType,
|
|
hook_context: HookContext,
|
|
hooks: typing.List[Hook],
|
|
hints: typing.Optional[typing.Mapping] = None,
|
|
) -> EvaluationContext:
|
|
kwargs = {"hook_context": hook_context, "hints": hints}
|
|
executed_hooks = _execute_hooks_unchecked(
|
|
flag_type=flag_type, hooks=hooks, hook_method=HookType.BEFORE, **kwargs
|
|
)
|
|
filtered_hooks = list(filter(lambda hook: hook is not None, executed_hooks))
|
|
|
|
if filtered_hooks:
|
|
return reduce(lambda a, b: a.merge(b), filtered_hooks)
|
|
|
|
return EvaluationContext()
|
|
|
|
|
|
def _execute_hooks(
|
|
flag_type: FlagType, hooks: typing.List[Hook], hook_method: HookType, **kwargs
|
|
) -> list:
|
|
"""
|
|
Run multiple hooks of any hook type. All of these hooks will be run through an
|
|
exception check.
|
|
|
|
:param flag_type: particular type of flag
|
|
:param hooks: a list of hooks
|
|
:param hook_method: the type of hook that is being run
|
|
:param kwargs: arguments that need to be provided to the hook method
|
|
:return: a list of results from the applied hook methods
|
|
"""
|
|
if hooks:
|
|
filtered_hooks = list(
|
|
filter(
|
|
lambda hook: hook.supports_flag_value_type(flag_type=flag_type), hooks
|
|
)
|
|
)
|
|
return [
|
|
_execute_hook_checked(hook, hook_method, **kwargs)
|
|
for hook in filtered_hooks
|
|
]
|
|
return []
|
|
|
|
|
|
def _execute_hooks_unchecked(
|
|
flag_type: FlagType, hooks, hook_method: HookType, **kwargs
|
|
) -> list:
|
|
"""
|
|
Execute a single hook without checking whether an exception is thrown. This is
|
|
used in the before and after hooks since any exception will be caught in the
|
|
client.
|
|
|
|
:param flag_type: particular type of flag
|
|
:param hooks: a list of hooks
|
|
:param hook_method: the type of hook that is being run
|
|
:param kwargs: arguments that need to be provided to the hook method
|
|
:return: a list of results from the applied hook methods
|
|
"""
|
|
if hooks:
|
|
filtered_hooks = list(
|
|
filter(
|
|
lambda hook: hook.supports_flag_value_type(flag_type=flag_type), hooks
|
|
)
|
|
)
|
|
return [getattr(hook, hook_method.value)(**kwargs) for hook in filtered_hooks]
|
|
|
|
return []
|
|
|
|
|
|
def _execute_hook_checked(hook: Hook, hook_method: HookType, **kwargs):
|
|
"""
|
|
Try and run a single hook and catch any exception thrown. This is used in the
|
|
after all and error hooks since any error thrown at this point needs to be caught.
|
|
|
|
:param hook: a list of hooks
|
|
:param hook_method: the type of hook that is being run
|
|
:param kwargs: arguments that need to be provided to the hook method
|
|
:return: the result of the hook method
|
|
"""
|
|
try:
|
|
return getattr(hook, hook_method.value)(**kwargs)
|
|
except Exception: # noqa
|
|
logging.error(f"Exception when running {hook_method.value} hooks")
|