45 lines
1.6 KiB
Python
45 lines
1.6 KiB
Python
"""Bounded HTTP response helpers for public structured-data adapters."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
def read_bounded_response_text(response: Any, *, max_bytes: int) -> str:
|
|
"""Read a public response without buffering beyond the configured limit."""
|
|
status_code = int(getattr(response, "status_code", 0) or 0)
|
|
if status_code != 200:
|
|
raise ValueError(f"public_source_http_{status_code}")
|
|
|
|
content_length = str(getattr(response, "headers", {}).get("Content-Length") or "")
|
|
if content_length.isdigit() and int(content_length) > max_bytes:
|
|
raise ValueError("public_source_response_too_large")
|
|
|
|
iter_content = getattr(response, "iter_content", None)
|
|
if callable(iter_content):
|
|
chunks: list[bytes] = []
|
|
size = 0
|
|
try:
|
|
for chunk in iter_content(chunk_size=64 * 1024):
|
|
if not chunk:
|
|
continue
|
|
size += len(chunk)
|
|
if size > max_bytes:
|
|
raise ValueError("public_source_response_too_large")
|
|
chunks.append(chunk)
|
|
finally:
|
|
close = getattr(response, "close", None)
|
|
if callable(close):
|
|
close()
|
|
content = b"".join(chunks)
|
|
else:
|
|
content = getattr(response, "content", None)
|
|
if content is None:
|
|
content = str(getattr(response, "text", "") or "").encode("utf-8")
|
|
if len(content) > max_bytes:
|
|
raise ValueError("public_source_response_too_large")
|
|
return content.decode("utf-8", "ignore")
|
|
|
|
|
|
__all__ = ["read_bounded_response_text"]
|