HTTP Client API Reference¶

class CaseInsensitiveDict(dict):
    """Case-insensitive key lookup ``dict`` subclass that preserves original casing.

    Provides case-insensitive HTTP header storage: ``d["Content-Type"]``
    and ``d["content-type"]`` resolve to the same slot, but iteration and
    wire serialisation yield the original casing the caller supplied.

    Internally the underlying ``dict`` stores ``{lowercase_key: value}``
    for O(1) lookups, while a parallel ``_keys`` mapping records
    ``{lowercase_key: original_key}`` for casing-preserving iteration.

    This is the type used for ``Response.headers``,
    ``StreamingResponse.headers``, and the internal ``req_headers`` dict
    that flows through ``_prepare_request``.  HTTP header names are
    case-insensitive per :rfc:`7230` \u00a73.2, but the wire format and
    echo tests expect the casing the caller supplied.

    It is a ``dict`` subclass, so it is accepted everywhere a ``dict``
    is expected.  Equality is case-insensitive on keys:
    ``CaseInsensitiveDict({"X-Foo": "bar"}) == {"x-foo": "bar"}``.
    """

    # Parallel store: lowercase_key → original_key supplied by the caller.
    # Kept in sync with the underlying dict at all times.
    _keys: dict[str, str]

    def __init__(
        self,
        data: dict | list[tuple[str, str]] | None = None,
        **kwargs: str,
    ) -> None:
        super().__init__()
        self._keys = {}
        if data is not None:
            self.update(data)
        if kwargs:
            self.update(kwargs)

    # ------------------------------------------------------------------ #
    # Core write operations — all funnel through __setitem__ / __delitem__
    # ------------------------------------------------------------------ #

    def __setitem__(self, key: str, value: str) -> None:  # type: ignore[override]
        lower = key.lower()
        self._keys[lower] = key  # preserve (or update) original casing
        super().__setitem__(lower, value)

    def __delitem__(self, key: str) -> None:
        lower = key.lower()
        self._keys.pop(lower, None)
        super().__delitem__(lower)

    # ------------------------------------------------------------------ #
    # Read operations — normalise lookup key to lowercase
    # ------------------------------------------------------------------ #

    def __getitem__(self, key: str) -> str:  # type: ignore[override]
        return super().__getitem__(key.lower())  # type: ignore[return-value]

    def __contains__(self, key: object) -> bool:
        return super().__contains__(key.lower() if isinstance(key, str) else key)

    def get(self, key: str, default: str | None = None) -> str | None:  # type: ignore[override]
        return super().get(key.lower(), default)  # type: ignore[return-value]

    def pop(self, key: str, *args: str) -> str:  # type: ignore[override]
        lower = key.lower()
        self._keys.pop(lower, None)
        return super().pop(lower, *args)  # type: ignore[return-value]

    def setdefault(self, key: str, default: str = "") -> str:  # type: ignore[override]
        if key.lower() not in self:
            self[key] = default
        return self[key]

    def update(  # type: ignore[override]
        self,
        data: dict | list[tuple[str, str]] | None = None,
        **kwargs: str,
    ) -> None:
        if data is not None:
            items = data.items() if hasattr(data, "items") else data
            for k, v in items:
                self[k] = v
        for k, v in kwargs.items():
            self[k] = v

    # ------------------------------------------------------------------ #
    # Iteration — yield original casing so wire format is preserved
    # ------------------------------------------------------------------ #

    def __iter__(self):  # type: ignore[override]
        for lower in super().__iter__():
            yield self._keys.get(lower, lower)

    def keys(self):  # type: ignore[override]
        return list(self.__iter__())

    def values(self):  # type: ignore[override]
        return list(super().values())

    def items(self):  # type: ignore[override]
        """Yield ``(original_key, value)`` pairs; preserves casing on the wire."""
        for lower, value in super().items():
            yield self._keys.get(lower, lower), value

    # ------------------------------------------------------------------ #
    # Equality — case-insensitive on keys
    # ------------------------------------------------------------------ #

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, dict):
            return NotImplemented
        if len(self) != len(other):
            return False
        # Use our case-insensitive .get() so "X-Foo" == "x-foo" for key lookups
        return all(self.get(k) == v for k, v in other.items())

    def __hash__(self) -> None:  # type: ignore[override]
        return None  # dicts are unhashable; satisfy type checkers

    # ------------------------------------------------------------------ #
    # Miscellaneous
    # ------------------------------------------------------------------ #

    def copy(self) -> "CaseInsensitiveDict":
        return CaseInsensitiveDict(self.items())

    def __reduce__(self):  # type: ignore[override]
        """Ensure pickle/copy reconstructs via __init__ to restore _keys."""
        return (type(self), (list(self.items()),))

    def __repr__(self) -> str:
        return f"{type(self).__name__}({dict(self.items())!r})"

