Core Events

Tags: events

This document describes the events dispatched by HARP Proxy’s core.

You can read about the concept and mechanics of the event-driven architecture of HARP Proxy in the contributor’s guide.

Configuration Events

During the setup and teardown phase, the harp.config module dispatches events to allow applications and other components to register themselves with the system.

You can add listeners to those events using the Application Protocol.

⚡️ EVENT_BIND

Dispatched by SystemBuilder.dispatch_bind_event(...) when the container is being configured.

Its main purpose is to allow applications to define services, by registering some service definitions with the container.

Dispatched as EVENT_BIND with a OnBindEvent instance.

Example

Here is an example of a listener coroutine for the EVENT_BIND event:

from whistle import AsyncEventDispatcher

from harp.config.events import EVENT_BIND, OnBindEvent


async def on_bind(event: OnBindEvent):
    print("System is being bound")


if __name__ == "__main__":
    # for example completeness only, you should use the system dispatcher
    dispatcher = AsyncEventDispatcher()
    dispatcher.add_listener(EVENT_BIND, on_bind)

⚡️ EVENT_BOUND

Dispatched by SystemBuilder.dispatch_bound_event(...) after the container has been compiled to a provider. At this point, all service dependencies are resolved, instances can be requested from the provider.

Its main purpose is to allow applications to instanciate and manipulate live services on startup.

Dispatched as EVENT_BOUND with a OnBoundEvent instance.

Example

Here is an example of a listener coroutine for the EVENT_BOUND event:

from whistle import AsyncEventDispatcher

from harp.config.events import EVENT_BOUND, OnBoundEvent


async def on_bound(event: OnBoundEvent):
    print("System is bound")


if __name__ == "__main__":
    # for example completeness only, you should use the system dispatcher
    dispatcher = AsyncEventDispatcher()
    dispatcher.add_listener(EVENT_BOUND, on_bound)

⚡️ EVENT_READY

Dispatched by SystemBuilder.dispatch_ready_event(...) after the system has been fully assembled and is (about to be) ready to start processing requests.

Dispatched as EVENT_READY with a OnReadyEvent instance.

The soon-to-be-served ASGI Application is available here, and this event is mostly used to decorate it with ASGI middlewares (e.g. Sentry or Prometheus integrations).

Example

Here is an example of a listener coroutine for the EVENT_READY event:

from whistle import AsyncEventDispatcher

from harp.config.events import EVENT_READY, OnReadyEvent


async def on_ready(event: OnReadyEvent):
    print("System is ready")


if __name__ == "__main__":
    # for example completeness only, you should use the system dispatcher
    dispatcher = AsyncEventDispatcher()
    dispatcher.add_listener(EVENT_READY, on_ready)

⚡️ EVENT_SHUTDOWN

Dispatched by System.dispose(...) when the system is being shut down.

Dispatched as EVENT_SHUTDOWN with a OnShutdown instance.

This event purpose is to allow applications to clean up resources on shutdown. For example, if applications define background asynchronous tasks, it’s a good idea to terminate them here.

Example

Here is an example of a listener coroutine for the EVENT_SHUTDOWN event:

from whistle import AsyncEventDispatcher

from harp.config.events import EVENT_SHUTDOWN, OnShutdownEvent


async def on_shutdown(event: OnShutdownEvent):
    print("System is shutting down")


if __name__ == "__main__":
    dispatcher = AsyncEventDispatcher()
    dispatcher.add_listener(EVENT_SHUTDOWN, on_shutdown)

⇄️ Sequence Diagram

Todo

Add sequence diagram

🌲 Class Diagram

classes whistle.event.Event Event dispatcher : NoneType name : NoneType propagation_stopped : bool stop_propagation() harp.config.events.OnBindEvent OnBindEvent container : Container name : str settings : GlobalSettings harp.config.events.OnBindEvent->whistle.event.Event harp.config.events.OnBoundEvent OnBoundEvent name : str provider : Services resolver : str harp.config.events.OnBoundEvent->whistle.event.Event harp.config.events.OnReadyEvent OnReadyEvent binds : list[Bind] kernel : ASGIKernel name : str provider : Services harp.config.events.OnReadyEvent->whistle.event.Event harp.config.events.OnShutdownEvent OnShutdownEvent kernel : ASGIKernel name : str provider : Services harp.config.events.OnShutdownEvent->whistle.event.Event

Core / ASGI Events

During the lifecycle of an ASGI Request, the harp.asgi module dispatches events to allow (low-level) applications to process or filter inbound requests and outbound responses.

Note

The ASGI events are rather low-level, and are usually only used to implement framework-level features by the HARP Core. You should not need to use them in your application code, or at least, it should not be the first thing you go for.

If you need to hook into the request/response lifecycle, you are probably better of using either the Proxy Events for inbound request processing (and their associated responses), or the Http Client Events for outgoing requests (and their associated responses).

⚡️ EVENT_CORE_STARTED

Dispatched by the ASGIKernel when the “lifespan.startup” ASGI message is received.

