octomachinery.routing.routers module

Octomachinery event dispatchers collection.

class octomachinery.routing.routers.ConcurrentRouter(*other_routers: Router)

Bases: GidgetHubRouterBase

GitHub event router invoking event handlers simultaneously.

async dispatch(event: GitHubEvent, *args: Any, **kwargs: Any) → None

Invoke coroutine callbacks for the given event together.

class octomachinery.routing.routers.GidgetHubRouterBase(*other_routers: Router)

Bases: Router, OctomachineryRouterBase

GidgetHub-based router exposing callback matching separately.

async dispatch(event: GidgetHubWebhookEvent | Event, *args: Any, **kwargs: Any) → None

Invoke handler tasks for the given event sequentially.

emit_routes_for(event_name: str, event_payload: Any) → Iterator[Callable[[_P], Awaitable[None]]]

Emit callbacks that match given event and payload.

Parameters:

• event_name (str) – name of the GitHub event

• event_payload (str) – details of the GitHub event

Yields:

coroutine event handlers

class octomachinery.routing.routers.NonBlockingConcurrentRouter(*args, **kwargs)

Bases: ConcurrentRouter

Non-blocking GitHub event router scheduling handler tasks.

async dispatch(event: GitHubEvent, *args: Any, **kwargs: Any) → None

Schedule coroutine callbacks for the given event together.