Skip to content

aea.helpers.http_requests

Minimal HTTP helpers backed by :mod:urllib from the standard library.

Replaces the requests package for the few HTTP calls the core framework needs (registry API, package downloads, GitHub tag fetches).

_ConditionalNoRedirectHandler Objects

class _ConditionalNoRedirectHandler(urllib.request.HTTPRedirectHandler)

Match requests' default redirect behaviour per HTTP method.

requests.get() follows redirects by default (allow_redirects=True), while requests.post()/put()/delete()/patch() do NOT follow redirects (allow_redirects=False). urllib.request's default follows everything — including silently turning POST→GET on 30x, which is a silent behaviour change. This handler suppresses redirects only for unsafe methods and lets the default HTTPRedirectHandler follow them for safe methods like GET/HEAD.

Also backports 308 (Permanent Redirect) handling: Python 3.11+ handles 308 natively in HTTPRedirectHandler, but 3.10 is missing http_error_308 AND redirect_request does not include 308 in its allowed-codes tuple. Without this backport, a CDN returning 308 on a GET would surface as HTTPResponse(308, ...) and break downstream callers like download_file.

redirect_request

def redirect_request(req: urllib.request.Request, fp: Any, code: int, msg: str,
                     headers: Any,
                     newurl: str) -> Optional[urllib.request.Request]

Suppress redirects for unsafe methods, follow for safe ones.

Arguments:

  • req: the original request.
  • fp: the response file-like object.
  • code: HTTP status code.
  • msg: HTTP status message.
  • headers: response headers.
  • newurl: the redirect target URL.

Returns:

a new Request to follow, or None to suppress.

http_error_308

def http_error_308(req: urllib.request.Request, fp: Any, code: int, msg: str,
                   headers: Any) -> Optional[Any]

Treat 308 the same as 301 (permanent redirect).

Python 3.11+ provides this natively; this backport makes 308 work on 3.10.

Arguments:

  • req: the original request.
  • fp: the response file-like object.
  • code: HTTP status code (308).
  • msg: HTTP status message.
  • headers: response headers.

Returns:

redirected response or None.

HTTPResponse Objects

class HTTPResponse()

Lightweight response wrapper matching the subset of requests.Response used in aea.

__init__

def __init__(status_code: int, data: bytes, url: str = "") -> None

Initialize HTTPResponse.

Arguments:

  • status_code: HTTP status code.
  • data: response body bytes.
  • url: request URL.

text

@property
def text() -> str

Response body as string.

content

@property
def content() -> bytes

Response body as bytes.

json

def json() -> Any

Parse response body as JSON.

read

def read() -> bytes

Read response body (for compatibility with file-like usage).

ConnectionError Objects

class ConnectionError(OSError)

HTTP connection error.

request

def request(method: str,
            url: str,
            params: Optional[Dict[str, str]] = None,
            data: Optional[Union[bytes, Dict[str, str]]] = None,
            headers: Optional[Dict[str, str]] = None,
            files: Optional[Dict[str, Any]] = None,
            timeout: float = DEFAULT_TIMEOUT) -> HTTPResponse

Perform an HTTP request using urllib.

Arguments:

  • method: HTTP method (GET, POST, PUT, etc.).
  • url: the URL.
  • params: optional query parameters.
  • data: optional body data (bytes or dict for form-encoded).
  • headers: optional headers dict.
  • files: optional dict of {field: file_obj} for multipart upload.
  • timeout: request timeout in seconds.

Raises:

  • ValueError: if the URL scheme is not http or https.
  • ConnectionError: on connection failure.

Returns:

HTTPResponse.

get

def get(url: str,
        timeout: float = DEFAULT_TIMEOUT,
        **kwargs: Any) -> HTTPResponse

HTTP GET.

Arguments:

  • url: the URL.
  • timeout: request timeout in seconds.
  • kwargs: additional keyword arguments passed to request().

Returns:

HTTPResponse.

post

def post(url: str,
         timeout: float = DEFAULT_TIMEOUT,
         **kwargs: Any) -> HTTPResponse

HTTP POST.

Arguments:

  • url: the URL.
  • timeout: request timeout in seconds.
  • kwargs: additional keyword arguments passed to request().

Returns:

HTTPResponse.

def head(url: str,
         timeout: float = DEFAULT_TIMEOUT,
         **kwargs: Any) -> HTTPResponse

HTTP HEAD.

Arguments:

  • url: the URL.
  • timeout: request timeout in seconds.
  • kwargs: additional keyword arguments passed to request().

Returns:

HTTPResponse.

_ExceptionsShim Objects

class _ExceptionsShim()

Namespace shim exposing the old requests.exceptions.* names.

Provides ConnectionError and RequestException as aliases of our ConnectionError for backwards compatibility with code that catches requests.exceptions.ConnectionError via the old wrapper.

ConnectionError

noqa: A003

download_to_file

def download_to_file(url: str,
                     filepath: str,
                     timeout: float = DEFAULT_TIMEOUT,
                     chunk_size: int = 262144) -> int

Stream the response body to a file in fixed-size chunks.

Avoids buffering large downloads in memory.

Arguments:

  • url: the URL to download.
  • filepath: local file path to write the response body to.
  • timeout: request timeout in seconds.
  • chunk_size: read chunk size in bytes (default 256KB).

Raises:

  • ValueError: if the URL scheme is not http or https.
  • ConnectionError: on connection failure.

Returns:

HTTP status code.