class HttpClientError(Exception):
    """Base exception for all httpclient operations."""

class HTTPError(HttpClientError):
    """Raised on non-2xx status when raise_for_status() is called."""

    def __init__(self, status_code: int, body: str, url: str) -> None:
        self.status_code = status_code
        self.body = body
        self.url = url
        super().__init__(f"HTTP {status_code} for {url}")

class TooManyRedirects(HTTPError):
    """Raised when redirect limit is exceeded."""

    def __init__(self, url: str, max_redirects: int) -> None:
        super().__init__(0, "", url)
        self.max_redirects = max_redirects
        Exception.__init__(self, f"Too many redirects (>{max_redirects}) for {url}")

class HttpConnectionError(HttpClientError):
    """Raised on connection failures.

    Attributes:
        host: Remote hostname that the connection targeted.
        port: Remote port number.
        message: Human-readable error description.
    """

    def __init__(self, message: str, *, host: str = "", port: int = 0) -> None:
        self.host = host
        self.port = port
        self.message = message
        super().__init__(message)

class HttpTimeoutError(HttpClientError):
    """Raised on request timeout.

    Attributes:
        url: The URL that timed out.
        timeout: The timeout value in seconds that was exceeded.
        message: Human-readable error description.
    """

    def __init__(self, message: str, *, url: str = "", timeout: float = 0.0) -> None:
        self.url = url
        self.timeout = timeout
        self.message = message
        super().__init__(message)

class Socks5Error(HttpConnectionError):
    """Raised on SOCKS5 proxy handshake failures."""

class Response:
    """HTTP response object.

    Attributes:
        status_code: HTTP status code.
        headers: Response headers as dict (last value wins for duplicates).
        content: Raw response body as bytes.
        url: Final URL after redirects.
    """

    __slots__ = ("status_code", "headers", "content", "url", "_text", "_json")

    def __init__(
        self,
        status_code: int,
        headers: CaseInsensitiveDict,
        content: bytes,
        url: str,
    ) -> None:
        self.status_code = status_code
        self.headers = headers
        self.content = content
        self.url = url
        self._text: str | None = None
        self._json: Any = None

    @property
    def text(self) -> str:
        """Decode response body as text."""
        if self._text is None:
            encoding = self._guess_encoding()
            self._text = self.content.decode(encoding, errors="replace")
        return self._text

    def json(self) -> Any:
        """Parse response body as JSON."""
        if self._json is None:
            self._json = _json.loads(self.content)
        return self._json

    @property
    def ok(self) -> bool:
        """True if status_code is 2xx."""
        return 200 <= self.status_code < 300

    def raise_for_status(self) -> None:
        """Raise HTTPError if status is not 2xx."""
        if not self.ok:
            raise HTTPError(self.status_code, self.text, self.url)

    def _guess_encoding(self) -> str:
        return _guess_encoding_from_headers(self.headers)

    # ── Context managers (no-op, body is already fully read) ──

    def __enter__(self) -> Response:
        return self

    def __exit__(self, *args: Any) -> None:
        self.close()

    async def __aenter__(self) -> Response:
        return self

    async def __aexit__(self, *args: Any) -> None:
        self.close()

    def close(self) -> None:
        """No-op close for a fully-read response."""

    async def aclose(self) -> None:
        """No-op async close for a fully-read response."""

    def __repr__(self) -> str:
        return f"<Response [{self.status_code}]>"

