Easy mechanism for logging response bodies using event hooks?

[simonw]

Today I found myself needing to use the event hooks mechanism to log full details of the request and responses running through HTTPX. I wrote this up in detail here: https://til.simonwillison.net/httpx/openai-log-requests-responses

Logging the request and response headers was easy. The thing I had trouble with was logging the response body.

Here's what I tried first:

import textwrap

def log_response(response: httpx.Response):
    request = response.request
    print(f"Request: {request.method} {request.url}")
    print(f"  Headers: {request.headers}")
    print("  Body:")
    print(textwrap.indent(request.content.decode(), "    "))
    print(f"Response: status_code={response.status_code}")
    print(f"  Headers: {response.headers}")
    print("  Body:")
    print(textwrap.indent(response.content.decode(), "    "))

client = openai.OpenAI(
    http_client=httpx.Client(
        event_hooks={
            "response": [log_response]
        }
    )
)

This didn't quite work. The problem is that the OpenAI client library uses streaming responses, and accessing response.content in an event hook breaks that - it consumes the full stream, such that subsequent calls to iter_lines() and friends fail to return content.

I ended up putting together a nasty hack, described in detail in my TIL but effectively looking like this:

from httpx._transports.default import ResponseStream

class _LoggingStream(ResponseStream):
    def __iter__(self) -> Iterator[bytes]:
        for chunk in super().__iter__():
            print(f"    {chunk.decode()}")
            yield chunk


def log_response(response: httpx.Response):
    # ... previous code goes here
    response.stream._stream = _LoggingStream(response.stream._stream)

Effectively I'm patching the response.stream to print out chunks as it iterates over them.

This is nasty. It's using undocumented HTTPX internals, and it feels rough and unpleasant.

Is there a better pattern I could use here? If not, could HTTPX make any small changes that would make this easier to achieve?

[karpetrosyan]

Hi!

You can try something like this:

import httpx


class LogResponse(httpx.Response):
    def iter_bytes(self, *args, **kwargs):
        for chunk in super().iter_bytes(*args, **kwargs):
            print(chunk)
            yield chunk


class LogTransport(httpx.BaseTransport):
    def __init__(self, transport: httpx.BaseTransport):
        self.transport = transport

    def handle_request(self, request: httpx.Request) -> httpx.Response:
        response = self.transport.handle_request(request)

        return LogResponse(
            status_code=response.status_code,
            headers=response.headers,
            stream=response.stream,
            extensions=response.extensions,
        )


transport = LogTransport(httpx.HTTPTransport())
client = httpx.Client(transport=transport)

response = client.get("https://www.youtube.com/")

Also, you can chain transports using transport argument, if you already have some.

[simonw]

Thank you! That's a much neater solution than mine, especially since it avoids using any undocumented internals.

I updated my TIL and credited you there: https://til.simonwillison.net/httpx/openai-log-requests-responses#user-content-a-better-solution