`tempestweb.observability`¶

Provedores no padrão adapter para o que um app em produção precisa e não quer acoplar: telemetria, logging estruturado, error boundary, feature flags e autenticação. Cada um tem uma implementação padrão e aceita a sua.

Guia com exemplos: Observabilidade.

tempestweb.observability ¶

Trilho O — production / observability providers (adapter pattern).

Every provider here follows the same shape: a tiny, stable interface application code calls, plus one or more swappable adapters behind it. Changing the backend (console -> Sentry, in-memory flags -> LaunchDarkly, ...) never touches a call site.

Modules

* :mod:`telemetry` (O0) — ``track`` / ``identify`` with console/Sentry/PostHog
  adapters.
* :mod:`logger` (O1) — structured logging with pluggable sinks and typed
  levels.
* :mod:`error_boundary` (O2) — render-error fallback widget/decorator plus a
  report hook into telemetry.
* :mod:`feature_flags` (O3) — runtime toggles with in-memory / GrowthBook /
  LaunchDarkly adapters.
* :mod:`auth` (O4) — token store, route guard, JWT helpers and a refresh queue
  that serializes concurrent renewals.

Import everything from this package level rather than from submodules.

AuthState ¶

A snapshot of the current authentication state.

Attributes:

Name	Type	Description
`token`	`str \| None`	The current access token, or `None` when logged out.
`user`	`dict[str, Any] \| None`	The current user payload, or `None` when logged out.

Source code in tempestweb/observability/auth.py

class AuthState:
    """A snapshot of the current authentication state.

    Attributes:
        token: The current access token, or ``None`` when logged out.
        user: The current user payload, or ``None`` when logged out.
    """

    def __init__(self, token: str | None, user: dict[str, Any] | None) -> None:
        """Initialize the snapshot.

        Args:
            token: The current access token, or ``None``.
            user: The current user payload, or ``None``.
        """
        self.token: str | None = token
        self.user: dict[str, Any] | None = user

AuthStore ¶

An observable store of the current token and user.

Mutations (login / logout / set_token) notify subscribers, which is how an auth change drives a re-render (e.g. swapping a login screen for the app). The store holds no refresh logic itself — pair it with a :class:RefreshQueue for that.

Source code in tempestweb/observability/auth.py

class AuthStore:
    """An observable store of the current token and user.

    Mutations (``login`` / ``logout`` / ``set_token``) notify subscribers, which
    is how an auth change drives a re-render (e.g. swapping a login screen for the
    app). The store holds no refresh logic itself — pair it with a
    :class:`RefreshQueue` for that.
    """

    def __init__(self) -> None:
        """Initialize an empty (logged-out) store."""
        self._token: str | None = None
        self._user: dict[str, Any] | None = None
        self._listeners: list[AuthListener] = []

    @property
    def token(self) -> str | None:
        """The current access token, or ``None`` when logged out.

        Returns:
            The token, or ``None``.
        """
        return self._token

    @property
    def user(self) -> dict[str, Any] | None:
        """The current user payload, or ``None`` when logged out.

        Returns:
            The user payload, or ``None``.
        """
        return self._user

    @property
    def is_authenticated(self) -> bool:
        """Whether a token is currently present.

        Returns:
            ``True`` if a token is set, ``False`` otherwise.
        """
        return self._token is not None

    @property
    def state(self) -> AuthState:
        """An immutable snapshot of the current state.

        Returns:
            An :class:`AuthState` capturing token and user.
        """
        return AuthState(self._token, self._user)

    def login(self, token: str, user: dict[str, Any] | None = None) -> None:
        """Set the token (and optional user) and notify subscribers.

        Args:
            token: The access token to store.
            user: The user payload to store, if known.

        Returns:
            None.
        """
        self._token = token
        self._user = dict(user) if user is not None else None
        self._notify()

    def set_token(self, token: str) -> None:
        """Replace the access token (e.g. after a refresh) and notify.

        Args:
            token: The new access token.

        Returns:
            None.
        """
        self._token = token
        self._notify()

    def logout(self) -> None:
        """Clear the token and user and notify subscribers.

        Returns:
            None.
        """
        self._token = None
        self._user = None
        self._notify()

    def subscribe(self, listener: AuthListener) -> Callable[[], None]:
        """Register a listener fired on every auth change.

        Args:
            listener: A zero-argument callback invoked on change.

        Returns:
            An unsubscribe callable.
        """
        self._listeners.append(listener)

        def unsubscribe() -> None:
            """Remove the listener if still registered."""
            if listener in self._listeners:
                self._listeners.remove(listener)

        return unsubscribe

    def _notify(self) -> None:
        """Notify every subscriber of a state change.

        Returns:
            None.
        """
        for listener in list(self._listeners):
            listener()

token `property` ¶

token: str | None

The current access token, or None when logged out.

Returns:

Type	Description
`str \| None`	The token, or `None`.

user `property` ¶

user: dict[str, Any] | None

The current user payload, or None when logged out.

Returns:

Type	Description
`dict[str, Any] \| None`	The user payload, or `None`.

is_authenticated `property` ¶

is_authenticated: bool

Whether a token is currently present.

Returns:

Type	Description
`bool`	`True` if a token is set, `False` otherwise.

state `property` ¶

state: AuthState

An immutable snapshot of the current state.

Returns:

Name	Type	Description
`An`	`AuthState`	class:`AuthState` capturing token and user.

login(token: str, user: dict[str, Any] | None = None) -> None

Set the token (and optional user) and notify subscribers.

Parameters:

Name	Type	Description	Default
`token`	`str`	The access token to store.	required
`user`	`dict[str, Any] \| None`	The user payload to store, if known.	`None`

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/auth.py

def login(self, token: str, user: dict[str, Any] | None = None) -> None:
    """Set the token (and optional user) and notify subscribers.

    Args:
        token: The access token to store.
        user: The user payload to store, if known.

    Returns:
        None.
    """
    self._token = token
    self._user = dict(user) if user is not None else None
    self._notify()

set_token ¶

set_token(token: str) -> None

Replace the access token (e.g. after a refresh) and notify.

Parameters:

Name	Type	Description	Default
`token`	`str`	The new access token.	required

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/auth.py

def set_token(self, token: str) -> None:
    """Replace the access token (e.g. after a refresh) and notify.

    Args:
        token: The new access token.

    Returns:
        None.
    """
    self._token = token
    self._notify()

logout ¶

logout() -> None

Clear the token and user and notify subscribers.

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/auth.py

def logout(self) -> None:
    """Clear the token and user and notify subscribers.

    Returns:
        None.
    """
    self._token = None
    self._user = None
    self._notify()

subscribe(listener: AuthListener) -> Callable[[], None]

Register a listener fired on every auth change.

Parameters:

Name	Type	Description	Default
`listener`	`AuthListener`	A zero-argument callback invoked on change.	required

Returns:

Type	Description
`Callable[[], None]`	An unsubscribe callable.

Source code in tempestweb/observability/auth.py

def subscribe(self, listener: AuthListener) -> Callable[[], None]:
    """Register a listener fired on every auth change.

    Args:
        listener: A zero-argument callback invoked on change.

    Returns:
        An unsubscribe callable.
    """
    self._listeners.append(listener)

    def unsubscribe() -> None:
        """Remove the listener if still registered."""
        if listener in self._listeners:
            self._listeners.remove(listener)

    return unsubscribe

JWTError ¶

Bases: ValueError

Raised when a JWT cannot be parsed into a claims payload.

Source code in tempestweb/observability/auth.py

class JWTError(ValueError):
    """Raised when a JWT cannot be parsed into a claims payload."""

RefreshQueue ¶

Serializes concurrent token refreshes into a single in-flight renewal.

When a token expires, many requests can discover it at once and each try to refresh. Without coordination that fires N parallel renewals, races the store, and can invalidate each other's refresh tokens. This queue ensures exactly one refresh runs: the first caller starts it, every concurrent caller awaits the same result, and the new token is pushed into the store once. After it settles the queue resets so a later expiry refreshes again.

Source code in tempestweb/observability/auth.py

class RefreshQueue:
    """Serializes concurrent token refreshes into a single in-flight renewal.

    When a token expires, many requests can discover it at once and each try to
    refresh. Without coordination that fires N parallel renewals, races the
    store, and can invalidate each other's refresh tokens. This queue ensures
    **exactly one** refresh runs: the first caller starts it, every concurrent
    caller awaits the same result, and the new token is pushed into the store
    once. After it settles the queue resets so a later expiry refreshes again.
    """

    def __init__(self, store: AuthStore, refresh_fn: RefreshFn) -> None:
        """Initialize the queue.

        Args:
            store: The auth store whose token is updated after a refresh.
            refresh_fn: An async function performing the renewal and returning
                the new access token.
        """
        self._store: AuthStore = store
        self._refresh_fn: RefreshFn = refresh_fn
        self._pending: asyncio.Future[str] | None = None
        self._calls: int = 0

    @property
    def refresh_calls(self) -> int:
        """The number of times the underlying ``refresh_fn`` was actually run.

        Useful in tests to assert that concurrent callers collapsed into a single
        renewal.

        Returns:
            The count of real refresh invocations.
        """
        return self._calls

    async def refresh(self) -> str:
        """Return a fresh token, coalescing concurrent callers into one renewal.

        The first caller schedules the real renewal as a single
        :class:`asyncio.Task` and stores it; every concurrent caller awaits that
        same task instead of starting its own. The task resolves once for all
        waiters, then the in-flight slot is cleared so a future expiry triggers a
        new renewal. If the renewal raises, the exception propagates to every
        waiter and the slot is cleared so a retry is possible.

        Returns:
            The new access token.

        Raises:
            Exception: Whatever ``refresh_fn`` raises, propagated to all waiters.
        """
        if self._pending is not None:
            return await self._pending

        task: asyncio.Task[str] = asyncio.ensure_future(self._run())
        self._pending = task
        try:
            return await task
        finally:
            self._pending = None

    async def _run(self) -> str:
        """Run the real refresh once and push the new token into the store.

        Returns:
            The new access token.
        """
        self._calls += 1
        token: str = await self._refresh_fn()
        self._store.set_token(token)
        return token

refresh_calls `property` ¶

refresh_calls: int

The number of times the underlying refresh_fn was actually run.

Useful in tests to assert that concurrent callers collapsed into a single renewal.

Returns:

Type	Description
`int`	The count of real refresh invocations.

refresh `async` ¶

refresh() -> str

Return a fresh token, coalescing concurrent callers into one renewal.

The first caller schedules the real renewal as a single :class:asyncio.Task and stores it; every concurrent caller awaits that same task instead of starting its own. The task resolves once for all waiters, then the in-flight slot is cleared so a future expiry triggers a new renewal. If the renewal raises, the exception propagates to every waiter and the slot is cleared so a retry is possible.

Returns:

Type	Description
`str`	The new access token.

Raises:

Type	Description
`Exception`	Whatever `refresh_fn` raises, propagated to all waiters.

Source code in tempestweb/observability/auth.py

async def refresh(self) -> str:
    """Return a fresh token, coalescing concurrent callers into one renewal.

    The first caller schedules the real renewal as a single
    :class:`asyncio.Task` and stores it; every concurrent caller awaits that
    same task instead of starting its own. The task resolves once for all
    waiters, then the in-flight slot is cleared so a future expiry triggers a
    new renewal. If the renewal raises, the exception propagates to every
    waiter and the slot is cleared so a retry is possible.

    Returns:
        The new access token.

    Raises:
        Exception: Whatever ``refresh_fn`` raises, propagated to all waiters.
    """
    if self._pending is not None:
        return await self._pending

    task: asyncio.Task[str] = asyncio.ensure_future(self._run())
    self._pending = task
    try:
        return await task
    finally:
        self._pending = None

ErrorBoundary ¶

Bases: Component

A component that contains a render error in its wrapped subtree.

On :meth:render it invokes child_builder. If that returns a widget, the widget is rendered unchanged. If it raises, the boundary captures the error into an :class:ErrorInfo, calls on_error (if set) for reporting, and returns fallback_builder(info) instead — so the exception never escapes and the surrounding tree keeps rendering.