class Auth:
    """Base class for HTTP authentication."""

    def auth_headers(self, method: str, url: str) -> dict[str, str]:
        """Return authorization headers.

        Args:
            method: HTTP method.
            url: Request URL.

        Returns:
            Dict of headers to add to the request.
        """
        raise NotImplementedError

class BasicAuth(Auth):
    """HTTP Basic authentication."""

    def __init__(self, username: str, password: str) -> None:
        self._username = username
        self._password = password

    def auth_headers(self, method: str, url: str) -> dict[str, str]:
        """Return Basic Authorization header."""
        credentials = f"{self._username}:{self._password}".encode()
        return {"Authorization": "Basic " + base64.b64encode(credentials).decode()}

class DigestAuth(Auth):
    """HTTP Digest authentication."""

    def __init__(self, username: str, password: str) -> None:
        self._username = username
        self._password = password
        self._nc = 0

    def auth_headers(self, method: str, url: str) -> dict[str, str]:
        """Not usable without a server challenge."""
        raise NotImplementedError("DigestAuth requires a server challenge")

    def auth_headers_from_challenge(
        self, method: str, path: str, challenge: str
    ) -> dict[str, str]:
        """Compute Digest auth headers from a WWW-Authenticate challenge.

        Args:
            method: HTTP method.
            path: Request path (URI).
            challenge: The WWW-Authenticate header value.

        Returns:
            Dict with the Authorization header.
        """
        params = _parse_digest_challenge(challenge)
        realm = params.get("realm", "")
        nonce = params.get("nonce", "")
        qop = params.get("qop", "")
        opaque = params.get("opaque", "")
        algorithm = params.get("algorithm", "MD5").upper()

        self._nc += 1
        nc_hex = f"{self._nc:08x}"
        cnonce = os.urandom(16).hex()

        if algorithm == "SHA-256":
            hash_fn = hashlib.sha256
        else:
            hash_fn = hashlib.md5

        ha1 = hash_fn(f"{self._username}:{realm}:{self._password}".encode()).hexdigest()
        ha2 = hash_fn(f"{method}:{path}".encode()).hexdigest()

        if qop == "auth":
            response = hash_fn(
                f"{ha1}:{nonce}:{nc_hex}:{cnonce}:{qop}:{ha2}".encode()
            ).hexdigest()
        else:
            response = hash_fn(f"{ha1}:{nonce}:{ha2}".encode()).hexdigest()

        header = (
            f'Digest username="{self._username}", realm="{realm}", '
            f'nonce="{nonce}", uri="{path}", response="{response}"'
        )
        if qop:
            header += f', qop={qop}, nc={nc_hex}, cnonce="{cnonce}"'
        if opaque:
            header += f', opaque="{opaque}"'
        header += f", algorithm={algorithm}"

        return {"Authorization": header}

