Skip to content

Dispatching Webhooks Internally

fastapi_patterns.dispatching

Helpers for redispatching validated payloads through FastAPI.

Use DispatchState in conjunction with the Lifespan manager to enable dispatching of new requests through the FastAPI machinery complete with routing, middleware execution, and dependency injection. This enables request processing similar to fastapi.BackgroundTasks except that it creates a new ASGI scope and clean dependency chain.

Type Aliases:

Name Description
DispatchMessage

Callable for dispatching a new ASGI message.

Classes:

Name Description
DispatchState

Manages the dispatching and lifecycle of asynchronous tasks.

Attributes:

Name Type Description
DispatchTaskRunner

Dependency injection for [DispatchMessage][..DispatchMessage].

DispatchTaskRunner = t.Annotated[DispatchMessage, fastapi.Depends(_get_dispatch_task_runner)] module-attribute

Dependency injection for DispatchMessage.

DispatchMessage = t.Callable[[str, str, pydantic.BaseModel], None]

Callable for dispatching a new ASGI message.

Use the DispatchTaskRunner dependency to inject this into anywhere that FastAPI's dependency injection is available. The callable creates an independent ASGI scope and schedules a new task using the DispatchState lifespan object.

Parameters:

Name Type Description Default
task_name str

name to assign to the new task

 required
path str

path to the target endpoint

 required
payload BaseModel

validated request payload

 required

DispatchState(app)

Manages the dispatching and lifecycle of asynchronous tasks.

This class provides functionality to schedule, track, and clean up asynchronous tasks linked to specific requests in the application. It integrates with the FastAPI lifecycle, logging task-related events, and ensuring proper resource cleanup.

Methods:

Name Description
schedule_request

Schedule a request task for execution by the application.
task_finished

Handle task completion.

schedule_request(task_name, receive_message, scope)

Schedule a request task for execution by the application.

A new asyncio task is created and scheduled by calling the application instance with the provided scope and message generator. Response handling is disabled by passing a blackhole send function. Task completion is handled by registering task_finished as a callback.

Parameters:

Name Type Description Default
task_name str

name of the task to schedule

 required
receive_message Receive

ASGI message generator

 required
scope Scope

ASGI scope

 required

task_finished(task)

Handle task completion.

Simple function that ensures that task completion is logged particularly when an exception has occurred.