Source code in tempestweb/observability/error_boundary.py

class ErrorBoundary(Component):
    """A component that contains a render error in its wrapped subtree.

    On :meth:`render` it invokes ``child_builder``. If that returns a widget, the
    widget is rendered unchanged. If it raises, the boundary captures the error
    into an :class:`ErrorInfo`, calls ``on_error`` (if set) for reporting, and
    returns ``fallback_builder(info)`` instead — so the exception never escapes
    and the surrounding tree keeps rendering.
    """

    # Component is a Pydantic model; callable fields are stored as-is because the
    # core's Widget already sets ``arbitrary_types_allowed`` in its model_config.
    child_builder: ChildBuilder = Field(
        description="Builds the protected subtree; may raise during render."
    )
    fallback_builder: FallbackBuilder = Field(
        default=default_fallback,
        description="Builds the fallback subtree from the captured error.",
    )
    on_error: ErrorReporter | None = Field(
        default=None,
        description="Optional hook invoked with the captured error for reporting.",
    )

    def render(self) -> Widget:
        """Render the protected subtree, falling back on any render error.

        Returns:
            The child's widget on success, or the fallback widget on failure.
        """
        try:
            return self.child_builder()
        except Exception as exc:  # noqa: BLE001 - boundary intentionally broad
            info: ErrorInfo = ErrorInfo.from_exception(exc)
            if self.on_error is not None:
                self.on_error(info)
            return self.fallback_builder(info)

render ¶

render() -> Widget

Render the protected subtree, falling back on any render error.

Returns:

Type	Description
`Widget`	The child's widget on success, or the fallback widget on failure.

Source code in tempestweb/observability/error_boundary.py

def render(self) -> Widget:
    """Render the protected subtree, falling back on any render error.

    Returns:
        The child's widget on success, or the fallback widget on failure.
    """
    try:
        return self.child_builder()
    except Exception as exc:  # noqa: BLE001 - boundary intentionally broad
        info: ErrorInfo = ErrorInfo.from_exception(exc)
        if self.on_error is not None:
            self.on_error(info)
        return self.fallback_builder(info)

ErrorInfo `dataclass` ¶

A captured render failure, passed to the fallback and report hooks.

Attributes:

Name	Type	Description
`error`	`BaseException`	The exception instance that was raised during render.
`error_type`	`str`	The exception class name (e.g. `"ValueError"`).
`message`	`str`	The exception's string message.
`stack`	`str`	The formatted traceback, preserved for reporting rather than being swallowed.

Source code in tempestweb/observability/error_boundary.py

@dataclass(frozen=True)
class ErrorInfo:
    """A captured render failure, passed to the fallback and report hooks.

    Attributes:
        error: The exception instance that was raised during render.
        error_type: The exception class name (e.g. ``"ValueError"``).
        message: The exception's string message.
        stack: The formatted traceback, preserved for reporting rather than
            being swallowed.
    """

    error: BaseException
    error_type: str
    message: str
    stack: str

    @classmethod
    def from_exception(cls, error: BaseException) -> ErrorInfo:
        """Build an :class:`ErrorInfo` from a raised exception.

        Args:
            error: The exception caught during render.

        Returns:
            A populated :class:`ErrorInfo` capturing type, message and stack.
        """
        stack: str = "".join(
            traceback.format_exception(type(error), error, error.__traceback__)
        )
        return cls(
            error=error,
            error_type=type(error).__name__,
            message=str(error),
            stack=stack,
        )

from_exception `classmethod` ¶

from_exception(error: BaseException) -> ErrorInfo

Build an :class:ErrorInfo from a raised exception.

Parameters:

Name	Type	Description	Default
`error`	`BaseException`	The exception caught during render.	required

Returns:

Type	Description
`ErrorInfo`	A populated :class:`ErrorInfo` capturing type, message and stack.

Source code in tempestweb/observability/error_boundary.py

@classmethod
def from_exception(cls, error: BaseException) -> ErrorInfo:
    """Build an :class:`ErrorInfo` from a raised exception.

    Args:
        error: The exception caught during render.

    Returns:
        A populated :class:`ErrorInfo` capturing type, message and stack.
    """
    stack: str = "".join(
        traceback.format_exception(type(error), error, error.__traceback__)
    )
    return cls(
        error=error,
        error_type=type(error).__name__,
        message=str(error),
        stack=stack,
    )

FeatureFlagsAdapter ¶

Bases: Protocol

The minimal contract every feature-flag backend must satisfy.

The interface is intentionally tiny (roughly twenty lines to implement): fetch a value, and register a change subscription. get must never raise for an unknown key — it returns the provided default — so the provider can stay fail-safe.

Source code in tempestweb/observability/feature_flags.py

@runtime_checkable
class FeatureFlagsAdapter(Protocol):
    """The minimal contract every feature-flag backend must satisfy.

    The interface is intentionally tiny (roughly twenty lines to implement):
    fetch a value, and register a change subscription. ``get`` must never raise
    for an unknown key — it returns the provided ``default`` — so the provider
    can stay fail-safe.
    """

    def get(self, key: str, default: FlagValue = None) -> FlagValue:
        """Return the value of a flag, or ``default`` if unknown.

        Args:
            key: The flag key.
            default: The value to return when the flag is not present.

        Returns:
            The flag value, or ``default`` when unknown.
        """
        ...

    def subscribe(self, listener: ChangeListener) -> Callable[[], None]:
        """Register a listener fired whenever any flag changes.

        Args:
            listener: A zero-argument callback invoked on change.

        Returns:
            An unsubscribe callable that removes the listener.
        """
        ...

get ¶

get(key: str, default: FlagValue = None) -> FlagValue

Return the value of a flag, or default if unknown.

Parameters:

Name	Type	Description	Default
`key`	`str`	The flag key.	required
`default`	`FlagValue`	The value to return when the flag is not present.	`None`

Returns:

Type	Description
`FlagValue`	The flag value, or `default` when unknown.

Source code in tempestweb/observability/feature_flags.py

def get(self, key: str, default: FlagValue = None) -> FlagValue:
    """Return the value of a flag, or ``default`` if unknown.

    Args:
        key: The flag key.
        default: The value to return when the flag is not present.

    Returns:
        The flag value, or ``default`` when unknown.
    """
    ...

subscribe(listener: ChangeListener) -> Callable[[], None]

Register a listener fired whenever any flag changes.

Parameters:

Name	Type	Description	Default
`listener`	`ChangeListener`	A zero-argument callback invoked on change.	required

Returns:

Type	Description
`Callable[[], None]`	An unsubscribe callable that removes the listener.

Source code in tempestweb/observability/feature_flags.py

def subscribe(self, listener: ChangeListener) -> Callable[[], None]:
    """Register a listener fired whenever any flag changes.

    Args:
        listener: A zero-argument callback invoked on change.

    Returns:
        An unsubscribe callable that removes the listener.
    """
    ...

FeatureFlagsProvider ¶

A backend-agnostic facade application code calls to read flags.

The provider forwards reads to its :class:FeatureFlagsAdapter and fans the adapter's change notifications out to its own subscribers. is_enabled coerces any value to a boolean so a gate check is uniform regardless of the underlying value type. Swapping the adapter changes the flag source while leaving every is_enabled / get / on_change call untouched.

Source code in tempestweb/observability/feature_flags.py

class FeatureFlagsProvider:
    """A backend-agnostic facade application code calls to read flags.

    The provider forwards reads to its :class:`FeatureFlagsAdapter` and fans the
    adapter's change notifications out to its own subscribers. ``is_enabled``
    coerces any value to a boolean so a gate check is uniform regardless of the
    underlying value type. Swapping the adapter changes the flag source while
    leaving every ``is_enabled`` / ``get`` / ``on_change`` call untouched.
    """

    def __init__(self, adapter: FeatureFlagsAdapter) -> None:
        """Initialize the provider and bridge the adapter's change stream.

        Args:
            adapter: The flag backend to read from and subscribe to.
        """
        self._adapter: FeatureFlagsAdapter = adapter
        self._listeners: list[ChangeListener] = []
        # Bridge the adapter's single change stream to our fan-out so callers
        # subscribe to the provider, not the concrete adapter.
        self._adapter.subscribe(self._notify)

    @property
    def adapter(self) -> FeatureFlagsAdapter:
        """The adapter currently backing this provider.

        Returns:
            The active :class:`FeatureFlagsAdapter`.
        """
        return self._adapter

    def get(self, key: str, default: FlagValue = None) -> FlagValue:
        """Return a flag's value, or ``default`` when unknown.

        Args:
            key: The flag key.
            default: The value returned when the flag is absent.

        Returns:
            The flag value, or ``default``.
        """
        return self._adapter.get(key, default)

    def is_enabled(self, key: str, *, default: bool = False) -> bool:
        """Return whether a flag is truthy, defaulting safely when unknown.

        Args:
            key: The flag key.
            default: The boolean returned when the flag is absent.

        Returns:
            ``bool(value)`` for a present flag, otherwise ``default``.
        """
        value: FlagValue = self._adapter.get(key, default)
        return bool(value)

    def on_change(self, listener: ChangeListener) -> Callable[[], None]:
        """Register a listener fired whenever any flag changes.

        Args:
            listener: A zero-argument callback invoked on change.

        Returns:
            An unsubscribe callable that removes ``listener``.
        """
        self._listeners.append(listener)

        def unsubscribe() -> None:
            """Remove the registered listener if still present."""
            if listener in self._listeners:
                self._listeners.remove(listener)

        return unsubscribe

    def _notify(self) -> None:
        """Fan a change notification out to every registered listener.

        Returns:
            None.
        """
        for listener in list(self._listeners):
            listener()

adapter `property` ¶

adapter: FeatureFlagsAdapter

The adapter currently backing this provider.

Returns:

Type	Description
`FeatureFlagsAdapter`	The active :class:`FeatureFlagsAdapter`.

get ¶

get(key: str, default: FlagValue = None) -> FlagValue

Return a flag's value, or default when unknown.

Parameters:

Name	Type	Description	Default
`key`	`str`	The flag key.	required
`default`	`FlagValue`	The value returned when the flag is absent.	`None`

Returns:

Type	Description
`FlagValue`	The flag value, or `default`.

Source code in tempestweb/observability/feature_flags.py

def get(self, key: str, default: FlagValue = None) -> FlagValue:
    """Return a flag's value, or ``default`` when unknown.

    Args:
        key: The flag key.
        default: The value returned when the flag is absent.

    Returns:
        The flag value, or ``default``.
    """
    return self._adapter.get(key, default)

is_enabled ¶

is_enabled(key: str, *, default: bool = False) -> bool

Return whether a flag is truthy, defaulting safely when unknown.

Parameters:

Name	Type	Description	Default
`key`	`str`	The flag key.	required
`default`	`bool`	The boolean returned when the flag is absent.	`False`

Returns:

Type	Description
`bool`	`bool(value)` for a present flag, otherwise `default`.

Source code in tempestweb/observability/feature_flags.py

def is_enabled(self, key: str, *, default: bool = False) -> bool:
    """Return whether a flag is truthy, defaulting safely when unknown.

    Args:
        key: The flag key.
        default: The boolean returned when the flag is absent.

    Returns:
        ``bool(value)`` for a present flag, otherwise ``default``.
    """
    value: FlagValue = self._adapter.get(key, default)
    return bool(value)

on_change ¶

on_change(listener: ChangeListener) -> Callable[[], None]

Register a listener fired whenever any flag changes.

Parameters:

Name	Type	Description	Default
`listener`	`ChangeListener`	A zero-argument callback invoked on change.	required

Returns:

Type	Description
`Callable[[], None]`	An unsubscribe callable that removes `listener`.

Source code in tempestweb/observability/feature_flags.py

def on_change(self, listener: ChangeListener) -> Callable[[], None]:
    """Register a listener fired whenever any flag changes.

    Args:
        listener: A zero-argument callback invoked on change.

    Returns:
        An unsubscribe callable that removes ``listener``.
    """
    self._listeners.append(listener)

    def unsubscribe() -> None:
        """Remove the registered listener if still present."""
        if listener in self._listeners:
            self._listeners.remove(listener)

    return unsubscribe

