http_client.py

"""
Lightweight HTTP client — zero dependencies, Python 3.7+ standard library only.
"""

import json as _json
import time
import ssl
from urllib.request import Request, urlopen
from urllib.error import HTTPError as _HTTPError, URLError
from urllib.parse import urlencode, urljoin


class HTTPResponse:
    """Simple response wrapper."""

    def __init__(self, status_code, body, headers):
        self.status_code = status_code
        self._body = body
        self.headers = dict(headers)
        self.ok = 200 <= status_code < 300

    @property
    def text(self):
        return self._body.decode("utf-8", errors="replace")

    def json(self):
        return _json.loads(self.text)

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


class HTTPError(Exception):
    def __init__(self, status_code, message):
        self.status_code = status_code
        self.message = message
        super().__init__(f"HTTP {status_code}: {message}")


class TimeoutError(Exception):
    pass


class HTTPClient:
    """Zero-dependency HTTP client with retries and session support."""

    def __init__(self, base_url="", timeout=30, retries=0, backoff_factor=1.0):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.retries = retries
        self.backoff_factor = backoff_factor
        self.headers = {"User-Agent": "python-http-client/1.0"}
        self._ctx = ssl.create_default_context()

    def _build_url(self, path):
        if path.startswith("http"):
            return path
        return f"{self.base_url}/{path.lstrip('/')}"

    def request(self, method, path, json=None, data=None, headers=None,
                params=None, timeout=None, retries=None, backoff=None):
        url = self._build_url(path)
        if params:
            url += ("&" if "?" in url else "?") + urlencode(params)

        req_headers = {**self.headers, **(headers or {})}
        body = None

        if json is not None:
            body = _json.dumps(json).encode("utf-8")
            req_headers["Content-Type"] = "application/json"
        elif data is not None:
            body = urlencode(data).encode("utf-8")
            req_headers["Content-Type"] = "application/x-www-form-urlencoded"

        req = Request(url, data=body, headers=req_headers, method=method.upper())
        _timeout = timeout or self.timeout
        _retries = retries if retries is not None else self.retries
        _backoff = backoff or self.backoff_factor

        last_err = None
        for attempt in range(_retries + 1):
            try:
                resp = urlopen(req, timeout=_timeout, context=self._ctx)
                return HTTPResponse(resp.status, resp.read(), resp.headers)
            except _HTTPError as e:
                return HTTPResponse(e.code, e.read(), e.headers)
            except URLError as e:
                if "timed out" in str(e):
                    last_err = TimeoutError(str(e))
                else:
                    last_err = e
            if attempt < _retries:
                time.sleep(_backoff * (2 ** attempt))

        raise last_err

    def get(self, path, **kw):
        return self.request("GET", path, **kw)

    def post(self, path, **kw):
        return self.request("POST", path, **kw)

    def put(self, path, **kw):
        return self.request("PUT", path, **kw)

    def delete(self, path, **kw):
        return self.request("DELETE", path, **kw)

    def patch(self, path, **kw):
        return self.request("PATCH", path, **kw)


if __name__ == "__main__":
    client = HTTPClient()
    r = client.get("https://httpbin.org/get")
    print(f"Status: {r.status_code}")
    print(f"OK: {r.ok}")
    print(f"Body preview: {r.text[:200]}")