pythonnative/src/pythonnative/net.py at main · pythonnative/pythonnative

244 lines (200 loc) · 7.9 KB
"""Async HTTP client (``pn.fetch``).
A small, dependency-free coroutine wrapper around
:mod:`urllib.request`. Operates on bytes internally and exposes a
:class:`Response` with `text()`, `json()`, and `bytes` accessors.
The implementation is deliberately minimal — it covers the
"call a JSON API" path that's overwhelmingly the use case for mobile
apps. For streaming, multipart uploads, or HTTP/2, integrate
``httpx`` / ``aiohttp`` directly; this module won't try to compete.
    ```python
    import pythonnative as pn
    async def load_user(user_id):
        resp = await pn.fetch(
            f"https://api.example.com/users/{user_id}",
            headers={"Accept": "application/json"},
        resp.raise_for_status()
        return resp.json()
from __future__ import annotations
import asyncio
import json
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field
from typing import Any, Dict, Mapping, Optional, Union
# A package-level SSL context lets callers (or tests) override TLS
# verification globally without monkeypatching the stdlib. Defaults
# to the system trust store.
_default_ssl_context: Optional[ssl.SSLContext] = None
def set_default_ssl_context(context: Optional[ssl.SSLContext]) -> None:
    """Override the SSL context used by [`fetch`][pythonnative.fetch].
    ``None`` (the default) means ``urllib`` builds a context from
    the system trust store. Tests can pass an unverified context.
    global _default_ssl_context
    _default_ssl_context = context
# ======================================================================
# Response object
# ======================================================================
class Response:
    """The result of a [`fetch`][pythonnative.fetch] call.
    Attributes:
        status: HTTP status code (e.g. ``200``).
        url: Final URL after any redirects.
        headers: Response headers, case-insensitive.
        content: Raw response body.
    status: int
    url: str
    headers: Mapping[str, str] = field(default_factory=dict)
    content: bytes = b""
    @property
    def ok(self) -> bool:
        """``True`` if the status is in the 2xx range."""
        return 200 <= self.status < 300
    def text(self, encoding: Optional[str] = None) -> str:
        """Decode ``content`` to ``str``.
        Args:
            encoding: Optional override; defaults to the
                ``charset`` parameter of ``Content-Type`` (or UTF-8).
        """
        enc = encoding or _charset_from_headers(self.headers) or "utf-8"
        try:
            return self.content.decode(enc)
        except (UnicodeDecodeError, LookupError):
            return self.content.decode("utf-8", errors="replace")
    def json(self) -> Any:
        """Parse the response body as JSON.
        Raises:
            json.JSONDecodeError: If the body isn't valid JSON.
        """
        return json.loads(self.text())
    def raise_for_status(self) -> None:
        """Raise [`HTTPError`][pythonnative.net.HTTPError] for non-2xx responses."""
        if not self.ok:
            raise HTTPError(self.status, self.url, self.text())
class HTTPError(Exception):
    """Raised by [`Response.raise_for_status`][pythonnative.net.Response.raise_for_status]."""
    def __init__(self, status: int, url: str, body: str) -> None:
        super().__init__(f"HTTP {status} from {url}: {body[:200]}")
        self.status = status
        self.url = url
        self.body = body
def _charset_from_headers(headers: Mapping[str, str]) -> Optional[str]:
    ctype = headers.get("Content-Type") or headers.get("content-type")
    if not ctype:
        return None
    for part in ctype.split(";"):
        part = part.strip()
        if part.lower().startswith("charset="):
            return part[len("charset=") :].strip().strip('"')
    return None
# ======================================================================
# Public fetch coroutine
# ======================================================================
async def fetch(
    url: str,
    method: str = "GET",
    headers: Optional[Mapping[str, str]] = None,
    body: Union[bytes, str, Mapping[str, Any], None] = None,
    params: Optional[Mapping[str, Any]] = None,
    timeout: float = 30.0,
) -> Response:
    """Make an HTTP request and return a [`Response`][pythonnative.net.Response].
        url: Target URL. Relative URLs are not supported.
        method: HTTP method (``"GET"``, ``"POST"``, ``"PUT"`` …).
        headers: Optional request headers.
        body: Request body. ``bytes`` are sent as-is; ``str`` is
            UTF-8 encoded; ``dict`` is JSON-encoded with a
            ``Content-Type: application/json`` header added (unless
            already supplied).
        params: Optional mapping of query-string parameters appended
            to ``url`` (urlencoded).
        timeout: Seconds to wait for the response (excluding the
            time spent on DNS / connect).
    Returns:
        A [`Response`][pythonnative.net.Response].
    Raises:
        TimeoutError: If the request doesn't complete within
            ``timeout`` seconds.
        OSError: For network errors (DNS failure, connection refused,
            etc.) — re-raised from ``urllib``.
    Example:
        ```python
        resp = await pn.fetch(
            "https://api.example.com/posts",
            method="POST",
            body={"title": "Hello"},
        resp.raise_for_status()
        ```
    request = _build_request(url=url, method=method, headers=headers, body=body, params=params)
    return await asyncio.to_thread(_dispatch_request, request, timeout)
def _build_request(
    url: str,
    method: str,
    headers: Optional[Mapping[str, str]],
    body: Union[bytes, str, Mapping[str, Any], None],
    params: Optional[Mapping[str, Any]],
) -> urllib.request.Request:
    if params:
        sep = "&" if "?" in url else "?"
        url = url + sep + urllib.parse.urlencode(params, doseq=True)
    header_dict: Dict[str, str] = dict(headers or {})
    payload: Optional[bytes] = None
    if body is None:
        pass
    elif isinstance(body, (bytes, bytearray)):
        payload = bytes(body)
    elif isinstance(body, str):
        payload = body.encode("utf-8")
    elif isinstance(body, Mapping):
        payload = json.dumps(body, default=str).encode("utf-8")
        header_dict.setdefault("Content-Type", "application/json")
        raise TypeError(f"Unsupported body type: {type(body)!r}")
    return urllib.request.Request(
        url=url,
        data=payload,
        method=method.upper(),
        headers=header_dict,
def _dispatch_request(request: urllib.request.Request, timeout: float) -> Response:
    context = _default_ssl_context
        with urllib.request.urlopen(request, timeout=timeout, context=context) as resp:
            content = resp.read()
            return Response(
                status=resp.status,
                url=resp.geturl(),
                headers={k: v for k, v in resp.headers.items()},
                content=content,
    except urllib.error.HTTPError as exc:
        # HTTPError is itself a response object — propagate the body
        # so callers can inspect it before deciding to raise.
        body = exc.read() if hasattr(exc, "read") else b""
        return Response(
            status=exc.code,
            url=getattr(exc, "url", request.full_url),
            headers={k: v for k, v in (exc.headers or {}).items()},
            content=body,
    except urllib.error.URLError as exc:
        # Convert urllib's TimeoutError wrapper into a plain TimeoutError
        # so callers get a recognisable exception type.
        reason = getattr(exc, "reason", exc)
        if isinstance(reason, TimeoutError):
            raise reason
        raise OSError(str(reason)) from exc
__all__ = ["fetch", "Response", "HTTPError", "set_default_ssl_context"]
Provide feedback

Saved searches

Use saved searches to filter your results more quickly

FilesExpand file tree

net.py

Latest commit

History

net.py

File metadata and controls