GrowthBookFeatureFlagsAdapter ¶

An adapter that maps flag reads onto an injected GrowthBook instance.

growthbook is not a tempestweb dependency; the caller injects a client exposing is_on(key) / get_feature_value(key, default). GrowthBook does not push change events in this minimal wrapper, so :meth:refresh re-evaluates and notifies subscribers after the caller reloads features.

Source code in tempestweb/observability/feature_flags.py

class GrowthBookFeatureFlagsAdapter:
    """An adapter that maps flag reads onto an injected GrowthBook instance.

    ``growthbook`` is not a tempestweb dependency; the caller injects a client
    exposing ``is_on(key)`` / ``get_feature_value(key, default)``. GrowthBook
    does not push change events in this minimal wrapper, so :meth:`refresh`
    re-evaluates and notifies subscribers after the caller reloads features.
    """

    def __init__(self, client: Any) -> None:  # noqa: ANN401 - injected third-party GrowthBook client
        """Initialize the adapter.

        Args:
            client: A GrowthBook-compatible client exposing ``is_on(key)`` and
                ``get_feature_value(key, default)``.
        """
        self._client: Any = client
        self._listeners: list[ChangeListener] = []

    def get(self, key: str, default: FlagValue = None) -> FlagValue:
        """Return a feature value from GrowthBook.

        Args:
            key: The feature key.
            default: The value returned when the feature is absent.

        Returns:
            The feature value, or ``default``.
        """
        value: FlagValue = self._client.get_feature_value(key, default)
        return value

    def subscribe(self, listener: ChangeListener) -> Callable[[], None]:
        """Register a change listener fired by :meth:`refresh`.

        Args:
            listener: A zero-argument callback invoked on change.

        Returns:
            An unsubscribe callable.
        """
        self._listeners.append(listener)

        def unsubscribe() -> None:
            """Remove the listener if still registered."""
            if listener in self._listeners:
                self._listeners.remove(listener)

        return unsubscribe

    def refresh(self) -> None:
        """Notify subscribers after the caller reloads GrowthBook features.

        Returns:
            None.
        """
        for listener in list(self._listeners):
            listener()

get ¶

get(key: str, default: FlagValue = None) -> FlagValue

Return a feature value from GrowthBook.

Parameters:

Name	Type	Description	Default
`key`	`str`	The feature key.	required
`default`	`FlagValue`	The value returned when the feature is absent.	`None`

Returns:

Type	Description
`FlagValue`	The feature value, or `default`.

Source code in tempestweb/observability/feature_flags.py

def get(self, key: str, default: FlagValue = None) -> FlagValue:
    """Return a feature value from GrowthBook.

    Args:
        key: The feature key.
        default: The value returned when the feature is absent.

    Returns:
        The feature value, or ``default``.
    """
    value: FlagValue = self._client.get_feature_value(key, default)
    return value

subscribe(listener: ChangeListener) -> Callable[[], None]

Register a change listener fired by :meth:refresh.

Parameters:

Name	Type	Description	Default
`listener`	`ChangeListener`	A zero-argument callback invoked on change.	required

Returns:

Type	Description
`Callable[[], None]`	An unsubscribe callable.

Source code in tempestweb/observability/feature_flags.py

def subscribe(self, listener: ChangeListener) -> Callable[[], None]:
    """Register a change listener fired by :meth:`refresh`.

    Args:
        listener: A zero-argument callback invoked on change.

    Returns:
        An unsubscribe callable.
    """
    self._listeners.append(listener)

    def unsubscribe() -> None:
        """Remove the listener if still registered."""
        if listener in self._listeners:
            self._listeners.remove(listener)

    return unsubscribe

refresh ¶

refresh() -> None

Notify subscribers after the caller reloads GrowthBook features.

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/feature_flags.py

def refresh(self) -> None:
    """Notify subscribers after the caller reloads GrowthBook features.

    Returns:
        None.
    """
    for listener in list(self._listeners):
        listener()

InMemoryFeatureFlagsAdapter ¶

A dependency-free adapter backed by an in-process dict.

Ideal for tests, local development and a safe default when no remote backend is configured. Mutating a flag through :meth:set notifies subscribers, which is how a flag flip drives a re-render in unit tests.

Source code in tempestweb/observability/feature_flags.py

class InMemoryFeatureFlagsAdapter:
    """A dependency-free adapter backed by an in-process dict.

    Ideal for tests, local development and a safe default when no remote backend
    is configured. Mutating a flag through :meth:`set` notifies subscribers,
    which is how a flag flip drives a re-render in unit tests.
    """

    def __init__(self, flags: dict[str, FlagValue] | None = None) -> None:
        """Initialize the adapter with optional seed flags.

        Args:
            flags: Initial flag values. A copy is stored so later mutation of the
                caller's dict has no effect.
        """
        self._flags: dict[str, FlagValue] = dict(flags or {})
        self._listeners: list[ChangeListener] = []

    def get(self, key: str, default: FlagValue = None) -> FlagValue:
        """Return a flag's value, or ``default`` when unknown.

        Args:
            key: The flag key.
            default: The value returned when the flag is absent.

        Returns:
            The flag value, or ``default``.
        """
        return self._flags.get(key, default)

    def set(self, key: str, value: FlagValue) -> None:
        """Set a flag and notify subscribers.

        Args:
            key: The flag key.
            value: The new value.

        Returns:
            None.
        """
        self._flags[key] = value
        self._emit()

    def subscribe(self, listener: ChangeListener) -> Callable[[], None]:
        """Register a change listener.

        Args:
            listener: A zero-argument callback invoked on change.

        Returns:
            An unsubscribe callable.
        """
        self._listeners.append(listener)

        def unsubscribe() -> None:
            """Remove the listener if still registered."""
            if listener in self._listeners:
                self._listeners.remove(listener)

        return unsubscribe

    def _emit(self) -> None:
        """Notify every subscriber of a change.

        Returns:
            None.
        """
        for listener in list(self._listeners):
            listener()

get ¶

get(key: str, default: FlagValue = None) -> FlagValue

Return a flag's value, or default when unknown.

Parameters:

Name	Type	Description	Default
`key`	`str`	The flag key.	required
`default`	`FlagValue`	The value returned when the flag is absent.	`None`

Returns:

Type	Description
`FlagValue`	The flag value, or `default`.

Source code in tempestweb/observability/feature_flags.py

def get(self, key: str, default: FlagValue = None) -> FlagValue:
    """Return a flag's value, or ``default`` when unknown.

    Args:
        key: The flag key.
        default: The value returned when the flag is absent.

    Returns:
        The flag value, or ``default``.
    """
    return self._flags.get(key, default)

set ¶

set(key: str, value: FlagValue) -> None

Set a flag and notify subscribers.

Parameters:

Name	Type	Description	Default
`key`	`str`	The flag key.	required
`value`	`FlagValue`	The new value.	required

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/feature_flags.py

def set(self, key: str, value: FlagValue) -> None:
    """Set a flag and notify subscribers.

    Args:
        key: The flag key.
        value: The new value.

    Returns:
        None.
    """
    self._flags[key] = value
    self._emit()

subscribe(listener: ChangeListener) -> Callable[[], None]

Register a change listener.

Parameters:

Name	Type	Description	Default
`listener`	`ChangeListener`	A zero-argument callback invoked on change.	required

Returns:

Type	Description
`Callable[[], None]`	An unsubscribe callable.

Source code in tempestweb/observability/feature_flags.py

def subscribe(self, listener: ChangeListener) -> Callable[[], None]:
    """Register a change listener.

    Args:
        listener: A zero-argument callback invoked on change.

    Returns:
        An unsubscribe callable.
    """
    self._listeners.append(listener)

    def unsubscribe() -> None:
        """Remove the listener if still registered."""
        if listener in self._listeners:
            self._listeners.remove(listener)

    return unsubscribe

LaunchDarklyFeatureFlagsAdapter ¶

An adapter that maps flag reads onto an injected LaunchDarkly client.

launchdarkly-server-sdk is not a tempestweb dependency; the caller injects a client exposing variation(key, context, default) plus a stored evaluation context. LaunchDarkly streams updates, so the caller wires the SDK's update callback to :meth:notify.

Source code in tempestweb/observability/feature_flags.py

class LaunchDarklyFeatureFlagsAdapter:
    """An adapter that maps flag reads onto an injected LaunchDarkly client.

    ``launchdarkly-server-sdk`` is not a tempestweb dependency; the caller
    injects a client exposing ``variation(key, context, default)`` plus a stored
    evaluation context. LaunchDarkly streams updates, so the caller wires the
    SDK's update callback to :meth:`notify`.
    """

    def __init__(self, client: Any, context: Any) -> None:  # noqa: ANN401 - injected third-party LaunchDarkly client/context
        """Initialize the adapter.

        Args:
            client: A LaunchDarkly-compatible client exposing
                ``variation(key, context, default)``.
            context: The evaluation context (user/device) passed to every
                ``variation`` call.
        """
        self._client: Any = client
        self._context: Any = context
        self._listeners: list[ChangeListener] = []

    def get(self, key: str, default: FlagValue = None) -> FlagValue:
        """Return a flag variation from LaunchDarkly.

        Args:
            key: The flag key.
            default: The value returned when evaluation falls back.

        Returns:
            The evaluated variation, or ``default``.
        """
        value: FlagValue = self._client.variation(key, self._context, default)
        return value

    def subscribe(self, listener: ChangeListener) -> Callable[[], None]:
        """Register a change listener fired by :meth:`notify`.

        Args:
            listener: A zero-argument callback invoked on change.

        Returns:
            An unsubscribe callable.
        """
        self._listeners.append(listener)

        def unsubscribe() -> None:
            """Remove the listener if still registered."""
            if listener in self._listeners:
                self._listeners.remove(listener)

        return unsubscribe

    def notify(self) -> None:
        """Notify subscribers when LaunchDarkly streams a flag update.

        Returns:
            None.
        """
        for listener in list(self._listeners):
            listener()

get ¶

get(key: str, default: FlagValue = None) -> FlagValue

Return a flag variation from LaunchDarkly.

Parameters:

Name	Type	Description	Default
`key`	`str`	The flag key.	required
`default`	`FlagValue`	The value returned when evaluation falls back.	`None`

Returns:

Type	Description
`FlagValue`	The evaluated variation, or `default`.

Source code in tempestweb/observability/feature_flags.py

def get(self, key: str, default: FlagValue = None) -> FlagValue:
    """Return a flag variation from LaunchDarkly.

    Args:
        key: The flag key.
        default: The value returned when evaluation falls back.

    Returns:
        The evaluated variation, or ``default``.
    """
    value: FlagValue = self._client.variation(key, self._context, default)
    return value

subscribe(listener: ChangeListener) -> Callable[[], None]

Register a change listener fired by :meth:notify.

Parameters:

Name	Type	Description	Default
`listener`	`ChangeListener`	A zero-argument callback invoked on change.	required

Returns:

Type	Description
`Callable[[], None]`	An unsubscribe callable.

Source code in tempestweb/observability/feature_flags.py

def subscribe(self, listener: ChangeListener) -> Callable[[], None]:
    """Register a change listener fired by :meth:`notify`.

    Args:
        listener: A zero-argument callback invoked on change.

    Returns:
        An unsubscribe callable.
    """
    self._listeners.append(listener)

    def unsubscribe() -> None:
        """Remove the listener if still registered."""
        if listener in self._listeners:
            self._listeners.remove(listener)

    return unsubscribe

notify ¶

notify() -> None

Notify subscribers when LaunchDarkly streams a flag update.

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/feature_flags.py

def notify(self) -> None:
    """Notify subscribers when LaunchDarkly streams a flag update.

    Returns:
        None.
    """
    for listener in list(self._listeners):
        listener()

Logger ¶

A structured logger that fans records out to its sinks above a threshold.

