Skip to content

Resolver

Func module-attribute

Func = Callable[Concatenate[Any, P], Any]

EventRouter

EventRouter(*, allow_multiple_routes: bool = False, allow_no_route: bool = True)

Router for events, allowing registration of conditions and handlers.

Parameters:

Name Type Description Default
allow_multiple_routes bool

option to allow multiples routes on same event, otherwise raise MultipleRoutesError.

False
allow_no_route bool

option to allow no routes on event, otherwise raise NoRouteFoundError.

True

equal

equal(value_path: str, expected: Any) -> Callable[[Func[P]], Func[P]]

Register a route with an equality condition.

Parameters:

Name Type Description Default
value_path str

The path to the value in the event.

required
expected Any

The expected value.

required

one_of

one_of(value_path: str, options: Container[V]) -> Callable[[Func[P]], Func[P]]

Register a route with a one-of condition.

Parameters:

Name Type Description Default
value_path str

The path to the value in the event.

required
options Container[V]

The container of expected values.

required

contain

contain(value_path: str, *items: V) -> Callable[[Func[P]], Func[P]]

Register a route where value should contain items.

Parameters:

Name Type Description Default
value_path str

The path to the value in the event.

required
items V

Items to in the event.

()

when

when(condition: Condition) -> Callable[[Func[P]], Func[P]]

Register a route with a custom condition.

Parameters:

Name Type Description Default
condition Condition

The condition to trigger this route.

required

exception_handler

exception_handler(exc_type: type[Error]) -> Callable[[Callable[[Error], Any]], Callable[[Error], Any]]
exception_handler(exc_type: Sequence[type[Error]]) -> Callable[[Callable[..., Any]], Callable[..., Any]]
exception_handler(exc_type: type[Error] | Sequence[type[Error]]) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Register a function to handle certain exception type.

Parameters:

Name Type Description Default
exc_type type[Error] | Sequence[type[Error]]

exception type to handle. Could be either a unique exception type or a list of types.

required

fallback

fallback() -> Callable[[Func[P]], Func[P]]
fallback(func: Func[P]) -> Func[P]
fallback(func: Func[P] | None = None) -> Callable[[Func[P]], Func[P]] | Func[P]

Register a fallback route if no registered routes match the event.

EventResolver

EventResolver(*, allow_multiple_routes: bool = False, allow_no_route: bool = True)

Bases: EventRouter

Resolves events by finding and executing matching routes from registered routers.

This class extends EventRouter to provide a mechanism for resolving events against a collection of routes. It can include routes from other EventRouter instances and provides a central resolve method to process an event.

Parameters:

Name Type Description Default
allow_multiple_routes bool

option to allow multiples routes on same event, otherwise raise MultipleRoutesError.

False
allow_no_route bool

option to allow no routes on event, otherwise raise NoRouteFoundError.

True

equal

equal(value_path: str, expected: Any) -> Callable[[Func[P]], Func[P]]

Register a route with an equality condition.

Parameters:

Name Type Description Default
value_path str

The path to the value in the event.

required
expected Any

The expected value.

required

one_of

one_of(value_path: str, options: Container[V]) -> Callable[[Func[P]], Func[P]]

Register a route with a one-of condition.

Parameters:

Name Type Description Default
value_path str

The path to the value in the event.

required
options Container[V]

The container of expected values.

required

contain

contain(value_path: str, *items: V) -> Callable[[Func[P]], Func[P]]

Register a route where value should contain items.

Parameters:

Name Type Description Default
value_path str

The path to the value in the event.

required
items V

Items to in the event.

()

when

when(condition: Condition) -> Callable[[Func[P]], Func[P]]

Register a route with a custom condition.

Parameters:

Name Type Description Default
condition Condition

The condition to trigger this route.

required

exception_handler

exception_handler(exc_type: type[Error]) -> Callable[[Callable[[Error], Any]], Callable[[Error], Any]]
exception_handler(exc_type: Sequence[type[Error]]) -> Callable[[Callable[..., Any]], Callable[..., Any]]
exception_handler(exc_type: type[Error] | Sequence[type[Error]]) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Register a function to handle certain exception type.

Parameters:

Name Type Description Default
exc_type type[Error] | Sequence[type[Error]]

exception type to handle. Could be either a unique exception type or a list of types.

required

fallback

fallback() -> Callable[[Func[P]], Func[P]]
fallback(func: Func[P]) -> Func[P]
fallback(func: Func[P] | None = None) -> Callable[[Func[P]], Func[P]] | Func[P]

Register a fallback route if no registered routes match the event.

include_router

include_router(router: EventRouter, base_condition: Condition | None = None) -> None

Include router routes and exception handlers into this resolver.

Parameters:

Name Type Description Default
router EventRouter

The EventRouter instance to include.

required
base_condition Condition | None

An optional condition to apply to all routes from the included router. If provided, it will be combined with each route's existing condition.

None

resolve

resolve(event: Mapping[Any, V]) -> Sequence[Any]

Resolve the event to the matching routes and execute their functions.

Parameters:

Name Type Description Default
event Mapping[Any, V]

The event to resolve.

required