Skip to content

Event handler decorator

P = ParamSpec('P') module-attribute

WRT = AsyncIterator[EventType | None] module-attribute

logger = logging.getLogger(__name__) module-attribute

temporary_value_store = {} module-attribute

AbortCallError

Bases: Exception

Raise this to abort the call without raising error further.

AlreadyRunningError

Bases: FrontendError

Error raised when trying to start a decorated event that is already running.

ErrorCallProtocol

Bases: Protocol

__call__(state, exc) async

PostCallProtocol

Bases: Protocol

__call__(state) async

PreCallProtocol

Bases: Protocol

__call__(state, *args) async

event_handler_decorator(event_handler=None, *, pre_call=None, post_call=None, error_handler=None) 

event_handler_decorator(
    event_handler: None = None,
    *,
    pre_call: PreCallProtocol | None = None,
    post_call: PostCallProtocol | None = None,
    error_handler: ErrorCallProtocol | None = None,
) -> Callable[[Callable[P, object]], Callable[P, WRT]]

event_handler_decorator(
    event_handler: Callable[P, object],
    *,
    pre_call: PreCallProtocol | None = None,
    post_call: PostCallProtocol | None = None,
    error_handler: ErrorCallProtocol | None = None,
) -> Callable[P, WRT]

General decorator for event handlers.

Handles NotificationErrors (aborting the call if raised in the pre_call).

Works with all types of event handler
  • sync return
  • async return
  • sync yield
  • async yield
  • background return
  • background yield
PARAMETER DESCRIPTION
event_handler

The event handler to decorate.

TYPE: Callable[P, object] | None DEFAULT: None

pre_call

A function that is called before the event handler with event handlers state and call args.

TYPE: PreCallProtocol | None DEFAULT: None

post_call

A function that is called after a successful event handler call with event handler state

TYPE: PostCallProtocol | None DEFAULT: None

error_handler

A function that is called with any errors raised by the event handler.

TYPE: ErrorCallProtocol | None DEFAULT: None

handle_notification_errors(func)

Decorator that adds error handling for NotificationErrors.

Unwraps the notification and returns it if the error is a NotificationError. Otherwise, propagates the error.

in_background_event(state, attr_to_test)

Check if in background event by trying to set an attribute on the state (as a noop).

set_attr_value_during_event(attr_name, value_during=_NOT_SET, value_on_success=_NOT_SET, value_on_error=_NOT_SET, unique_key_override=None)

Event handler decorator that sets a value on the state during the event.

Useful to e.g. set a running status that will be reset regardless of errors during the event.

Note: Only works on background event handlers (otherwise nothing is changed until the end of the event anyway).

Note: Uses the attr name (or unique_key_override) along with the current session_id to store the value temporarily. If the value already exists in the temporary store, an error will be raised (i.e. only one instance of the event can be running per session at a time).

Note: Either post_call or error_handler is called, not both.

PARAMETER DESCRIPTION
attr_name

Name of the attribute to set on the event_handler_state.

TYPE: str

value_during

Value to set during the event. (default won't set anything during)

TYPE: Any DEFAULT: _NOT_SET

value_on_success

Value to set on success. (default will return value to original)

TYPE: Any DEFAULT: _NOT_SET

value_on_error

Value to set on error. (default will set to same value as on_success)

TYPE: Any DEFAULT: _NOT_SET

unique_key_override

Instead of using the attr_name, use this instead for temporarily storing the value.

TYPE: str | None DEFAULT: None

set_state_attr(state, attr_name, attr_val) async

Set an attribute on the state.

Handles case where state being set in background task