Records below level are dropped before any sink runs, so an expensive network sink never sees a filtered-out DEBUG line. A sink that raises is isolated: the remaining sinks still receive the record, because one broken destination must not take down logging for the rest.

Source code in tempestweb/observability/logger.py

class Logger:
    """A structured logger that fans records out to its sinks above a threshold.

    Records below ``level`` are dropped before any sink runs, so an expensive
    network sink never sees a filtered-out ``DEBUG`` line. A sink that raises is
    isolated: the remaining sinks still receive the record, because one broken
    destination must not take down logging for the rest.
    """

    def __init__(self, sinks: list[LoggerSink], level: LogLevel = "INFO") -> None:
        """Initialize the logger.

        Args:
            sinks: The destinations every passing record is delivered to. A copy
                is stored so later mutation of the caller's list has no effect.
            level: The minimum severity a record must have to be delivered.
        """
        self._sinks: list[LoggerSink] = list(sinks)
        self._level: LogLevel = level

    @property
    def level(self) -> LogLevel:
        """The current minimum severity threshold.

        Returns:
            The active :data:`LogLevel`.
        """
        return self._level

    def set_level(self, level: LogLevel) -> None:
        """Change the minimum severity threshold at runtime.

        Args:
            level: The new minimum severity.

        Returns:
            None.
        """
        self._level = level

    def _enabled(self, level: LogLevel) -> bool:
        """Whether a record at ``level`` clears the current threshold.

        Args:
            level: The severity to test.

        Returns:
            ``True`` if a record at ``level`` should be delivered.
        """
        return _LEVEL_ORDER[level] >= _LEVEL_ORDER[self._level]

    def log(self, level: LogLevel, message: str, **fields: Any) -> None:  # noqa: ANN401 - structured log fields are arbitrary JSON-able
        """Emit a record at an explicit level.

        Args:
            level: The severity of the record.
            message: The log message.
            **fields: Arbitrary structured fields attached to the record.

        Returns:
            None.
        """
        if not self._enabled(level):
            return
        record: LogRecord = LogRecord(level=level, message=message, fields=fields)
        for sink in self._sinks:
            try:
                sink(record)
            except Exception:  # noqa: BLE001 - one bad sink must not break others
                continue

    def debug(self, message: str, **fields: Any) -> None:  # noqa: ANN401 - structured log fields are arbitrary JSON-able
        """Emit a ``DEBUG`` record.

        Args:
            message: The log message.
            **fields: Arbitrary structured fields.

        Returns:
            None.
        """
        self.log("DEBUG", message, **fields)

    def info(self, message: str, **fields: Any) -> None:  # noqa: ANN401 - structured log fields are arbitrary JSON-able
        """Emit an ``INFO`` record.

        Args:
            message: The log message.
            **fields: Arbitrary structured fields.

        Returns:
            None.
        """
        self.log("INFO", message, **fields)

    def warning(self, message: str, **fields: Any) -> None:  # noqa: ANN401 - structured log fields are arbitrary JSON-able
        """Emit a ``WARNING`` record.

        Args:
            message: The log message.
            **fields: Arbitrary structured fields.

        Returns:
            None.
        """
        self.log("WARNING", message, **fields)

    def error(self, message: str, **fields: Any) -> None:  # noqa: ANN401 - structured log fields are arbitrary JSON-able
        """Emit an ``ERROR`` record.

        Args:
            message: The log message.
            **fields: Arbitrary structured fields.

        Returns:
            None.
        """
        self.log("ERROR", message, **fields)

    def critical(self, message: str, **fields: Any) -> None:  # noqa: ANN401 - structured log fields are arbitrary JSON-able
        """Emit a ``CRITICAL`` record.

        Args:
            message: The log message.
            **fields: Arbitrary structured fields.

        Returns:
            None.
        """
        self.log("CRITICAL", message, **fields)

level `property` ¶

level: LogLevel

The current minimum severity threshold.

Returns:

Type	Description
`LogLevel`	The active :data:`LogLevel`.

set_level ¶

set_level(level: LogLevel) -> None

Change the minimum severity threshold at runtime.

Parameters:

Name	Type	Description	Default
`level`	`LogLevel`	The new minimum severity.	required

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/logger.py

def set_level(self, level: LogLevel) -> None:
    """Change the minimum severity threshold at runtime.

    Args:
        level: The new minimum severity.

    Returns:
        None.
    """
    self._level = level

log ¶

log(level: LogLevel, message: str, **fields: Any) -> None

Emit a record at an explicit level.

Parameters:

Name	Type	Description	Default
`level`	`LogLevel`	The severity of the record.	required
`message`	`str`	The log message.	required
`**fields`	`Any`	Arbitrary structured fields attached to the record.	`{}`

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/logger.py

def log(self, level: LogLevel, message: str, **fields: Any) -> None:  # noqa: ANN401 - structured log fields are arbitrary JSON-able
    """Emit a record at an explicit level.

    Args:
        level: The severity of the record.
        message: The log message.
        **fields: Arbitrary structured fields attached to the record.

    Returns:
        None.
    """
    if not self._enabled(level):
        return
    record: LogRecord = LogRecord(level=level, message=message, fields=fields)
    for sink in self._sinks:
        try:
            sink(record)
        except Exception:  # noqa: BLE001 - one bad sink must not break others
            continue

debug ¶

debug(message: str, **fields: Any) -> None

Emit a DEBUG record.

Parameters:

Name	Type	Description	Default
`message`	`str`	The log message.	required
`**fields`	`Any`	Arbitrary structured fields.	`{}`

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/logger.py

def debug(self, message: str, **fields: Any) -> None:  # noqa: ANN401 - structured log fields are arbitrary JSON-able
    """Emit a ``DEBUG`` record.

    Args:
        message: The log message.
        **fields: Arbitrary structured fields.

    Returns:
        None.
    """
    self.log("DEBUG", message, **fields)

info ¶

info(message: str, **fields: Any) -> None

Emit an INFO record.

Parameters:

Name	Type	Description	Default
`message`	`str`	The log message.	required
`**fields`	`Any`	Arbitrary structured fields.	`{}`

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/logger.py

def info(self, message: str, **fields: Any) -> None:  # noqa: ANN401 - structured log fields are arbitrary JSON-able
    """Emit an ``INFO`` record.

    Args:
        message: The log message.
        **fields: Arbitrary structured fields.

    Returns:
        None.
    """
    self.log("INFO", message, **fields)

warning ¶

warning(message: str, **fields: Any) -> None

Emit a WARNING record.

Parameters:

Name	Type	Description	Default
`message`	`str`	The log message.	required
`**fields`	`Any`	Arbitrary structured fields.	`{}`

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/logger.py

def warning(self, message: str, **fields: Any) -> None:  # noqa: ANN401 - structured log fields are arbitrary JSON-able
    """Emit a ``WARNING`` record.

    Args:
        message: The log message.
        **fields: Arbitrary structured fields.

    Returns:
        None.
    """
    self.log("WARNING", message, **fields)

error ¶

error(message: str, **fields: Any) -> None

Emit an ERROR record.

Parameters:

Name	Type	Description	Default
`message`	`str`	The log message.	required
`**fields`	`Any`	Arbitrary structured fields.	`{}`

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/logger.py

def error(self, message: str, **fields: Any) -> None:  # noqa: ANN401 - structured log fields are arbitrary JSON-able
    """Emit an ``ERROR`` record.

    Args:
        message: The log message.
        **fields: Arbitrary structured fields.

    Returns:
        None.
    """
    self.log("ERROR", message, **fields)

critical ¶

critical(message: str, **fields: Any) -> None

Emit a CRITICAL record.

Parameters:

Name	Type	Description	Default
`message`	`str`	The log message.	required
`**fields`	`Any`	Arbitrary structured fields.	`{}`

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/logger.py

def critical(self, message: str, **fields: Any) -> None:  # noqa: ANN401 - structured log fields are arbitrary JSON-able
    """Emit a ``CRITICAL`` record.

    Args:
        message: The log message.
        **fields: Arbitrary structured fields.

    Returns:
        None.
    """
    self.log("CRITICAL", message, **fields)

LoggerSink ¶

Bases: Protocol

A destination for log records.

A sink is any callable taking a single :class:LogRecord. This is deliberately the same shape as list.append and print-style helpers, so capturing logs in a test is just passing my_list.append as a sink.

Source code in tempestweb/observability/logger.py

class LoggerSink(Protocol):
    """A destination for log records.

    A sink is any callable taking a single :class:`LogRecord`. This is
    deliberately the same shape as ``list.append`` and ``print``-style helpers,
    so capturing logs in a test is just passing ``my_list.append`` as a sink.
    """

    def __call__(self, record: LogRecord) -> None:
        """Consume a single log record.

        Args:
            record: The structured record to handle.

        Returns:
            None.
        """
        ...

LogRecord `dataclass` ¶

One structured log entry handed to every sink.

Attributes:

Name	Type	Description
`level`	`LogLevel`	The severity of this record.
`message`	`str`	The human-readable log message.
`fields`	`dict[str, Any]`	Arbitrary JSON-able structured fields attached at the call site.

Source code in tempestweb/observability/logger.py

@dataclass(frozen=True)
class LogRecord:
    """One structured log entry handed to every sink.

    Attributes:
        level: The severity of this record.
        message: The human-readable log message.
        fields: Arbitrary JSON-able structured fields attached at the call site.
    """

    level: LogLevel
    message: str
    fields: dict[str, Any] = field(default_factory=dict)

PatchMetrics `dataclass` ¶

Latency histogram and counters for the patch round trip.

A round is one event's whole cost as the operator experiences it: the handler, the rebuild, the diff and handing the batch to the transport. Splitting it finer would measure the core, which its own benchmark already does; this measures the server.

Attributes:

Name	Type	Description
`buckets`	`tuple[float, ...]`	Upper bounds in seconds, ascending.
`counts`	`list[int]`	Cumulative count per bucket (Prometheus semantics).
`total_seconds`	`float`	Sum of observed durations, for the average.
`rounds`	`int`	How many rounds were observed.
`patches`	`int`	How many patches those rounds produced.

Source code in tempestweb/observability/server.py

@dataclass
class PatchMetrics:
    """Latency histogram and counters for the patch round trip.

    A round is one event's whole cost as the operator experiences it: the handler,
    the rebuild, the diff and handing the batch to the transport. Splitting it
    finer would measure the core, which its own benchmark already does; this
    measures the server.

    Attributes:
        buckets: Upper bounds in seconds, ascending.
        counts: Cumulative count per bucket (Prometheus semantics).
        total_seconds: Sum of observed durations, for the average.
        rounds: How many rounds were observed.
        patches: How many patches those rounds produced.
    """

    buckets: tuple[float, ...] = DEFAULT_BUCKETS
    counts: list[int] = field(default_factory=list)
    total_seconds: float = 0.0
    rounds: int = 0
    patches: int = 0

    def __post_init__(self) -> None:
        """Size the bucket counters to the configured bounds."""
        if not self.counts:
            self.counts = [0] * len(self.buckets)

    def observe(self, seconds: float, patches: int) -> None:
        """Record one patch round.

        Args:
            seconds: How long the round took.
            patches: How many patches it produced.
        """
        self.rounds += 1
        self.patches += patches
        self.total_seconds += seconds
        for index, bound in enumerate(self.buckets):
            if seconds <= bound:
                self.counts[index] += 1

    def prometheus(self) -> str:
        """Render the histogram and counters as Prometheus text.

        Returns:
            The metric lines, newline-terminated.
        """
        lines = [
            "# HELP tempestweb_patch_seconds Event-to-patch round duration.",
            "# TYPE tempestweb_patch_seconds histogram",
        ]
        cumulative = 0
        for bound, count in zip(self.buckets, self.counts, strict=True):
            cumulative = max(cumulative, count)
            lines.append(f'tempestweb_patch_seconds_bucket{{le="{bound}"}} {count}')
        lines += [
            f'tempestweb_patch_seconds_bucket{{le="+Inf"}} {self.rounds}',
            f"tempestweb_patch_seconds_sum {self.total_seconds:.6f}",
            f"tempestweb_patch_seconds_count {self.rounds}",
            "# HELP tempestweb_patches_total Patches sent to clients.",
            "# TYPE tempestweb_patches_total counter",
            f"tempestweb_patches_total {self.patches}",
        ]
        return "\n".join(lines) + "\n"

