Source code for stamina.instrumentation._hooks

# SPDX-FileCopyrightText: 2022 Hynek Schlawack <>
# SPDX-License-Identifier: MIT

from __future__ import annotations

from typing import Iterable

from ._data import RetryHook, RetryHookFactory
from ._logging import LoggingOnRetryHook
from ._prometheus import PrometheusOnRetryHook
from ._structlog import StructlogOnRetryHook

def init_hooks(
    maybe_delayed: tuple[RetryHook | RetryHookFactory, ...]
) -> tuple[RetryHook, ...]:
    Execute delayed hook factories and return a tuple of finalized hooks.
    hooks = []
    for hook_or_factory in maybe_delayed:
        if isinstance(hook_or_factory, RetryHookFactory):

    return tuple(hooks)

def get_default_hooks() -> tuple[RetryHookFactory, ...]:
    Return the default hooks according to availability.
    hooks = []

        import prometheus_client  # noqa: F401

    except ImportError:

        import structlog  # noqa: F401

    except ImportError:

    return tuple(hooks)

[docs] def set_on_retry_hooks( hooks: Iterable[RetryHook | RetryHookFactory] | None, ) -> None: """ Set hooks that are called after a retry has been scheduled. Args: hooks: Hooks to call after a retry has been scheduled. Passing None resets to default. To deactivate instrumentation, pass an empty iterable. .. versionadded:: 23.2.0 """ from .._config import CONFIG CONFIG.on_retry = tuple(hooks) if hooks is not None else hooks # type: ignore[assignment,arg-type]
[docs] def get_on_retry_hooks() -> tuple[RetryHook, ...]: """ Get hooks that are called after a retry has been scheduled. Returns: Hooks that will run if a retry is scheduled. Factories are called if they haven't already. .. versionadded:: 23.2.0 """ from .._config import CONFIG return CONFIG.on_retry