class StreamingResponse:
    """HTTP streaming response -- holds the connection open.

    Use as a context manager to ensure cleanup::

        with get(url, stream=True) as r:
            for chunk in r.iter_bytes():
                process(chunk)

        async with await async_get(url, stream=True) as r:
            async for line in r.aiter_lines():
                handle(line)
    """

    __slots__ = (
        "status_code",
        "headers",
        "url",
        "_encoding",
        "_decompressor",
        "_sync_resp",
        "_sync_conn",
        "_async_reader",
        "_async_writer",
        "_async_timeout",
        "_is_chunked",
        "_content_length",
        "_bytes_remaining",
        "_closed",
    )

    status_code: int
    headers: CaseInsensitiveDict
    url: str
    _encoding: str
    _decompressor: zlib._Decompress | None
    _sync_resp: http.client.HTTPResponse | None
    _sync_conn: http.client.HTTPConnection | None
    _async_reader: asyncio.StreamReader | None
    _async_writer: asyncio.StreamWriter | None
    _async_timeout: float | None
    _is_chunked: bool
    _content_length: int | None
    _bytes_remaining: int | None
    _closed: bool

    def __init__(self) -> None:
        raise TypeError("Use _from_sync() or _from_async()")

    @classmethod
    def _from_sync(
        cls,
        status_code: int,
        headers: CaseInsensitiveDict,
        url: str,
        resp: http.client.HTTPResponse,
        conn: http.client.HTTPConnection,
        content_encoding: str = "",
    ) -> "StreamingResponse":
        obj = object.__new__(cls)
        obj.status_code = status_code
        obj.headers = headers
        obj.url = url
        obj._encoding = _guess_encoding_from_headers(headers)
        obj._decompressor = (
            _make_decompressor(content_encoding) if content_encoding else None
        )
        obj._sync_resp = resp
        obj._sync_conn = conn
        obj._async_reader = None
        obj._async_writer = None
        obj._async_timeout = None
        obj._is_chunked = False
        obj._content_length = None
        obj._bytes_remaining = None
        obj._closed = False
        return obj

    @classmethod
    def _from_async(
        cls,
        status_code: int,
        headers: CaseInsensitiveDict,
        url: str,
        reader: asyncio.StreamReader,
        writer: asyncio.StreamWriter,
        is_chunked: bool,
        content_length: int | None,
        timeout: float,
        content_encoding: str = "",
    ) -> "StreamingResponse":
        obj = object.__new__(cls)
        obj.status_code = status_code
        obj.headers = headers
        obj.url = url
        obj._encoding = _guess_encoding_from_headers(headers)
        obj._decompressor = (
            _make_decompressor(content_encoding) if content_encoding else None
        )
        obj._sync_resp = None
        obj._sync_conn = None
        obj._async_reader = reader
        obj._async_writer = writer
        obj._async_timeout = timeout
        obj._is_chunked = is_chunked
        obj._content_length = content_length
        obj._bytes_remaining = content_length
        obj._closed = False
        return obj

    @property
    def ok(self) -> bool:
        """True if status_code is 2xx."""
        return 200 <= self.status_code < 300

    def raise_for_status(self) -> None:
        """Raise HTTPError if status is not 2xx."""
        if not self.ok:
            raise HTTPError(self.status_code, "", self.url)

    # ── Sync iteration ──

    def iter_bytes(self, chunk_size: int = 4096) -> Iterator[bytes]:
        """Yield response body in chunks."""
        if self._sync_resp is None:
            raise RuntimeError("iter_bytes() on async response")
        try:
            while True:
                chunk = self._sync_resp.read(chunk_size)
                if not chunk:
                    break
                if self._decompressor:
                    chunk = self._decompressor.decompress(chunk)
                yield chunk
            if self._decompressor:
                remaining = self._decompressor.flush()
                if remaining:
                    yield remaining
        except (OSError, http.client.HTTPException) as exc:
            raise HttpConnectionError(str(exc)) from exc

    def iter_lines(self) -> Iterator[str]:
        """Yield response body line by line (decoded)."""
        if self._sync_resp is None:
            raise RuntimeError("iter_lines() on async response")
        try:
            while True:
                line = self._sync_resp.readline()
                if not line:
                    break
                yield line.decode(self._encoding, errors="replace").rstrip("\r\n")
        except (OSError, http.client.HTTPException) as exc:
            raise HttpConnectionError(str(exc)) from exc

    def read(self) -> bytes:
        """Consume entire stream into bytes."""
        return b"".join(self.iter_bytes())

    # ── Async iteration ──

    async def aiter_bytes(self, chunk_size: int = 4096) -> AsyncIterator[bytes]:
        """Async yield response body in chunks."""
        if self._async_reader is None:
            raise RuntimeError("aiter_bytes() on sync response")
        try:
            raw_iter = self._select_raw_iterator(chunk_size)
            async for chunk in raw_iter:
                if self._decompressor:
                    chunk = self._decompressor.decompress(chunk)
                yield chunk
            if self._decompressor:
                remaining = self._decompressor.flush()
                if remaining:
                    yield remaining
        except asyncio.TimeoutError:
            raise HttpTimeoutError(
                f"Streaming read timed out for {self.url}",
                url=self.url,
                timeout=self._async_timeout or 0.0,
            )
        except OSError as exc:
            raise HttpConnectionError(str(exc)) from exc

    async def _select_raw_iterator(self, chunk_size: int) -> AsyncIterator[bytes]:
        """Select and yield from the appropriate raw byte iterator."""
        if self._is_chunked:
            async for chunk in self._aiter_chunked():
                yield chunk
        elif self._bytes_remaining is not None:
            async for chunk in self._aiter_fixed_length(chunk_size):
                yield chunk
        else:
            async for chunk in self._aiter_until_eof(chunk_size):
                yield chunk

    async def _aiter_fixed_length(self, chunk_size: int) -> AsyncIterator[bytes]:
        """Read a known-length response body in chunks."""
        assert self._async_reader is not None
        while self._bytes_remaining is not None and self._bytes_remaining > 0:
            to_read = min(chunk_size, self._bytes_remaining)
            data = await asyncio.wait_for(
                self._async_reader.read(to_read),
                timeout=self._async_timeout,
            )
            if not data:
                break
            self._bytes_remaining -= len(data)
            yield data

    async def _aiter_until_eof(self, chunk_size: int) -> AsyncIterator[bytes]:
        """Read response body until EOF in chunks."""
        assert self._async_reader is not None
        while True:
            data = await asyncio.wait_for(
                self._async_reader.read(chunk_size),
                timeout=self._async_timeout,
            )
            if not data:
                break
            yield data

    async def _aiter_chunked(self) -> AsyncIterator[bytes]:
        """Decode chunked transfer encoding from async reader."""
        assert self._async_reader is not None  # guaranteed by aiter_bytes guard
        reader = self._async_reader
        timeout = self._async_timeout
        while True:
            size_line = await asyncio.wait_for(reader.readline(), timeout=timeout)
            size_str = size_line.decode("latin-1").split(";")[0].strip()
            if not size_str:
                break
            chunk_size = int(size_str, 16)
            if chunk_size == 0:
                await asyncio.wait_for(
                    reader.readline(), timeout=timeout
                )  # trailing \r\n
                break
            data = await asyncio.wait_for(
                reader.readexactly(chunk_size), timeout=timeout
            )
            await asyncio.wait_for(reader.readline(), timeout=timeout)  # trailing \r\n
            yield data

    async def aiter_lines(self) -> AsyncIterator[str]:
        """Async yield response body line by line (decoded)."""
        buf = ""
        async for chunk in self.aiter_bytes():
            buf += chunk.decode(self._encoding, errors="replace")
            while "\n" in buf:
                line, buf = buf.split("\n", 1)
                yield line.rstrip("\r")
        if buf:
            yield buf.rstrip("\r")

    async def aread(self) -> bytes:
        """Async consume entire stream into bytes."""
        parts = []
        async for chunk in self.aiter_bytes():
            parts.append(chunk)
        return b"".join(parts)

    # ── Context managers ──

    def __enter__(self) -> "StreamingResponse":
        return self

    def __exit__(self, *args: Any) -> None:
        self.close()

    async def __aenter__(self) -> "StreamingResponse":
        return self

    async def __aexit__(self, *args: Any) -> None:
        await self.aclose()

    def close(self) -> None:
        """Close the underlying sync connection."""
        if self._closed:
            return
        self._closed = True
        # Tier 2: best-effort observable -- active streaming resource
        if self._sync_resp is not None:
            try:
                self._sync_resp.close()
            except Exception:
                logger.debug(
                    "failed to close sync response for %s",
                    self.url,
                    exc_info=True,
                )
        if self._sync_conn is not None:
            try:
                self._sync_conn.close()
            except Exception:
                logger.debug(
                    "failed to close sync connection for %s",
                    self.url,
                    exc_info=True,
                )

    async def aclose(self) -> None:
        """Close the underlying async connection."""
        if self._closed:
            return
        self._closed = True
        # Tier 2: best-effort observable -- active streaming resource
        if self._async_writer is not None:
            try:
                self._async_writer.close()
                await self._async_writer.wait_closed()
            except Exception:
                logger.debug(
                    "failed to close async writer for %s",
                    self.url,
                    exc_info=True,
                )

    def __del__(self) -> None:
        if not self._closed:
            warnings.warn(
                f"Unclosed StreamingResponse for {self.url}",
                ResourceWarning,
                stacklevel=2,
            )
            self.close()

    def __repr__(self) -> str:
        return f"<StreamingResponse [{self.status_code}]>"