observe ¶

observe(seconds: float, patches: int) -> None

Record one patch round.

Parameters:

Name	Type	Description	Default
`seconds`	`float`	How long the round took.	required
`patches`	`int`	How many patches it produced.	required

Source code in tempestweb/observability/server.py

def observe(self, seconds: float, patches: int) -> None:
    """Record one patch round.

    Args:
        seconds: How long the round took.
        patches: How many patches it produced.
    """
    self.rounds += 1
    self.patches += patches
    self.total_seconds += seconds
    for index, bound in enumerate(self.buckets):
        if seconds <= bound:
            self.counts[index] += 1

prometheus ¶

prometheus() -> str

Render the histogram and counters as Prometheus text.

Returns:

Type	Description
`str`	The metric lines, newline-terminated.

Source code in tempestweb/observability/server.py

def prometheus(self) -> str:
    """Render the histogram and counters as Prometheus text.

    Returns:
        The metric lines, newline-terminated.
    """
    lines = [
        "# HELP tempestweb_patch_seconds Event-to-patch round duration.",
        "# TYPE tempestweb_patch_seconds histogram",
    ]
    cumulative = 0
    for bound, count in zip(self.buckets, self.counts, strict=True):
        cumulative = max(cumulative, count)
        lines.append(f'tempestweb_patch_seconds_bucket{{le="{bound}"}} {count}')
    lines += [
        f'tempestweb_patch_seconds_bucket{{le="+Inf"}} {self.rounds}',
        f"tempestweb_patch_seconds_sum {self.total_seconds:.6f}",
        f"tempestweb_patch_seconds_count {self.rounds}",
        "# HELP tempestweb_patches_total Patches sent to clients.",
        "# TYPE tempestweb_patches_total counter",
        f"tempestweb_patches_total {self.patches}",
    ]
    return "\n".join(lines) + "\n"

ServerObservability ¶

The server's observability seam: metrics, structured logs, tracing.

Every part is optional and independent. The default instance is inert, which is what makes it safe to call from the hot path unconditionally: no histogram, no logger, a no-op tracer.

Attributes:

Name	Type	Description
`metrics`	`PatchMetrics \| None`	The latency/throughput collector, or None.
`logger`	`Logger \| None`	The structured logger, or None.
`tracer`	`Tracer`	The tracer; :func:`noop_tracer` by default.

Source code in tempestweb/observability/server.py

class ServerObservability:
    """The server's observability seam: metrics, structured logs, tracing.

    Every part is optional and independent. The default instance is inert, which is
    what makes it safe to call from the hot path unconditionally: no histogram, no
    logger, a no-op tracer.

    Attributes:
        metrics: The latency/throughput collector, or None.
        logger: The structured logger, or None.
        tracer: The tracer; :func:`noop_tracer` by default.
    """

    def __init__(
        self,
        metrics: PatchMetrics | None = None,
        logger: Logger | None = None,
        tracer: Tracer | None = None,
    ) -> None:
        """Wire the parts an app asked for.

        Args:
            metrics: Collector for patch latency and throughput.
            logger: Structured logger for session lifecycle events.
            tracer: Tracing adapter; the no-op tracer when omitted.
        """
        self.metrics: PatchMetrics | None = metrics
        self.logger: Logger | None = logger
        self.tracer: Tracer = tracer or noop_tracer()

    @property
    def enabled(self) -> bool:
        """Whether anything is actually collected.

        Returns:
            True when metrics, a logger or a real tracer is wired.
        """
        return (
            self.metrics is not None
            or self.logger is not None
            or self.tracer is not _NOOP_TRACER
        )

    def observe_patches(self, seconds: float, patches: int) -> None:
        """Record one event-to-patch latency, when metrics are on.

        ``seconds`` is the wait the **client** experienced: from the event arriving
        to its patches being handed to the transport, rebuild included. That is the
        number an SLO is written against, and the reason the histogram is not taken
        around the handler alone.

        Args:
            seconds: How long the client waited.
            patches: How many patches the batch carries.
        """
        if self.metrics is not None:
            self.metrics.observe(seconds, patches)

    @contextmanager
    def session(self, session_id: str, **attributes: Any) -> Iterator[Any]:  # noqa: ANN401 — span
        """Trace and log one session's whole lifetime.

        The log carries the same ``session_id`` the span does, which is the point:
        a complaint about one client becomes a log query and a trace lookup with
        the same key.

        Args:
            session_id: The session's id.
            **attributes: Extra attributes for the span and the log records.

        Yields:
            The session's span.
        """
        started = time.perf_counter()
        if self.logger is not None:
            self.logger.info("session.open", session_id=session_id, **attributes)
        reason = "closed"
        try:
            with self.tracer.span(
                "tempestweb.session", session_id=session_id, **attributes
            ) as span:
                yield span
        except BaseException as exc:
            reason = type(exc).__name__
            raise
        finally:
            if self.logger is not None:
                self.logger.info(
                    "session.close",
                    session_id=session_id,
                    reason=reason,
                    duration_s=round(time.perf_counter() - started, 6),
                    **attributes,
                )

    @contextmanager
    def dispatch(self, session_id: str, event_type: str) -> Iterator[Any]:  # noqa: ANN401 — span
        """Trace one handler invocation.

        This is the *handler's* span, and deliberately not where the latency
        histogram is taken: the rebuild the handler triggers is **coalesced**, so it
        runs after the handler returns and may cover several events. Timing this
        block would report a number that stops before the work the client is waiting
        for — measured, and it read as ``0 patches`` per round. The histogram is
        taken where the batch actually leaves (:meth:`observe_patches`).

        Args:
            session_id: The session the event belongs to.
            event_type: The wire event type, as a span attribute.

        Yields:
            The handler's span.
        """
        with self.tracer.span(
            "tempestweb.dispatch", session_id=session_id, event_type=event_type
        ) as span:
            yield span

    @contextmanager
    def patch_batch(self, session_id: str, patches: int) -> Iterator[Any]:  # noqa: ANN401 — span
        """Trace one outgoing patch batch.

        Args:
            session_id: The session the batch belongs to.
            patches: How many patches the batch carries.

        Yields:
            The batch's span.
        """
        with self.tracer.span(
            "tempestweb.patch_batch", session_id=session_id, patches=patches
        ) as span:
            yield span

    def prometheus(self) -> str:
        """Render the metrics this instance collected.

        Returns:
            Prometheus text, empty when metrics are off.
        """
        return "" if self.metrics is None else self.metrics.prometheus()

enabled `property` ¶

enabled: bool

Whether anything is actually collected.

Returns:

Type	Description
`bool`	True when metrics, a logger or a real tracer is wired.

observe_patches ¶

observe_patches(seconds: float, patches: int) -> None

Record one event-to-patch latency, when metrics are on.

seconds is the wait the client experienced: from the event arriving to its patches being handed to the transport, rebuild included. That is the number an SLO is written against, and the reason the histogram is not taken around the handler alone.

Parameters:

Name	Type	Description	Default
`seconds`	`float`	How long the client waited.	required
`patches`	`int`	How many patches the batch carries.	required

Source code in tempestweb/observability/server.py

def observe_patches(self, seconds: float, patches: int) -> None:
    """Record one event-to-patch latency, when metrics are on.

    ``seconds`` is the wait the **client** experienced: from the event arriving
    to its patches being handed to the transport, rebuild included. That is the
    number an SLO is written against, and the reason the histogram is not taken
    around the handler alone.

    Args:
        seconds: How long the client waited.
        patches: How many patches the batch carries.
    """
    if self.metrics is not None:
        self.metrics.observe(seconds, patches)

session ¶

session(session_id: str, **attributes: Any) -> Iterator[Any]

Trace and log one session's whole lifetime.

The log carries the same session_id the span does, which is the point: a complaint about one client becomes a log query and a trace lookup with the same key.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session's id.	required
`**attributes`	`Any`	Extra attributes for the span and the log records.	`{}`

Yields:

Type	Description
`Any`	The session's span.

Source code in tempestweb/observability/server.py

@contextmanager
def session(self, session_id: str, **attributes: Any) -> Iterator[Any]:  # noqa: ANN401 — span
    """Trace and log one session's whole lifetime.

    The log carries the same ``session_id`` the span does, which is the point:
    a complaint about one client becomes a log query and a trace lookup with
    the same key.

    Args:
        session_id: The session's id.
        **attributes: Extra attributes for the span and the log records.

    Yields:
        The session's span.
    """
    started = time.perf_counter()
    if self.logger is not None:
        self.logger.info("session.open", session_id=session_id, **attributes)
    reason = "closed"
    try:
        with self.tracer.span(
            "tempestweb.session", session_id=session_id, **attributes
        ) as span:
            yield span
    except BaseException as exc:
        reason = type(exc).__name__
        raise
    finally:
        if self.logger is not None:
            self.logger.info(
                "session.close",
                session_id=session_id,
                reason=reason,
                duration_s=round(time.perf_counter() - started, 6),
                **attributes,
            )

dispatch ¶

dispatch(session_id: str, event_type: str) -> Iterator[Any]

Trace one handler invocation.

This is the handler's span, and deliberately not where the latency histogram is taken: the rebuild the handler triggers is coalesced, so it runs after the handler returns and may cover several events. Timing this block would report a number that stops before the work the client is waiting for — measured, and it read as 0 patches per round. The histogram is taken where the batch actually leaves (:meth:observe_patches).

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session the event belongs to.	required
`event_type`	`str`	The wire event type, as a span attribute.	required

Yields:

Type	Description
`Any`	The handler's span.

Source code in tempestweb/observability/server.py

@contextmanager
def dispatch(self, session_id: str, event_type: str) -> Iterator[Any]:  # noqa: ANN401 — span
    """Trace one handler invocation.

    This is the *handler's* span, and deliberately not where the latency
    histogram is taken: the rebuild the handler triggers is **coalesced**, so it
    runs after the handler returns and may cover several events. Timing this
    block would report a number that stops before the work the client is waiting
    for — measured, and it read as ``0 patches`` per round. The histogram is
    taken where the batch actually leaves (:meth:`observe_patches`).

    Args:
        session_id: The session the event belongs to.
        event_type: The wire event type, as a span attribute.

    Yields:
        The handler's span.
    """
    with self.tracer.span(
        "tempestweb.dispatch", session_id=session_id, event_type=event_type
    ) as span:
        yield span

patch_batch ¶

patch_batch(session_id: str, patches: int) -> Iterator[Any]

Trace one outgoing patch batch.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session the batch belongs to.	required
`patches`	`int`	How many patches the batch carries.	required

Yields:

Type	Description
`Any`	The batch's span.

Source code in tempestweb/observability/server.py

@contextmanager
def patch_batch(self, session_id: str, patches: int) -> Iterator[Any]:  # noqa: ANN401 — span
    """Trace one outgoing patch batch.

    Args:
        session_id: The session the batch belongs to.
        patches: How many patches the batch carries.

    Yields:
        The batch's span.
    """
    with self.tracer.span(
        "tempestweb.patch_batch", session_id=session_id, patches=patches
    ) as span:
        yield span

prometheus ¶

prometheus() -> str

Render the metrics this instance collected.

Returns:

Type	Description
`str`	Prometheus text, empty when metrics are off.

Source code in tempestweb/observability/server.py

def prometheus(self) -> str:
    """Render the metrics this instance collected.

    Returns:
        Prometheus text, empty when metrics are off.
    """
    return "" if self.metrics is None else self.metrics.prometheus()

Span ¶

Bases: Protocol

One unit of traced work, ended by the context manager that opened it.

Source code in tempestweb/observability/server.py

class Span(Protocol):
    """One unit of traced work, ended by the context manager that opened it."""

    def set_attribute(self, key: str, value: Any) -> None:  # noqa: ANN401 — span attrs are scalars
        """Record one attribute on this span.

        Args:
            key: The attribute name.
            value: A scalar the exporter can carry.
        """
        ...