It happens once per process, before any other ASGI messages are recevived.

Dispatched as EVENT_CORE_STARTED with a whistle.Event instance (default event class that contains no specific context).

Example

Here is an example of a listener coroutine for the EVENT_CORE_STARTED event:

from whistle import AsyncEventDispatcher, Event

from harp.asgi.events import EVENT_CORE_STARTED


async def on_core_started(event: Event):
    print(f"ASGI Core started: {event}")


if __name__ == "__main__":
    # for example completeness only, you should use the system dispatcher
    dispatcher = AsyncEventDispatcher()
    dispatcher.add_listener(EVENT_CORE_STARTED, on_core_started)

See also

ASGI Lifespan Protocol (from ASGI Specicication)

⚡️ EVENT_CORE_REQUEST

Dispatched by the ASGIKernel when an inbound HttpRequest is received, before anything is done with it.

Listeners can use event.set_controller(...), bypassing further controller resolution.

Dispatched as EVENT_CORE_REQUEST with a RequestEvent instance.

Example

Here is an example of a listener coroutine for the EVENT_CORE_REQUEST event:

from whistle import AsyncEventDispatcher

from harp.asgi.events import EVENT_CORE_REQUEST, RequestEvent


async def on_core_request(event: RequestEvent):
    print(f"Request received: {event}")


if __name__ == "__main__":
    # for example completeness only, you should use the system dispatcher
    dispatcher = AsyncEventDispatcher()
    dispatcher.add_listener(EVENT_CORE_REQUEST, on_core_request)

⚡️ EVENT_CORE_CONTROLLER

Dispatched by the ASGIKernel when a controller callable has been resolved by the kernel’s controller resolver.

It is used to eventually modify the controller, for example with decorators, or change it altogether.

Dispatched as EVENT_CORE_CONTROLLER with a ControllerEvent instance.

Example

Here is an example of a listener coroutine for the EVENT_CORE_CONTROLLER event:

from whistle import AsyncEventDispatcher

from harp.asgi.events import EVENT_CORE_REQUEST, ControllerEvent


async def on_core_controller(event: ControllerEvent):
    print(f"Controller requested: {event}")


if __name__ == "__main__":
    # for example completeness only, you should use the system dispatcher
    dispatcher = AsyncEventDispatcher()
    dispatcher.add_listener(EVENT_CORE_REQUEST, on_core_controller)

⚡️ EVENT_CORE_VIEW

EVENT_CORE_VIEW is dispatched by the ASGIKernel when a controller callable has been called but it did not return an HttpResponse.

It is used to implement custom response handlers, for example dictionaries return values.

If after it has been fully dispatched, the event does not contain a response, then a HTTP 500 response is returned.

Dispatched as EVENT_CORE_VIEW with a ViewEvent instance.

Example

Here is an example of a listener coroutine for the EVENT_CORE_VIEW event:

from whistle import AsyncEventDispatcher

from harp.asgi.events import EVENT_CORE_VIEW, ViewEvent


async def on_core_view(event: ViewEvent):
    print(f"View received: {event}")


if __name__ == "__main__":
    # for example completeness only, you should use the system dispatcher
    dispatcher = AsyncEventDispatcher()
    dispatcher.add_listener(EVENT_CORE_VIEW, on_core_view)

⚡️ EVENT_CORE_RESPONSE

EVENT_CORE_RESPONSE is dispatched by the ASGIKernel when an outbound HttpResponse is about to be sent.

Listeners can use event.response = ... event attribute to change the response.

Dispatched as EVENT_CORE_RESPONSE with a ResponseEvent instance.

Example

Here is an example of a listener coroutine for the EVENT_CORE_RESPONSE event:

from whistle import AsyncEventDispatcher

from harp.asgi.events import EVENT_CORE_RESPONSE, ResponseEvent


async def on_core_response(event: ResponseEvent):
    print(f"Response received: {event}")


if __name__ == "__main__":
    # for example completeness only, you should use the system dispatcher
    dispatcher = AsyncEventDispatcher()
    dispatcher.add_listener(EVENT_CORE_RESPONSE, on_core_response)

⇄️ Sequence Diagram

Todo

Add sequence diagram

🌲 Class Diagram

classes harp.asgi.events.ControllerEvent ControllerEvent name : str harp.asgi.events.RequestEvent RequestEvent controller name : str request set_controller(controller: Optional[Callable]) harp.asgi.events.ControllerEvent->harp.asgi.events.RequestEvent whistle.event.Event Event dispatcher : NoneType name : NoneType propagation_stopped : bool stop_propagation() harp.asgi.events.RequestEvent->whistle.event.Event harp.asgi.events.ResponseEvent ResponseEvent name : str response : str harp.asgi.events.ResponseEvent->harp.asgi.events.RequestEvent harp.asgi.events.ViewEvent ViewEvent name : str response : Optional['HttpResponse'] value : Any set_response(response: 'HttpResponse') harp.asgi.events.ViewEvent->harp.asgi.events.RequestEvent