ai-flow-python/app/util/http_util.py

import logging
from typing import Any

import httpx

logger = logging.getLogger("app.http")

_MAX_BODY_LOG_LENGTH = 2000
_BODY_PREVIEW_LENGTH = 500


def _truncate(text: str, max_len: int = _MAX_BODY_LOG_LENGTH) -> str:
    if len(text) <= max_len:
        return text
    return text[:max_len] + f"... (truncated, total {len(text)} chars)"


def _body_summary(text: str) -> str:
    if not text:
        return ""
    if len(text) > _MAX_BODY_LOG_LENGTH:
        return f"<omitted, body too large: {len(text)} chars>"
    return _truncate(text, _BODY_PREVIEW_LENGTH)


def _on_request(request: httpx.Request) -> None:
    body = request.content.decode("utf-8", errors="replace") if request.content else ""
    logger.info(
        "[HTTP >>> ] %s %s | body(%d): %s",
        request.method,
        request.url,
        len(body),
        _body_summary(body),
    )


def _on_response(response: httpx.Response) -> None:
    response.read()
    body = response.text
    logger.info(
        "[HTTP <<< ] %s %s | body(%d): %s",
        response.status_code,
        response.url,
        len(body),
        _body_summary(body),
    )


def logged_client(**kwargs: Any) -> httpx.Client:
    """Create an ``httpx.Client`` that logs every request body and response body."""
    hooks = kwargs.pop("event_hooks", {})
    req_hooks = list(hooks.get("request", []))
    resp_hooks = list(hooks.get("response", []))
    req_hooks.append(_on_request)
    resp_hooks.append(_on_response)
    kwargs["event_hooks"] = {"request": req_hooks, "response": resp_hooks}
    return httpx.Client(**kwargs)