set_attribute ¶

set_attribute(key: str, value: Any) -> None

Record one attribute on this span.

Parameters:

Name	Type	Description	Default
`key`	`str`	The attribute name.	required
`value`	`Any`	A scalar the exporter can carry.	required

Source code in tempestweb/observability/server.py

def set_attribute(self, key: str, value: Any) -> None:  # noqa: ANN401 — span attrs are scalars
    """Record one attribute on this span.

    Args:
        key: The attribute name.
        value: A scalar the exporter can carry.
    """
    ...

Tracer ¶

Bases: Protocol

The tracing seam: open a span, get it back, end it on exit.

Source code in tempestweb/observability/server.py

class Tracer(Protocol):
    """The tracing seam: open a span, get it back, end it on exit."""

    def span(self, name: str, **attributes: Any) -> Any:  # noqa: ANN401 — a context manager
        """Open a span.

        Args:
            name: The span name.
            **attributes: Initial attributes.

        Returns:
            A context manager yielding a :class:`Span`.
        """
        ...

span ¶

span(name: str, **attributes: Any) -> Any

Open a span.

Parameters:

Name	Type	Description	Default
`name`	`str`	The span name.	required
`**attributes`	`Any`	Initial attributes.	`{}`

Returns:

Type	Description
`Any`	A context manager yielding a :class:`Span`.

Source code in tempestweb/observability/server.py

def span(self, name: str, **attributes: Any) -> Any:  # noqa: ANN401 — a context manager
    """Open a span.

    Args:
        name: The span name.
        **attributes: Initial attributes.

    Returns:
        A context manager yielding a :class:`Span`.
    """
    ...

ConsoleTelemetryAdapter ¶

A zero-dependency adapter that prints events through a sink callable.

This is the default adapter and the Mode A (browser) fallback: in the browser the sink is console.log; under CPython it defaults to :func:print. Injecting the sink keeps it trivially testable.

Source code in tempestweb/observability/telemetry.py

class ConsoleTelemetryAdapter:
    """A zero-dependency adapter that prints events through a sink callable.

    This is the default adapter and the Mode A (browser) fallback: in the browser
    the sink is ``console.log``; under CPython it defaults to :func:`print`.
    Injecting the sink keeps it trivially testable.
    """

    def __init__(self, sink: Any = print) -> None:  # noqa: ANN401 - injected console-like sink
        """Initialize the adapter.

        Args:
            sink: A callable invoked with a single string argument for each
                event. Defaults to the built-in :func:`print`.
        """
        self._sink: Any = sink

    def track(self, event: str, props: dict[str, Any]) -> None:
        """Print a tracked event.

        Args:
            event: The event name.
            props: The event properties.

        Returns:
            None.
        """
        self._sink(f"[telemetry] track {event} {props}")

    def identify(self, user_id: str, traits: dict[str, Any]) -> None:
        """Print an identify call.

        Args:
            user_id: The user identifier.
            traits: The identity traits.

        Returns:
            None.
        """
        self._sink(f"[telemetry] identify {user_id} {traits}")

track ¶

track(event: str, props: dict[str, Any]) -> None

Print a tracked event.

Parameters:

Name	Type	Description	Default
`event`	`str`	The event name.	required
`props`	`dict[str, Any]`	The event properties.	required

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/telemetry.py

def track(self, event: str, props: dict[str, Any]) -> None:
    """Print a tracked event.

    Args:
        event: The event name.
        props: The event properties.

    Returns:
        None.
    """
    self._sink(f"[telemetry] track {event} {props}")

identify ¶

identify(user_id: str, traits: dict[str, Any]) -> None

Print an identify call.

Parameters:

Name	Type	Description	Default
`user_id`	`str`	The user identifier.	required
`traits`	`dict[str, Any]`	The identity traits.	required

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/telemetry.py

def identify(self, user_id: str, traits: dict[str, Any]) -> None:
    """Print an identify call.

    Args:
        user_id: The user identifier.
        traits: The identity traits.

    Returns:
        None.
    """
    self._sink(f"[telemetry] identify {user_id} {traits}")

PostHogTelemetryAdapter ¶

An adapter that maps telemetry onto an injected PostHog client.

posthog is not a tempestweb dependency; the caller injects a client exposing capture and identify. A distinct_id is tracked across calls so events emitted before an explicit identify still attach to the right person once identity is known.

Source code in tempestweb/observability/telemetry.py

class PostHogTelemetryAdapter:
    """An adapter that maps telemetry onto an injected PostHog client.

    ``posthog`` is not a tempestweb dependency; the caller injects a client
    exposing ``capture`` and ``identify``. A ``distinct_id`` is tracked across
    calls so events emitted before an explicit identify still attach to the right
    person once identity is known.
    """

    def __init__(self, client: Any, *, distinct_id: str = "anonymous") -> None:  # noqa: ANN401 - injected third-party PostHog client
        """Initialize the adapter.

        Args:
            client: An object compatible with the PostHog SDK exposing
                ``capture(distinct_id, event, properties)`` and
                ``identify(distinct_id, properties)``.
            distinct_id: The initial distinct id used until ``identify`` runs.
        """
        self._client: Any = client
        self._distinct_id: str = distinct_id

    def track(self, event: str, props: dict[str, Any]) -> None:
        """Forward an event to PostHog under the current distinct id.

        Args:
            event: The event name.
            props: The event properties.

        Returns:
            None.
        """
        self._client.capture(
            distinct_id=self._distinct_id, event=event, properties=props
        )

    def identify(self, user_id: str, traits: dict[str, Any]) -> None:
        """Bind the distinct id and forward an identify call to PostHog.

        Args:
            user_id: The user identifier, used as the new distinct id.
            traits: Person properties to attach.

        Returns:
            None.
        """
        self._distinct_id = user_id
        self._client.identify(distinct_id=user_id, properties=traits)

track ¶

track(event: str, props: dict[str, Any]) -> None

Forward an event to PostHog under the current distinct id.

Parameters:

Name	Type	Description	Default
`event`	`str`	The event name.	required
`props`	`dict[str, Any]`	The event properties.	required

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/telemetry.py

def track(self, event: str, props: dict[str, Any]) -> None:
    """Forward an event to PostHog under the current distinct id.

    Args:
        event: The event name.
        props: The event properties.

    Returns:
        None.
    """
    self._client.capture(
        distinct_id=self._distinct_id, event=event, properties=props
    )

identify ¶

identify(user_id: str, traits: dict[str, Any]) -> None

Bind the distinct id and forward an identify call to PostHog.

Parameters:

Name	Type	Description	Default
`user_id`	`str`	The user identifier, used as the new distinct id.	required
`traits`	`dict[str, Any]`	Person properties to attach.	required

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/telemetry.py

def identify(self, user_id: str, traits: dict[str, Any]) -> None:
    """Bind the distinct id and forward an identify call to PostHog.

    Args:
        user_id: The user identifier, used as the new distinct id.
        traits: Person properties to attach.

    Returns:
        None.
    """
    self._distinct_id = user_id
    self._client.identify(distinct_id=user_id, properties=traits)

SentryTelemetryAdapter ¶

An adapter that maps telemetry onto an injected Sentry client.

sentry_sdk is not a tempestweb dependency; the caller injects the module (or any object exposing capture_message and set_user). Events become breadcrumb-style messages; identities become the Sentry user scope.

Source code in tempestweb/observability/telemetry.py

class SentryTelemetryAdapter:
    """An adapter that maps telemetry onto an injected Sentry client.

    ``sentry_sdk`` is not a tempestweb dependency; the caller injects the module
    (or any object exposing ``capture_message`` and ``set_user``). Events become
    breadcrumb-style messages; identities become the Sentry user scope.
    """

    def __init__(self, client: Any) -> None:  # noqa: ANN401 - injected third-party Sentry client
        """Initialize the adapter.

        Args:
            client: An object compatible with ``sentry_sdk`` exposing
                ``capture_message(message, level=..., extras=...)`` and
                ``set_user(dict)``.
        """
        self._client: Any = client

    def track(self, event: str, props: dict[str, Any]) -> None:
        """Forward an event as a Sentry message with the props as extras.

        Args:
            event: The event name.
            props: The event properties, attached as Sentry ``extras``.

        Returns:
            None.
        """
        self._client.capture_message(event, level="info", extras=props)

    def identify(self, user_id: str, traits: dict[str, Any]) -> None:
        """Set the Sentry user scope.

        Args:
            user_id: The user identifier, mapped to ``id``.
            traits: Extra identity fields merged into the user dict.

        Returns:
            None.
        """
        self._client.set_user({"id": user_id, **traits})

track ¶

track(event: str, props: dict[str, Any]) -> None

Forward an event as a Sentry message with the props as extras.

Parameters:

Name	Type	Description	Default
`event`	`str`	The event name.	required
`props`	`dict[str, Any]`	The event properties, attached as Sentry `extras`.	required

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/telemetry.py

def track(self, event: str, props: dict[str, Any]) -> None:
    """Forward an event as a Sentry message with the props as extras.

    Args:
        event: The event name.
        props: The event properties, attached as Sentry ``extras``.

    Returns:
        None.
    """
    self._client.capture_message(event, level="info", extras=props)

identify ¶

identify(user_id: str, traits: dict[str, Any]) -> None

Set the Sentry user scope.

Parameters:

Name	Type	Description	Default
`user_id`	`str`	The user identifier, mapped to `id`.	required
`traits`	`dict[str, Any]`	Extra identity fields merged into the user dict.	required

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/telemetry.py

def identify(self, user_id: str, traits: dict[str, Any]) -> None:
    """Set the Sentry user scope.

    Args:
        user_id: The user identifier, mapped to ``id``.
        traits: Extra identity fields merged into the user dict.

    Returns:
        None.
    """
    self._client.set_user({"id": user_id, **traits})

TelemetryAdapter ¶

Bases: Protocol

The minimal contract every telemetry backend must satisfy.

An adapter is intentionally tiny: two methods that map the provider's vocabulary (track / identify) onto a concrete backend. Implementing a new backend is a handful of lines, which keeps the seam between application code and vendor SDK thin and swappable.

Source code in tempestweb/observability/telemetry.py

@runtime_checkable
class TelemetryAdapter(Protocol):
    """The minimal contract every telemetry backend must satisfy.

    An adapter is intentionally tiny: two methods that map the provider's
    vocabulary (``track`` / ``identify``) onto a concrete backend. Implementing a
    new backend is a handful of lines, which keeps the seam between application
    code and vendor SDK thin and swappable.
    """

    def track(self, event: str, props: dict[str, Any]) -> None:
        """Record a named event with arbitrary properties.

        Args:
            event: The event name (e.g. ``"push_subscribed"``).
            props: JSON-able properties describing the event. Must already be
                free of PII the caller does not want sent to the backend.

        Returns:
            None.
        """
        ...

    def identify(self, user_id: str, traits: dict[str, Any]) -> None:
        """Associate subsequent events with a user identity.

        Args:
            user_id: A stable identifier for the current user.
            traits: JSON-able traits to attach to the identity.

        Returns:
            None.
        """
        ...

track ¶

track(event: str, props: dict[str, Any]) -> None

Record a named event with arbitrary properties.

Parameters:

Name	Type	Description	Default
`event`	`str`	The event name (e.g. `"push_subscribed"`).	required
`props`	`dict[str, Any]`	JSON-able properties describing the event. Must already be free of PII the caller does not want sent to the backend.	required

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/telemetry.py

def track(self, event: str, props: dict[str, Any]) -> None:
    """Record a named event with arbitrary properties.

    Args:
        event: The event name (e.g. ``"push_subscribed"``).
        props: JSON-able properties describing the event. Must already be
            free of PII the caller does not want sent to the backend.

    Returns:
        None.
    """
    ...

identify ¶

identify(user_id: str, traits: dict[str, Any]) -> None

Associate subsequent events with a user identity.

Parameters:

Name	Type	Description	Default
`user_id`	`str`	A stable identifier for the current user.	required
`traits`	`dict[str, Any]`	JSON-able traits to attach to the identity.	required

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/telemetry.py

