HTTP Client Events

Tags: events

The http_client application dispatches a few events to interract with outbound HTTP requests and associated responses.

Filtering Events

Whenever an external HTTP request goes through the :service:`http_client` service to be sent out (assuming you’re using the default service implementation), the harp_apps.http_client.transport.AsyncFilterableTransport transport wrapper will dispach events to filter both httpx.Request and httpx.Response instances that represents the HTTP request and response, respectively.

⚡️ EVENT_FILTER_HTTP_CLIENT_REQUEST

Dispatched when an outbound HTTP request is about to be sent.

It is dispatched as EVENT_FILTER_HTTP_CLIENT_REQUEST with a HttpClientFilterEvent instance.

If a response is set on the event instance, then the actual outgoing request will be bypassed and the forged response will be returned.

Example

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

from whistle import AsyncEventDispatcher

from harp_apps.http_client.events import EVENT_FILTER_HTTP_CLIENT_REQUEST, HttpClientFilterEvent


async def on_filter_request(event: HttpClientFilterEvent):
    print(f"Filtering request: {event}")


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

⚡️ EVENT_FILTER_HTTP_CLIENT_RESPONSE

Dispatched when an inbound HTTP response is received (either from the external service, or because it was forged).

It is dispatched as EVENT_FILTER_HTTP_CLIENT_RESPONSE, with the same HttpClientFilterEvent instance as the previous event, with the response set.

You can change the response instance on the event to modify the response that will be returned to the caller.

Example

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

from whistle import AsyncEventDispatcher

from harp_apps.http_client.events import EVENT_FILTER_HTTP_CLIENT_RESPONSE, HttpClientFilterEvent


async def on_filter_response(event: HttpClientFilterEvent):
    print(f"Filtering response: {event}")


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

Gotchas

Those events are only dispatched if a request to an external service is ready to be sent. It will be bypassed if a response has been forged before, or if the request hits the cache. If you’re looking to filter incoming requests to the proxy, have a look at Proxy Events.

Please note that the same event instance will be used for both events. It means that if you stop propagation of the event, all further filtering will be skipped, for both events.