def identify(self, user_id: str, traits: dict[str, Any]) -> None:
    """Associate subsequent events with a user identity.

    Args:
        user_id: A stable identifier for the current user.
        traits: JSON-able traits to attach to the identity.

    Returns:
        None.
    """
    ...

TelemetryProvider ¶

A backend-agnostic facade application code calls to emit telemetry.

The provider holds exactly one :class:TelemetryAdapter and forwards every call to it. It also enforces two cross-cutting concerns that should never leak into call sites:

Sampling — a sample_rate in [0.0, 1.0] drops a fraction of track calls so a chatty event cannot flood the backend. identify is never sampled (identities must be reliable).
Global properties — default_props are merged into every tracked event (e.g. {"mode": "wasm"}), without each call site repeating them.

Swapping the adapter changes the destination of every event while leaving all track / identify calls untouched.

Source code in tempestweb/observability/telemetry.py

class TelemetryProvider:
    """A backend-agnostic facade application code calls to emit telemetry.

    The provider holds exactly one :class:`TelemetryAdapter` and forwards every
    call to it. It also enforces two cross-cutting concerns that should never
    leak into call sites:

    * **Sampling** — a ``sample_rate`` in ``[0.0, 1.0]`` drops a fraction of
      ``track`` calls so a chatty event cannot flood the backend. ``identify`` is
      never sampled (identities must be reliable).
    * **Global properties** — ``default_props`` are merged into every tracked
      event (e.g. ``{"mode": "wasm"}``), without each call site repeating them.

    Swapping the adapter changes the destination of every event while leaving all
    ``track`` / ``identify`` calls untouched.
    """

    def __init__(
        self,
        adapter: TelemetryAdapter,
        *,
        default_props: dict[str, Any] | None = None,
        sample_rate: float = 1.0,
    ) -> None:
        """Initialize the provider.

        Args:
            adapter: The backend adapter every event is forwarded to.
            default_props: Properties merged into every tracked event. A copy is
                stored so later mutation of the caller's dict has no effect.
            sample_rate: Fraction of ``track`` calls to forward, in ``[0.0,
                1.0]``. ``1.0`` forwards all; ``0.0`` forwards none.

        Raises:
            ValueError: If ``sample_rate`` is outside ``[0.0, 1.0]``.
        """
        if not 0.0 <= sample_rate <= 1.0:
            raise ValueError("sample_rate must be within [0.0, 1.0]")
        self._adapter: TelemetryAdapter = adapter
        self._default_props: dict[str, Any] = dict(default_props or {})
        self._sample_rate: float = sample_rate
        self._counter: int = 0

    @property
    def adapter(self) -> TelemetryAdapter:
        """The adapter currently backing this provider.

        Returns:
            The active :class:`TelemetryAdapter`.
        """
        return self._adapter

    def _should_sample(self) -> bool:
        """Decide deterministically whether the next event passes sampling.

        A deterministic counter (rather than randomness) keeps tests reproducible
        and gives an even, predictable spread: with ``sample_rate`` ``0.5`` every
        other event is forwarded.

        Returns:
            ``True`` if the event should be forwarded to the adapter.
        """
        if self._sample_rate >= 1.0:
            return True
        if self._sample_rate <= 0.0:
            return False
        self._counter += 1
        # Forward when the running count crosses the next 1/sample_rate boundary.
        return (self._counter * self._sample_rate) % 1.0 < self._sample_rate

    def track(self, event: str, props: dict[str, Any] | None = None) -> None:
        """Record a named event, subject to sampling and global properties.

        Args:
            event: The event name.
            props: Optional per-event properties; merged on top of
                ``default_props``.

        Returns:
            None.
        """
        if not self._should_sample():
            return
        merged: dict[str, Any] = {**self._default_props, **(props or {})}
        self._adapter.track(event, merged)

    def identify(self, user_id: str, traits: dict[str, Any] | None = None) -> None:
        """Associate subsequent events with a user identity (never sampled).

        Args:
            user_id: A stable identifier for the current user.
            traits: Optional traits to attach to the identity.

        Returns:
            None.
        """
        self._adapter.identify(user_id, traits or {})

adapter `property` ¶

adapter: TelemetryAdapter

The adapter currently backing this provider.

Returns:

Type	Description
`TelemetryAdapter`	The active :class:`TelemetryAdapter`.

track ¶

track(event: str, props: dict[str, Any] | None = None) -> None

Record a named event, subject to sampling and global properties.

Parameters:

Name	Type	Description	Default
`event`	`str`	The event name.	required
`props`	`dict[str, Any] \| None`	Optional per-event properties; merged on top of `default_props`.	`None`

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/telemetry.py

def track(self, event: str, props: dict[str, Any] | None = None) -> None:
    """Record a named event, subject to sampling and global properties.

    Args:
        event: The event name.
        props: Optional per-event properties; merged on top of
            ``default_props``.

    Returns:
        None.
    """
    if not self._should_sample():
        return
    merged: dict[str, Any] = {**self._default_props, **(props or {})}
    self._adapter.track(event, merged)

identify ¶

identify(user_id: str, traits: dict[str, Any] | None = None) -> None

Associate subsequent events with a user identity (never sampled).

Parameters:

Name	Type	Description	Default
`user_id`	`str`	A stable identifier for the current user.	required
`traits`	`dict[str, Any] \| None`	Optional traits to attach to the identity.	`None`

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/telemetry.py

def identify(self, user_id: str, traits: dict[str, Any] | None = None) -> None:
    """Associate subsequent events with a user identity (never sampled).

    Args:
        user_id: A stable identifier for the current user.
        traits: Optional traits to attach to the identity.

    Returns:
        None.
    """
    self._adapter.identify(user_id, traits or {})

create_auth_store ¶

create_auth_store() -> AuthStore

Create a fresh, logged-out :class:AuthStore.

Returns:

Type	Description
`AuthStore`	A new :class:`AuthStore`.

Source code in tempestweb/observability/auth.py

def create_auth_store() -> AuthStore:
    """Create a fresh, logged-out :class:`AuthStore`.

    Returns:
        A new :class:`AuthStore`.
    """
    return AuthStore()

create_refresh_queue ¶

create_refresh_queue(store: AuthStore, refresh_fn: RefreshFn) -> RefreshQueue

Create a :class:RefreshQueue bound to a store and refresh function.

Parameters:

Name	Type	Description	Default
`store`	`AuthStore`	The auth store updated after a successful refresh.	required
`refresh_fn`	`RefreshFn`	The async renewal function returning a new access token.	required

Returns:

Type	Description
`RefreshQueue`	A configured :class:`RefreshQueue`.

Source code in tempestweb/observability/auth.py

def create_refresh_queue(store: AuthStore, refresh_fn: RefreshFn) -> RefreshQueue:
    """Create a :class:`RefreshQueue` bound to a store and refresh function.

    Args:
        store: The auth store updated after a successful refresh.
        refresh_fn: The async renewal function returning a new access token.

    Returns:
        A configured :class:`RefreshQueue`.
    """
    return RefreshQueue(store, refresh_fn)

decode_jwt ¶

decode_jwt(token: str) -> dict[str, Any]

Decode a JWT's payload claims without verifying the signature.

This is a client-side convenience for inspecting expiry and display claims. It must never be used to make an authorization decision — only the server (with the signing key) may trust a token's claims.

Parameters:

Name	Type	Description	Default
`token`	`str`	A compact-serialization JWT (`header.payload.signature`).	required

Returns:

Type	Description
`dict[str, Any]`	The decoded claims as a dictionary.

Raises:

Type	Description
`JWTError`	If the token is malformed or its payload is not a JSON object.

Source code in tempestweb/observability/auth.py

def decode_jwt(token: str) -> dict[str, Any]:
    """Decode a JWT's payload claims **without verifying the signature**.

    This is a client-side convenience for inspecting expiry and display claims.
    It must never be used to make an authorization decision — only the server
    (with the signing key) may trust a token's claims.

    Args:
        token: A compact-serialization JWT (``header.payload.signature``).

    Returns:
        The decoded claims as a dictionary.

    Raises:
        JWTError: If the token is malformed or its payload is not a JSON object.
    """
    parts: list[str] = token.split(".")
    if len(parts) != 3:
        raise JWTError("token must have three dot-separated segments")
    raw: bytes = _b64url_decode(parts[1])
    try:
        claims: Any = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise JWTError("token payload is not valid JSON") from exc
    if not isinstance(claims, dict):
        raise JWTError("token payload is not a JSON object")
    return claims

is_jwt_expired ¶

is_jwt_expired(token: str, *, leeway_seconds: int = 0, now: float | None = None) -> bool

Return whether a JWT is expired based on its exp claim.

A token without an exp claim is treated as not expiring (returns False). A malformed token is treated as expired (returns True) so the caller refreshes rather than trusting garbage.

Parameters:

Name	Type	Description	Default
`token`	`str`	The JWT to inspect.	required
`leeway_seconds`	`int`	Seconds of clock-skew tolerance; the token is considered expired this many seconds before its real `exp` so a refresh is triggered slightly early.	`0`
`now`	`float \| None`	The current UNIX time in seconds; defaults to :func:`time.time`.	`None`

Returns:

Type	Description
`bool`	`True` if the token is expired (or unparseable), `False` otherwise.

Source code in tempestweb/observability/auth.py

def is_jwt_expired(
    token: str, *, leeway_seconds: int = 0, now: float | None = None
) -> bool:
    """Return whether a JWT is expired based on its ``exp`` claim.

    A token without an ``exp`` claim is treated as **not** expiring (returns
    ``False``). A malformed token is treated as expired (returns ``True``) so the
    caller refreshes rather than trusting garbage.

    Args:
        token: The JWT to inspect.
        leeway_seconds: Seconds of clock-skew tolerance; the token is considered
            expired this many seconds *before* its real ``exp`` so a refresh is
            triggered slightly early.
        now: The current UNIX time in seconds; defaults to :func:`time.time`.

    Returns:
        ``True`` if the token is expired (or unparseable), ``False`` otherwise.
    """
    try:
        claims: dict[str, Any] = decode_jwt(token)
    except JWTError:
        return True
    exp: Any = claims.get("exp")
    if exp is None:
        return False
    current: float = time.time() if now is None else now
    try:
        return current >= float(exp) - leeway_seconds
    except (TypeError, ValueError):
        return True

route_guard ¶

route_guard(store: AuthStore, *, redirect_to: str = '/login') -> Callable[[str], str]

Build a route guard that redirects unauthenticated navigation.

Parameters:

Name	Type	Description	Default
`store`	`AuthStore`	The auth store consulted for the current session.	required
`redirect_to`	`str`	The route an unauthenticated request is sent to.	`'/login'`

Returns:

Type	Description
`Callable[[str], str]`	A function mapping a requested route name to the route that should
`Callable[[str], str]`	actually render: the request unchanged when authenticated (or when it is
`Callable[[str], str]`	already the redirect target), otherwise `redirect_to`.

Source code in tempestweb/observability/auth.py

def route_guard(
    store: AuthStore, *, redirect_to: str = "/login"
) -> Callable[[str], str]:
    """Build a route guard that redirects unauthenticated navigation.

    Args:
        store: The auth store consulted for the current session.
        redirect_to: The route an unauthenticated request is sent to.

    Returns:
        A function mapping a requested route name to the route that should
        actually render: the request unchanged when authenticated (or when it is
        already the redirect target), otherwise ``redirect_to``.
    """

    def guard(requested: str) -> str:
        """Resolve the effective route for a navigation request.

        Args:
            requested: The route the app is trying to navigate to.

        Returns:
            ``requested`` when allowed, otherwise ``redirect_to``.
        """
        if store.is_authenticated or requested == redirect_to:
            return requested
        return redirect_to

    return guard

server_decode_jwt ¶

server_decode_jwt(token: str, secret: str, **kwargs: Any) -> dict[str, Any]

Verify and decode a JWT on the server via tempest_fastapi_sdk.JWTUtils.

Mode B issues and validates its own tokens. Rather than re-implement signature verification, the server reuses the SDK's JWTUtils so token handling stays consistent with the rest of the user's backend stack.

Parameters:

Name	Type	Description	Default
`token`	`str`	The JWT to verify and decode.	required
`secret`	`str`	The signing secret used to verify the signature.	required
`**kwargs`	`Any`	Extra keyword arguments forwarded to `JWTUtils.decode` (e.g. `algorithms`), depending on the installed SDK version.	`{}`

Returns:

Type	Description
`dict[str, Any]`	The verified claims dictionary.

Raises:

Type	Description
`RuntimeError`	If `tempest_fastapi_sdk` is not installed. It is an optional, server-only dependency — install the `[auth]` extra.

Source code in tempestweb/observability/auth.py

def server_decode_jwt(token: str, secret: str, **kwargs: Any) -> dict[str, Any]:  # noqa: ANN401 - forwarded verbatim to JWTUtils.decode
    """Verify and decode a JWT on the server via ``tempest_fastapi_sdk.JWTUtils``.

    Mode B issues and validates its own tokens. Rather than re-implement
    signature verification, the server reuses the SDK's ``JWTUtils`` so token
    handling stays consistent with the rest of the user's backend stack.

    Args:
        token: The JWT to verify and decode.
        secret: The signing secret used to verify the signature.
        **kwargs: Extra keyword arguments forwarded to ``JWTUtils.decode`` (e.g.
            ``algorithms``), depending on the installed SDK version.

    Returns:
        The verified claims dictionary.

    Raises:
        RuntimeError: If ``tempest_fastapi_sdk`` is not installed. It is an
            optional, server-only dependency — install the ``[auth]`` extra.
    """
    try:
        from tempest_fastapi_sdk import JWTUtils  # type: ignore[import-not-found]
    except ImportError as exc:
        raise RuntimeError(
            "server_decode_jwt requires tempest-fastapi-sdk; install the "
            "[auth] extra to verify tokens on the server."
        ) from exc
    result: Any = JWTUtils.decode(token, secret, **kwargs)
    return dict(result)

default_fallback ¶

default_fallback(info: ErrorInfo) -> Widget

Render a minimal, renderer-agnostic fallback for a failed subtree.

Parameters:

Name	Type	Description	Default
`info`	`ErrorInfo`	The captured render failure.	required

Returns:

Name	Type	Description
`A`	`Widget`	class:`~tempest_core.Column` showing a generic apology and the
	`Widget`	error type (never the raw stack, which goes to the report hook).

Source code in tempestweb/observability/error_boundary.py

def default_fallback(info: ErrorInfo) -> Widget:
    """Render a minimal, renderer-agnostic fallback for a failed subtree.

    Args:
        info: The captured render failure.

    Returns:
        A :class:`~tempest_core.Column` showing a generic apology and the
        error type (never the raw stack, which goes to the report hook).
    """
    return Column(
        children=[
            Text(content="Something went wrong."),
            Text(content=f"({info.error_type})"),
        ]
    )

telemetry_reporter ¶

telemetry_reporter(provider: TelemetryProvider, *, event: str = 'render_error') -> ErrorReporter

Build a report hook that forwards captured errors to telemetry (O0).

Parameters:

Name	Type	Description	Default
`provider`	`TelemetryProvider`	The telemetry provider that receives the error event.	required
`event`	`str`	The telemetry event name to emit.	`'render_error'`

Returns:

Name	Type	Description
`An`	`ErrorReporter`	data:`ErrorReporter` that tracks `event` with the error type,
	`ErrorReporter`	message and stack as properties.

Source code in tempestweb/observability/error_boundary.py

def telemetry_reporter(
    provider: TelemetryProvider, *, event: str = "render_error"
) -> ErrorReporter:
    """Build a report hook that forwards captured errors to telemetry (O0).

    Args:
        provider: The telemetry provider that receives the error event.
        event: The telemetry event name to emit.

    Returns:
        An :data:`ErrorReporter` that tracks ``event`` with the error type,
        message and stack as properties.
    """

    def report(info: ErrorInfo) -> None:
        """Forward a captured render error to telemetry.

        Args:
            info: The captured render failure.

        Returns:
            None.
        """
        provider.track(
            event,
            {
                "error_type": info.error_type,
                "message": info.message,
                "stack": info.stack,
            },
        )

    return report

with_error_boundary ¶

with_error_boundary(*, fallback_builder: FallbackBuilder = default_fallback, on_error: ErrorReporter | None = None) -> Callable[[ChildBuilder], Callable[[], ErrorBoundary]]

Decorate a widget builder so it returns a boundary-wrapped component.

Parameters:

Name	Type	Description	Default
`fallback_builder`	`FallbackBuilder`	Builds the fallback subtree from the captured error.	`default_fallback`
`on_error`	`ErrorReporter \| None`	Optional report hook invoked on a render failure.	`None`

Returns:

Type	Description
`Callable[[ChildBuilder], Callable[[], ErrorBoundary]]`	A decorator that turns a `() -> Widget` builder into a ``() ->
`Callable[[ChildBuilder], Callable[[], ErrorBoundary]]`	ErrorBoundary`` builder, wrapping the original so its render errors are
`Callable[[ChildBuilder], Callable[[], ErrorBoundary]]`	contained.

Source code in tempestweb/observability/error_boundary.py

def with_error_boundary(
    *,
    fallback_builder: FallbackBuilder = default_fallback,
    on_error: ErrorReporter | None = None,
) -> Callable[[ChildBuilder], Callable[[], ErrorBoundary]]:
    """Decorate a widget builder so it returns a boundary-wrapped component.

    Args:
        fallback_builder: Builds the fallback subtree from the captured error.
        on_error: Optional report hook invoked on a render failure.

    Returns:
        A decorator that turns a ``() -> Widget`` builder into a ``() ->
        ErrorBoundary`` builder, wrapping the original so its render errors are
        contained.
    """

    def decorator(builder: ChildBuilder) -> Callable[[], ErrorBoundary]:
        """Wrap ``builder`` so calling it yields a protected :class:`ErrorBoundary`.

        Args:
            builder: The original widget builder to protect.

        Returns:
            A zero-argument callable producing an :class:`ErrorBoundary`.
        """

        @wraps(builder)
        def wrapped() -> ErrorBoundary:
            """Wrap the decorated builder in an :class:`ErrorBoundary`.

            The decorated function is not called here — it is handed over as
            ``child_builder``, so the boundary owns when (and whether) the child
            is built and can substitute the fallback if that build raises.

            Returns:
                The boundary standing in for the decorated builder.
            """
            return ErrorBoundary(
                child_builder=builder,
                fallback_builder=fallback_builder,
                on_error=on_error,
            )

        return wrapped

    return decorator

console_sink ¶

console_sink(record: LogRecord) -> None

Print a record to the console in a stable, greppable single-line format.

Parameters:

Name	Type	Description	Default
`record`	`LogRecord`	The structured record to print.	required

Returns:

Type	Description
`None`	None.

Source code in tempestweb/observability/logger.py

def console_sink(record: LogRecord) -> None:
    """Print a record to the console in a stable, greppable single-line format.

    Args:
        record: The structured record to print.

    Returns:
        None.
    """
    suffix: str = f" {record.fields}" if record.fields else ""
    print(f"[{record.level}] {record.message}{suffix}")

create_logger ¶

create_logger(sinks: list[LoggerSink] | None = None, level: LogLevel = 'INFO') -> Logger

Create a :class:Logger with the given sinks and threshold.

Parameters:

Name	Type	Description	Default
`sinks`	`list[LoggerSink] \| None`	The destinations to deliver records to. Defaults to a single :func:`console_sink` when omitted.	`None`
`level`	`LogLevel`	The minimum severity to deliver.	`'INFO'`

Returns:

Type	Description
`Logger`	A configured :class:`Logger`.

Source code in tempestweb/observability/logger.py

def create_logger(
    sinks: list[LoggerSink] | None = None, level: LogLevel = "INFO"
) -> Logger:
    """Create a :class:`Logger` with the given sinks and threshold.

    Args:
        sinks: The destinations to deliver records to. Defaults to a single
            :func:`console_sink` when omitted.
        level: The minimum severity to deliver.

    Returns:
        A configured :class:`Logger`.
    """
    return Logger(sinks=sinks if sinks is not None else [console_sink], level=level)

json_log_sink ¶

json_log_sink(record: LogRecord) -> None

Print one log record as a single JSON line.

A session log is only useful if it can be queried, and the console sink prints prose. This prints one object per line — the shape every log pipeline ingests — with the structured fields at the top level, so session_id is a field and not a substring.

Parameters:

Name	Type	Description	Default
`record`	`LogRecord`	The record to print.	required

Source code in tempestweb/observability/server.py

def json_log_sink(record: LogRecord) -> None:
    """Print one log record as a single JSON line.

    A session log is only useful if it can be queried, and the console sink prints
    prose. This prints one object per line — the shape every log pipeline ingests —
    with the structured fields at the top level, so ``session_id`` is a field and
    not a substring.

    Args:
        record: The record to print.
    """
    payload: dict[str, Any] = {
        "level": record.level,
        "message": record.message,
        **record.fields,
    }
    print(json.dumps(payload, sort_keys=True, default=str), flush=True)

noop_tracer ¶

noop_tracer() -> Tracer

The tracer used when an app asks for no tracing.

Returns:

Type	Description
`Tracer`	A tracer whose spans do nothing.

Source code in tempestweb/observability/server.py

def noop_tracer() -> Tracer:
    """The tracer used when an app asks for no tracing.

    Returns:
        A tracer whose spans do nothing.
    """
    return _NOOP_TRACER

otel_tracer ¶

otel_tracer(service_name: str = 'tempestweb') -> Tracer

Adapt OpenTelemetry as the tracer, importing it only when called.

The import lives inside the function on purpose: the tracing default must not make opentelemetry a dependency of every app that serves a page. Exporter and sampler configuration stay with OpenTelemetry itself (env vars or an SDK setup the app owns) — wrapping those would be a second, worse configuration surface.

Parameters:

Name	Type	Description	Default
`service_name`	`str`	The tracer name reported to the exporter.	`'tempestweb'`

Returns:

Type	Description
`Tracer`	A tracer backed by the OpenTelemetry API.

Raises:

Type	Description
`RuntimeError`	If `opentelemetry-api` is not installed, naming the extra that provides it.

Source code in tempestweb/observability/server.py

def otel_tracer(service_name: str = "tempestweb") -> Tracer:
    """Adapt OpenTelemetry as the tracer, importing it only when called.

    The import lives inside the function on purpose: the tracing default must not
    make ``opentelemetry`` a dependency of every app that serves a page. Exporter
    and sampler configuration stay with OpenTelemetry itself (env vars or an SDK
    setup the app owns) — wrapping those would be a second, worse configuration
    surface.

    Args:
        service_name: The tracer name reported to the exporter.

    Returns:
        A tracer backed by the OpenTelemetry API.

    Raises:
        RuntimeError: If ``opentelemetry-api`` is not installed, naming the extra
            that provides it.
    """
    try:
        import opentelemetry.trace as trace  # noqa: PLC0415
    except ImportError as exc:  # pragma: no cover — depends on the environment
        raise RuntimeError(
            "otel_tracer() needs opentelemetry-api: "
            'uv add "tempestweb[otel]" (or opentelemetry-api directly). '
            "Tracing is opt-in precisely so this import is not everyone's problem."
        ) from exc

    otel = trace.get_tracer(service_name)

    class _OtelTracer:
        """Adapter over the OpenTelemetry tracer."""

        @contextmanager
        def span(self, name: str, **attributes: Any) -> Iterator[Any]:  # noqa: ANN401 — otel span
            """Open an OpenTelemetry span.

            Args:
                name: The span name.
                **attributes: Initial attributes.

            Yields:
                The OpenTelemetry span.
            """
            with otel.start_as_current_span(name) as span:
                for key, value in attributes.items():
                    span.set_attribute(key, value)
                yield span

    return _OtelTracer()

tempestweb.observability¶