HTTP client (earthlens.base.http)

HttpClient is the shared requests-based transport for the REST-style backends. It owns the chores every backend used to hand-roll — a pooled session, a sensible User-Agent, a per-request timeout, a Retry-After-aware 429/5xx back-off loop, JSON decoding, and a streamed download with a progress bar — so a new backend keeps only the API-shaped parts: endpoint paths, query params, pagination, auth-header values, and response parsing.

Don't hand-roll a session or a retry loop. If you find yourself writing requests.Session(), a while loop that inspects 429 / Retry-After, or an iter_content chunk loop, reach for HttpClient instead. Backends that pre-date it are being migrated onto it.

Consuming it

Construct one client per backend with the default headers it needs, then call the verbs:

from earthlens.base.http import HttpClient

http = HttpClient(headers={"X-API-Key": api_key}, timeout=60.0)

# JSON GET with automatic 429/5xx retry + Retry-After honouring:
payload = http.get_json("https://api.example.org/v1/things", params={"bbox": "1,2,3,4"})

# Streamed download to disk with a tqdm bar (sized from Content-Length):
http.download("https://example.org/big.tif", dest, progress=True)

The default User-Agent is earthlens/{version} — deliberately non-Mozilla, because the DIGITAL.CSIC Anubis anti-bot wall (SPEIbase) blocks browser-like agents. Pass user_agent= for a descriptive contact string (e.g. Overpass / ohsome etiquette).

Testability

Both the transport and the wait are injectable, so the whole client is unit-testable with a fake session and no real delays:

client = HttpClient(session=fake_session, sleep=captured_waits.append)

Pagination and response-envelope parsing stay in each backend — the client owns only how bytes move, never the API shape.

API

earthlens.base.http.HttpClient

Reusable HTTP transport: session, headers, timeout, retry, download.

Wraps a :class:requests.Session, attaching default headers (a non-Mozilla User-Agent and Accept-Encoding: gzip, deflate) to every request and applying a shared timeout. The verbs (:meth:get, :meth:post, :meth:request) route through a Retry-After-aware back-off loop; :meth:get_json decodes the JSON body and :meth:download streams a response to disk. Both the session and the sleep function are injectable so the client is fully unit-testable without a live network.

Default headers set at construction are merged with (and overridden by) any per-request headers=, so a backend expresses its quirks — openaq's X-API-Key, a descriptive osm contact User-Agent — without subclassing.

Attributes:

Name	Type	Description
timeout		Per-request timeout in seconds — a single float, or a (connect, read) pair bounding the two phases separately.
max_retries		Maximum retries on a retryable status before raising.
backoff_factor		Base seconds for exponential back-off when no Retry-After header is present.
status_forcelist		HTTP statuses that trigger a retry.
max_backoff		Ceiling in seconds on any single retry wait.
retry_on_exceptions		Transport exception types that trigger a retry.
raise_for_status		Whether the final response is raise_for_status-ed.
min_interval		Minimum seconds between consecutive requests.

Examples:

• The default agent is non-Mozilla and version-stamped:

  >>> from earthlens.base.http import HttpClient
  >>> HttpClient().default_headers["User-Agent"].startswith("earthlens/")
  True
  

Source code in libs/core/src/earthlens/base/http.py

class HttpClient:
    """Reusable HTTP transport: session, headers, timeout, retry, download.

    Wraps a :class:`requests.Session`, attaching default headers (a
    non-Mozilla `User-Agent` and `Accept-Encoding: gzip, deflate`) to
    every request and applying a shared timeout. The verbs
    (:meth:`get`, :meth:`post`, :meth:`request`) route through a
    `Retry-After`-aware back-off loop; :meth:`get_json` decodes the JSON
    body and :meth:`download` streams a response to disk. Both the
    session and the sleep function are injectable so the client is fully
    unit-testable without a live network.

    Default headers set at construction are merged with (and overridden
    by) any per-request `headers=`, so a backend expresses its quirks —
    openaq's `X-API-Key`, a descriptive osm contact `User-Agent` — without
    subclassing.

    Attributes:
        timeout: Per-request timeout in seconds — a single float, or a
            `(connect, read)` pair bounding the two phases separately.
        max_retries: Maximum retries on a retryable status before raising.
        backoff_factor: Base seconds for exponential back-off when no
            `Retry-After` header is present.
        status_forcelist: HTTP statuses that trigger a retry.
        max_backoff: Ceiling in seconds on any single retry wait.
        retry_on_exceptions: Transport exception types that trigger a retry.
        raise_for_status: Whether the final response is `raise_for_status`-ed.
        min_interval: Minimum seconds between consecutive requests.

    Examples:
        - The default agent is non-Mozilla and version-stamped:
            ```python
            >>> from earthlens.base.http import HttpClient
            >>> HttpClient().default_headers["User-Agent"].startswith("earthlens/")
            True

            ```
    """

    def __init__(
        self,
        *,
        session: requests.Session | None = None,
        user_agent: str | None = None,
        headers: dict[str, str] | None = None,
        timeout: Timeout = DEFAULT_TIMEOUT,
        max_retries: int = DEFAULT_MAX_RETRIES,
        backoff_factor: float = DEFAULT_BACKOFF_FACTOR,
        status_forcelist: tuple[int, ...] = DEFAULT_STATUS_FORCELIST,
        max_backoff: float | None = DEFAULT_MAX_BACKOFF,
        retry_on_exceptions: tuple[type[BaseException], ...] = (),
        retry_predicate: Callable[[requests.Response], bool] | None = None,
        raise_for_status: bool = True,
        min_interval: float = 0.0,
        clock: Callable[[], float] = time.monotonic,
        sleep: Callable[[float], None] = time.sleep,
    ) -> None:
        """Build a client with default headers, timeout, and retry policy.

        Args:
            session: An existing :class:`requests.Session` to reuse.
                Defaults to a fresh session. Injectable so tests can
                supply a fake transport.
            user_agent: The default `User-Agent` header value. Defaults
                to `earthlens/{version}` (non-Mozilla by design; see
                :func:`_default_user_agent`).
            headers: Extra default headers merged onto every request
                (e.g. `{"X-API-Key": ...}`). Override per call with a
                request-level `headers=`.
            timeout: Per-request timeout in seconds — a single float bounds
                both the connect and read phases, or pass a `(connect, read)`
                pair to bound them separately (a short connect budget fails a
                dead host fast without shortening the read budget).
            max_retries: Maximum retries on a retryable status before the
                last response's error is raised.
            backoff_factor: Base seconds for exponential back-off when a
                response carries no `Retry-After` header.
            status_forcelist: HTTP statuses that trigger a retry.
            max_backoff: Ceiling in seconds on any single retry wait, so a
                large `Retry-After` cannot pin the thread indefinitely.
                `None` disables the cap.
            retry_on_exceptions: Exception types that also trigger a retry
                when raised by the transport (e.g.
                `(requests.ConnectionError, requests.Timeout)`). Empty
                (the default) retries on status only, never on a raised
                exception.
            retry_predicate: An optional callback `(response) -> bool`
                that, when it returns `True`, marks a response retryable
                even if its status is not in `status_forcelist` (e.g. a
                `200` whose body signals a rate-limit).
            raise_for_status: Whether to call `raise_for_status` on the
                final response. `False` returns the response unraised so
                the caller can inspect the status itself (e.g. to redact a
                secret-bearing URL from the error, or to branch on a
                `4xx`). Overridable per request.
            min_interval: Minimum seconds between consecutive requests
                (a proactive client-side rate limit). `0.0` (default)
                disables throttling.
            clock: Monotonic clock used for the `min_interval` throttle.
                Injectable so tests drive it deterministically.
            sleep: The sleep function used between retries and for the
                throttle. Defaults to :func:`time.sleep`; injectable so
                tests run without real delays.
        """
        self._session = session if session is not None else new_session()
        self._user_agent = user_agent or _default_user_agent()
        self.timeout = timeout
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor
        self.status_forcelist = tuple(status_forcelist)
        self.max_backoff = max_backoff
        self.retry_on_exceptions = tuple(retry_on_exceptions)
        self._retry_predicate = retry_predicate
        self.raise_for_status = raise_for_status
        self.min_interval = min_interval
        self._clock = clock
        self._sleep = sleep
        self._last_request: float | None = None
        self._throttle_lock = threading.Lock()
        self._default_headers: dict[str, str] = {
            "User-Agent": self._user_agent,
            "Accept-Encoding": "gzip, deflate",
        }
        if headers:
            self._default_headers.update(headers)

    @property
    def default_headers(self) -> dict[str, str]:
        """Return a copy of the headers merged onto every request."""
        return dict(self._default_headers)

    @property
    def session(self) -> requests.Session:
        """Return the underlying :class:`requests.Session`."""
        return self._session

    def _merge_headers(self, headers: dict[str, str] | None) -> dict[str, str]:
        """Merge per-request `headers` over the client's defaults."""
        merged = dict(self._default_headers)
        if headers:
            merged.update(headers)
        return merged

    def request(
        self,
        method: str,
        url: str,
        *,
        headers: dict[str, str] | None = None,
        timeout: Timeout | None = None,
        raise_for_status: bool | None = None,
        **kwargs: Any,
    ) -> requests.Response:
        """Send one request with the default headers, timeout, and retry.

        Args:
            method: HTTP verb (`"GET"`, `"POST"`, ...).
            url: Absolute request URL.
            headers: Per-request headers merged over the client defaults.
            timeout: Per-request timeout override (seconds), as a single
                float or a `(connect, read)` pair. Defaults to the client's
                `timeout`.
            raise_for_status: Per-request override of the client's
                `raise_for_status` policy. `None` (default) uses the
                client setting.
            **kwargs: Extra keyword arguments forwarded to `requests`
                (`params`, `data`, `json`, `stream`, ...).

        Returns:
            requests.Response: The response (after `raise_for_status`
                unless it is disabled).

        Raises:
            requests.HTTPError: On a non-retryable error status, or after
                the retryable status is exhausted (when `raise_for_status`
                is on).
        """
        merged = self._merge_headers(headers)
        effective_timeout = self.timeout if timeout is None else timeout
        return self._request_with_retry(
            method,
            url,
            headers=merged,
            timeout=effective_timeout,
            raise_for_status=raise_for_status,
            **kwargs,
        )

    def get(self, url: str, **kwargs: Any) -> requests.Response:
        """Send a `GET` request. See :meth:`request` for arguments."""
        return self.request("GET", url, **kwargs)

    def post(self, url: str, **kwargs: Any) -> requests.Response:
        """Send a `POST` request. See :meth:`request` for arguments."""
        return self.request("POST", url, **kwargs)

    def get_json(self, url: str, **kwargs: Any) -> Any:
        """Send a `GET` request and decode the JSON response body.

        Convenience over :meth:`get` for the REST endpoints that return
        JSON envelopes.

        Args:
            url: Absolute request URL.
            **kwargs: Keyword arguments forwarded to :meth:`get`
                (`params`, `headers`, `timeout`, ...).

        Returns:
            The parsed JSON body (typically a `dict` or `list`).

        Raises:
            requests.HTTPError: On a non-retryable error status, or after
                the retryable status is exhausted.
        """
        return self.get(url, **kwargs).json()

    def stream(self, url: str, **kwargs: Any) -> requests.Response:
        """Send a streaming `GET` (`stream=True`), retry-wrapped.

        Returns the open response without consuming its body, so the
        caller can iterate `iter_content`. Retries follow the same
        `Retry-After`/back-off policy as the other verbs; the retry
        decision reads only the status line, never the body.

        Args:
            url: Absolute request URL.
            **kwargs: Keyword arguments forwarded to :meth:`get`.

        Returns:
            requests.Response: The open streaming response.
        """
        return self.get(url, stream=True, **kwargs)

    def _stream_to_file(
        self,
        response: requests.Response,
        dest: Path,
        *,
        chunk: int,
        progress: bool,
        desc: str,
    ) -> None:
        """Write a streaming response's body to `dest` with a `tqdm` bar.

        Args:
            response: The open streaming response.
            dest: The file to write (typically a temp `.part` path).
            chunk: Streaming block size in bytes.
            progress: Whether to show the progress bar.
            desc: The bar label (the final file name).
        """
        total = _progress_total(response.headers)
        bar = tqdm(
            total=total,
            unit="B",
            unit_scale=True,
            unit_divisor=1024,
            disable=not progress,
            desc=desc,
        )
        try:
            with open(dest, "wb") as handle:
                for block in response.iter_content(chunk_size=chunk):
                    if block:
                        handle.write(block)
                        bar.update(len(block))
        finally:
            bar.close()

    def download(
        self,
        url: str,
        dest: str | Path,
        *,
        chunk: int = DEFAULT_CHUNK_SIZE,
        progress: bool = True,
        atomic: bool = True,
        expect_magic: bytes | tuple[bytes, ...] | None = None,
        headers: dict[str, str] | None = None,
        timeout: Timeout | None = None,
        **kwargs: Any,
    ) -> Path:
        """Stream `url` to `dest` atomically, optionally showing a `tqdm` bar.

        Absorbs the chunk-loop the streamed-download backends each
        re-implement: streams with `stream=True`, sizes a progress bar
        from `Content-Length` when present, and writes 1 MiB blocks. When
        `atomic` (the default) it writes to a sibling `<dest>.part` and
        renames on success, and it removes the temp on any failure — so a
        crashed or interrupted download never leaves a truncated `dest`.
        The whole download is retry-wrapped: a status in `status_forcelist`
        or an exception in `retry_on_exceptions` retries the attempt (after
        cleaning the temp), honouring the `Retry-After`/back-off policy.

        Args:
            url: Absolute request URL.
            dest: Output file path. Parent directories are created.
            chunk: Streaming block size in bytes (default 1 MiB).
            progress: Show a `tqdm` progress bar. `False` (or a
                non-interactive / test context) suppresses it.
            atomic: Write to `<dest>.part` then rename on success, cleaning up
                the temp on failure — so a crashed download never leaves a
                truncated `dest` and never removes an existing one. Keep this on
                (the default) whenever an existing `dest` must survive a failed
                attempt. `False` streams straight into `dest`, which is opened
                `"wb"` and therefore **truncated up front**: a mid-stream failure
                leaves `dest` short, and any previous contents are gone. The
                failure path does not additionally delete it, but that is damage
                limitation, not a guarantee — `atomic=False` is only appropriate
                when `dest` is known to be disposable.
            expect_magic: One or more byte prefixes the body must start with
                (e.g. `b"CDF"` / `b"\\x89HDF"` for NetCDF). A body that starts
                with none of them raises `ValueError` and the partial write is
                discarded, so an HTML error page served with a 200 status never
                lands as a `.nc`. `None` (the default) skips the check.
            headers: Per-request headers merged over the client defaults.
            timeout: Per-request timeout override (seconds), as a single
                float or a `(connect, read)` pair — the latter fails a dead
                host on the short connect budget without shortening the long
                read budget a large download needs.
            **kwargs: Extra keyword arguments forwarded to `requests`.

        Returns:
            Path: The `dest` path the bytes were written to.

        Raises:
            ValueError: When `expect_magic` is given and the body does not
                start with any of the supplied prefixes.
            requests.HTTPError: On an error status — `download` always
                calls `raise_for_status` (the client's `raise_for_status`
                flag governs the verb methods, not `download`; a file
                fetch never keeps an error body). Note the resulting
                `HTTPError` is itself subject to `retry_on_exceptions`:
                a client whose `retry_on_exceptions` includes a supertype
                of `requests.HTTPError` (e.g. `requests.RequestException`,
                as ghsl/glaciers pass) will **retry** an error status
                before raising it, mirroring their old download loops.
                Also the last transport exception after the retry budget
                is exhausted.
        """
        dest = Path(dest)
        dest.parent.mkdir(parents=True, exist_ok=True)
        # `expect_magic` promises a rejected body is discarded, which is only
        # possible if `dest` has not been written yet — so a magic check forces
        # staging even when the caller passed `atomic=False`. Otherwise the
        # validation would run on a file that had already replaced a good one.
        staged = atomic or expect_magic is not None
        tmp = dest.with_name(dest.name + ".part") if staged else dest

        def discard_partial() -> None:
            """Remove the partial write, but never a caller-owned `dest`.

            When the write was staged (`atomic`, or a magic check forcing it)
            the partial lives at a private `<dest>.part`, so removing it is
            always safe. Otherwise the stream wrote straight to `dest`, which
            `_stream_to_file` has already truncated by opening it `"wb"` —
            deleting it as well would only turn a truncated file into a missing
            one, and would destroy a file this call never owned.
            """
            if staged:
                tmp.unlink(missing_ok=True)

        merged = self._merge_headers(headers)
        effective_timeout = self.timeout if timeout is None else timeout
        attempt = 0
        while True:
            self._throttle()
            try:
                response = self._send(
                    "GET",
                    url,
                    stream=True,
                    headers=merged,
                    timeout=effective_timeout,
                    **kwargs,
                )
                try:
                    retryable = response.status_code in self.status_forcelist or (
                        self._retry_predicate is not None
                        and self._retry_predicate(response)
                    )
                    if retryable and attempt < self.max_retries:
                        retry_after = _parse_retry_after(
                            response.headers.get("Retry-After")
                        )
                        wait = self._backoff_wait(retry_after, attempt)
                        logger.warning(
                            f"HTTP {response.status_code} on {redact_url(url)}; retry "
                            f"{attempt + 1}/{self.max_retries} after {wait:.1f}s"
                        )
                        self._sleep(wait)
                        attempt += 1
                        continue
                    response.raise_for_status()
                    self._stream_to_file(
                        response, tmp, chunk=chunk, progress=progress, desc=dest.name
                    )
                    if expect_magic is not None:
                        _check_magic(tmp, expect_magic, url)
                finally:
                    response.close()
            except self.retry_on_exceptions as exc:
                discard_partial()
                if attempt >= self.max_retries:
                    raise
                wait = self._backoff_wait(None, attempt)
                logger.warning(
                    f"{type(exc).__name__} on {redact_url(url)}; retry "
                    f"{attempt + 1}/{self.max_retries} after {wait:.1f}s"
                )
                self._sleep(wait)
                attempt += 1
                continue
            except BaseException:
                discard_partial()
                raise
            if staged:
                # Guard the rename too, so the "removes the temp on any
                # failure" promise holds if the final replace fails.
                try:
                    tmp.replace(dest)
                except BaseException:
                    discard_partial()
                    raise
            return dest

    def _throttle(self) -> None:
        """Sleep so consecutive requests are >= `min_interval` apart.

        A no-op when `min_interval` is `0`. Uses the injected monotonic
        `clock`, records the send time, and sleeps via the injected
        `sleep` so tests drive the rate limit deterministically.

        Held under a lock for the whole read-sleep-write sequence, so
        `min_interval` bounds the *aggregate* request rate rather than the
        rate per thread. Read-then-write without it is a race: every thread
        sees the same `_last_request`, each concludes the interval has
        elapsed, and they all fire together — the burst the limit exists to
        prevent.

        No backend shares one client across threads *today*: the three that
        thread (worldpop, ghsl, hdx, all `prefer="threads"`) build a fresh
        client inside each worker, and `_run_items` is a sequential loop. The
        lock is cheap and makes a shared client safe whenever one appears,
        rather than leaving a latent race for that change to trip over.
        """
        if self.min_interval <= 0:
            return
        with self._throttle_lock:
            if self._last_request is not None:
                remaining = self.min_interval - (self._clock() - self._last_request)
                if remaining > 0:
                    self._sleep(remaining)
            self._last_request = self._clock()

    def _backoff_wait(self, retry_after: float | None, attempt: int) -> float:
        """Compute one retry wait: `Retry-After` else exponential back-off.

        Applies the `max_backoff` ceiling and a non-negative floor.

        Args:
            retry_after: Parsed `Retry-After` seconds, or `None`.
            attempt: The zero-based attempt index.

        Returns:
            The clamped wait in seconds.
        """
        wait = (
            retry_after
            if retry_after is not None
            else self.backoff_factor * (2**attempt)
        )
        if self.max_backoff is not None:
            wait = min(wait, self.max_backoff)
        return max(0.0, wait)

    def _request_with_retry(
        self,
        method: str,
        url: str,
        *,
        raise_for_status: bool | None = None,
        **kwargs: Any,
    ) -> requests.Response:
        """Send one request, retrying statuses/exceptions with back-off.

        Retries while the attempt budget holds and either the response
        status is in `status_forcelist`, the `retry_predicate` marks the
        response retryable, or the transport raised one of
        `retry_on_exceptions`. Waits `Retry-After` seconds when present
        and numeric, otherwise `backoff_factor * 2**attempt` (capped by
        `max_backoff`). Honours the `min_interval` throttle before every
        send. Once non-retryable or exhausted, the response is
        `raise_for_status`-ed unless that is disabled, then returned.

        Args:
            method: HTTP verb.
            url: Absolute request URL.
            raise_for_status: Per-request override; `None` uses the
                client policy.
            **kwargs: Keyword arguments forwarded to the session.

        Returns:
            requests.Response: The final response.

        Raises:
            requests.HTTPError: On a non-retryable error status when
                `raise_for_status` is on.
            BaseException: The last transport exception, re-raised after
                the `retry_on_exceptions` budget is exhausted.
        """
        effective_raise = (
            self.raise_for_status if raise_for_status is None else raise_for_status
        )
        attempt = 0
        while True:
            self._throttle()
            try:
                response = self._send(method, url, **kwargs)
            except self.retry_on_exceptions as exc:
                if attempt >= self.max_retries:
                    raise
                wait = self._backoff_wait(None, attempt)
                logger.warning(
                    f"{type(exc).__name__} on {redact_url(url)}; retry "
                    f"{attempt + 1}/{self.max_retries} after {wait:.1f}s"
                )
                self._sleep(wait)
                attempt += 1
                continue
            retryable = response.status_code in self.status_forcelist or (
                self._retry_predicate is not None and self._retry_predicate(response)
            )
            if retryable and attempt < self.max_retries:
                retry_after = _parse_retry_after(response.headers.get("Retry-After"))
                wait = self._backoff_wait(retry_after, attempt)
                logger.warning(
                    f"HTTP {response.status_code} on {redact_url(url)}; retry "
                    f"{attempt + 1}/{self.max_retries} after {wait:.1f}s"
                )
                # Release the (possibly streamed) connection before retrying;
                # a stream=True body is otherwise never consumed and its socket
                # leaks out of the pool.
                response.close()
                self._sleep(wait)
                attempt += 1
                continue
            if effective_raise:
                try:
                    response.raise_for_status()
                except requests.HTTPError:
                    # Close the final errored response too — for a streamed
                    # request the caller never receives it, so nothing else would.
                    response.close()
                    raise
            return response

    def _send(self, method: str, url: str, **kwargs: Any) -> requests.Response:
        """Dispatch to the session's verb method, falling back to `request`.

        Routing `GET` to `session.get` (rather than a generic
        `session.request`) keeps drop-in fake transports that implement
        only `get()` working — the shape the migrated backends' tests use.

        Args:
            method: HTTP verb.
            url: Absolute request URL.
            **kwargs: Keyword arguments forwarded to the session call.

        Returns:
            requests.Response: The raw response (no status check).
        """
        verb = getattr(self._session, method.lower(), None)
        if callable(verb):
            return cast("requests.Response", verb(url, **kwargs))
        return self._session.request(method, url, **kwargs)

default_headers property

Return a copy of the headers merged onto every request.

session property

Return the underlying :class:requests.Session.

__init__(*, session=None, user_agent=None, headers=None, timeout=DEFAULT_TIMEOUT, max_retries=DEFAULT_MAX_RETRIES, backoff_factor=DEFAULT_BACKOFF_FACTOR, status_forcelist=DEFAULT_STATUS_FORCELIST, max_backoff=DEFAULT_MAX_BACKOFF, retry_on_exceptions=(), retry_predicate=None, raise_for_status=True, min_interval=0.0, clock=time.monotonic, sleep=time.sleep)

Build a client with default headers, timeout, and retry policy.

Parameters:

Name	Type	Description	Default
session	Session | None	An existing :class:requests.Session to reuse. Defaults to a fresh session. Injectable so tests can supply a fake transport.	None
user_agent	str | None	The default User-Agent header value. Defaults to earthlens/{version} (non-Mozilla by design; see :func:_default_user_agent).	None
headers	dict[str, str] | None	Extra default headers merged onto every request (e.g. {"X-API-Key": ...}). Override per call with a request-level headers=.	None
timeout	Timeout	Per-request timeout in seconds — a single float bounds both the connect and read phases, or pass a (connect, read) pair to bound them separately (a short connect budget fails a dead host fast without shortening the read budget).	DEFAULT_TIMEOUT
max_retries	int	Maximum retries on a retryable status before the last response's error is raised.	DEFAULT_MAX_RETRIES
backoff_factor	float	Base seconds for exponential back-off when a response carries no Retry-After header.	DEFAULT_BACKOFF_FACTOR
status_forcelist	tuple[int, ...]	HTTP statuses that trigger a retry.	DEFAULT_STATUS_FORCELIST
max_backoff	float | None	Ceiling in seconds on any single retry wait, so a large Retry-After cannot pin the thread indefinitely. None disables the cap.	DEFAULT_MAX_BACKOFF
retry_on_exceptions	tuple[type[BaseException], ...]	Exception types that also trigger a retry when raised by the transport (e.g. (requests.ConnectionError, requests.Timeout)). Empty (the default) retries on status only, never on a raised exception.	()
retry_predicate	Callable[[Response], bool] | None	An optional callback (response) -> bool that, when it returns True, marks a response retryable even if its status is not in status_forcelist (e.g. a 200 whose body signals a rate-limit).	None
raise_for_status	bool	Whether to call raise_for_status on the final response. False returns the response unraised so the caller can inspect the status itself (e.g. to redact a secret-bearing URL from the error, or to branch on a 4xx). Overridable per request.	True
min_interval	float	Minimum seconds between consecutive requests (a proactive client-side rate limit). 0.0 (default) disables throttling.	0.0
clock	Callable[[], float]	Monotonic clock used for the min_interval throttle. Injectable so tests drive it deterministically.	monotonic
sleep	Callable[[float], None]	The sleep function used between retries and for the throttle. Defaults to :func:time.sleep; injectable so tests run without real delays.	sleep

Source code in libs/core/src/earthlens/base/http.py

def __init__(
    self,
    *,
    session: requests.Session | None = None,
    user_agent: str | None = None,
    headers: dict[str, str] | None = None,
    timeout: Timeout = DEFAULT_TIMEOUT,
    max_retries: int = DEFAULT_MAX_RETRIES,
    backoff_factor: float = DEFAULT_BACKOFF_FACTOR,
    status_forcelist: tuple[int, ...] = DEFAULT_STATUS_FORCELIST,
    max_backoff: float | None = DEFAULT_MAX_BACKOFF,
    retry_on_exceptions: tuple[type[BaseException], ...] = (),
    retry_predicate: Callable[[requests.Response], bool] | None = None,
    raise_for_status: bool = True,
    min_interval: float = 0.0,
    clock: Callable[[], float] = time.monotonic,
    sleep: Callable[[float], None] = time.sleep,
) -> None:
    """Build a client with default headers, timeout, and retry policy.

    Args:
        session: An existing :class:`requests.Session` to reuse.
            Defaults to a fresh session. Injectable so tests can
            supply a fake transport.
        user_agent: The default `User-Agent` header value. Defaults
            to `earthlens/{version}` (non-Mozilla by design; see
            :func:`_default_user_agent`).
        headers: Extra default headers merged onto every request
            (e.g. `{"X-API-Key": ...}`). Override per call with a
            request-level `headers=`.
        timeout: Per-request timeout in seconds — a single float bounds
            both the connect and read phases, or pass a `(connect, read)`
            pair to bound them separately (a short connect budget fails a
            dead host fast without shortening the read budget).
        max_retries: Maximum retries on a retryable status before the
            last response's error is raised.
        backoff_factor: Base seconds for exponential back-off when a
            response carries no `Retry-After` header.
        status_forcelist: HTTP statuses that trigger a retry.
        max_backoff: Ceiling in seconds on any single retry wait, so a
            large `Retry-After` cannot pin the thread indefinitely.
            `None` disables the cap.
        retry_on_exceptions: Exception types that also trigger a retry
            when raised by the transport (e.g.
            `(requests.ConnectionError, requests.Timeout)`). Empty
            (the default) retries on status only, never on a raised
            exception.
        retry_predicate: An optional callback `(response) -> bool`
            that, when it returns `True`, marks a response retryable
            even if its status is not in `status_forcelist` (e.g. a
            `200` whose body signals a rate-limit).
        raise_for_status: Whether to call `raise_for_status` on the
            final response. `False` returns the response unraised so
            the caller can inspect the status itself (e.g. to redact a
            secret-bearing URL from the error, or to branch on a
            `4xx`). Overridable per request.
        min_interval: Minimum seconds between consecutive requests
            (a proactive client-side rate limit). `0.0` (default)
            disables throttling.
        clock: Monotonic clock used for the `min_interval` throttle.
            Injectable so tests drive it deterministically.
        sleep: The sleep function used between retries and for the
            throttle. Defaults to :func:`time.sleep`; injectable so
            tests run without real delays.
    """
    self._session = session if session is not None else new_session()
    self._user_agent = user_agent or _default_user_agent()
    self.timeout = timeout
    self.max_retries = max_retries
    self.backoff_factor = backoff_factor
    self.status_forcelist = tuple(status_forcelist)
    self.max_backoff = max_backoff
    self.retry_on_exceptions = tuple(retry_on_exceptions)
    self._retry_predicate = retry_predicate
    self.raise_for_status = raise_for_status
    self.min_interval = min_interval
    self._clock = clock
    self._sleep = sleep
    self._last_request: float | None = None
    self._throttle_lock = threading.Lock()
    self._default_headers: dict[str, str] = {
        "User-Agent": self._user_agent,
        "Accept-Encoding": "gzip, deflate",
    }
    if headers:
        self._default_headers.update(headers)

download(url, dest, *, chunk=DEFAULT_CHUNK_SIZE, progress=True, atomic=True, expect_magic=None, headers=None, timeout=None, **kwargs)

Stream url to dest atomically, optionally showing a tqdm bar.

Absorbs the chunk-loop the streamed-download backends each re-implement: streams with stream=True, sizes a progress bar from Content-Length when present, and writes 1 MiB blocks. When atomic (the default) it writes to a sibling <dest>.part and renames on success, and it removes the temp on any failure — so a crashed or interrupted download never leaves a truncated dest. The whole download is retry-wrapped: a status in status_forcelist or an exception in retry_on_exceptions retries the attempt (after cleaning the temp), honouring the Retry-After/back-off policy.

Parameters:

Name	Type	Description	Default
url	str	Absolute request URL.	required
dest	str | Path	Output file path. Parent directories are created.	required
chunk	int	Streaming block size in bytes (default 1 MiB).	DEFAULT_CHUNK_SIZE
progress	bool	Show a tqdm progress bar. False (or a non-interactive / test context) suppresses it.	True
atomic	bool	Write to <dest>.part then rename on success, cleaning up the temp on failure — so a crashed download never leaves a truncated dest and never removes an existing one. Keep this on (the default) whenever an existing dest must survive a failed attempt. False streams straight into dest, which is opened "wb" and therefore truncated up front: a mid-stream failure leaves dest short, and any previous contents are gone. The failure path does not additionally delete it, but that is damage limitation, not a guarantee — atomic=False is only appropriate when dest is known to be disposable.	True
expect_magic	bytes | tuple[bytes, ...] | None	One or more byte prefixes the body must start with (e.g. b"CDF" / b"\x89HDF" for NetCDF). A body that starts with none of them raises ValueError and the partial write is discarded, so an HTML error page served with a 200 status never lands as a .nc. None (the default) skips the check.	None
headers	dict[str, str] | None	Per-request headers merged over the client defaults.	None
timeout	Timeout | None	Per-request timeout override (seconds), as a single float or a (connect, read) pair — the latter fails a dead host on the short connect budget without shortening the long read budget a large download needs.	None
**kwargs	Any	Extra keyword arguments forwarded to requests.	{}

Returns:

Name	Type	Description
Path	Path	The dest path the bytes were written to.

Raises:

Type	Description
ValueError	When expect_magic is given and the body does not start with any of the supplied prefixes.
HTTPError	On an error status — download always calls raise_for_status (the client's raise_for_status flag governs the verb methods, not download; a file fetch never keeps an error body). Note the resulting HTTPError is itself subject to retry_on_exceptions: a client whose retry_on_exceptions includes a supertype of requests.HTTPError (e.g. requests.RequestException, as ghsl/glaciers pass) will retry an error status before raising it, mirroring their old download loops. Also the last transport exception after the retry budget is exhausted.

Source code in libs/core/src/earthlens/base/http.py

def download(
    self,
    url: str,
    dest: str | Path,
    *,
    chunk: int = DEFAULT_CHUNK_SIZE,
    progress: bool = True,
    atomic: bool = True,
    expect_magic: bytes | tuple[bytes, ...] | None = None,
    headers: dict[str, str] | None = None,
    timeout: Timeout | None = None,
    **kwargs: Any,
) -> Path:
    """Stream `url` to `dest` atomically, optionally showing a `tqdm` bar.

    Absorbs the chunk-loop the streamed-download backends each
    re-implement: streams with `stream=True`, sizes a progress bar
    from `Content-Length` when present, and writes 1 MiB blocks. When
    `atomic` (the default) it writes to a sibling `<dest>.part` and
    renames on success, and it removes the temp on any failure — so a
    crashed or interrupted download never leaves a truncated `dest`.
    The whole download is retry-wrapped: a status in `status_forcelist`
    or an exception in `retry_on_exceptions` retries the attempt (after
    cleaning the temp), honouring the `Retry-After`/back-off policy.

    Args:
        url: Absolute request URL.
        dest: Output file path. Parent directories are created.
        chunk: Streaming block size in bytes (default 1 MiB).
        progress: Show a `tqdm` progress bar. `False` (or a
            non-interactive / test context) suppresses it.
        atomic: Write to `<dest>.part` then rename on success, cleaning up
            the temp on failure — so a crashed download never leaves a
            truncated `dest` and never removes an existing one. Keep this on
            (the default) whenever an existing `dest` must survive a failed
            attempt. `False` streams straight into `dest`, which is opened
            `"wb"` and therefore **truncated up front**: a mid-stream failure
            leaves `dest` short, and any previous contents are gone. The
            failure path does not additionally delete it, but that is damage
            limitation, not a guarantee — `atomic=False` is only appropriate
            when `dest` is known to be disposable.
        expect_magic: One or more byte prefixes the body must start with
            (e.g. `b"CDF"` / `b"\\x89HDF"` for NetCDF). A body that starts
            with none of them raises `ValueError` and the partial write is
            discarded, so an HTML error page served with a 200 status never
            lands as a `.nc`. `None` (the default) skips the check.
        headers: Per-request headers merged over the client defaults.
        timeout: Per-request timeout override (seconds), as a single
            float or a `(connect, read)` pair — the latter fails a dead
            host on the short connect budget without shortening the long
            read budget a large download needs.
        **kwargs: Extra keyword arguments forwarded to `requests`.

    Returns:
        Path: The `dest` path the bytes were written to.

    Raises:
        ValueError: When `expect_magic` is given and the body does not
            start with any of the supplied prefixes.
        requests.HTTPError: On an error status — `download` always
            calls `raise_for_status` (the client's `raise_for_status`
            flag governs the verb methods, not `download`; a file
            fetch never keeps an error body). Note the resulting
            `HTTPError` is itself subject to `retry_on_exceptions`:
            a client whose `retry_on_exceptions` includes a supertype
            of `requests.HTTPError` (e.g. `requests.RequestException`,
            as ghsl/glaciers pass) will **retry** an error status
            before raising it, mirroring their old download loops.
            Also the last transport exception after the retry budget
            is exhausted.
    """
    dest = Path(dest)
    dest.parent.mkdir(parents=True, exist_ok=True)
    # `expect_magic` promises a rejected body is discarded, which is only
    # possible if `dest` has not been written yet — so a magic check forces
    # staging even when the caller passed `atomic=False`. Otherwise the
    # validation would run on a file that had already replaced a good one.
    staged = atomic or expect_magic is not None
    tmp = dest.with_name(dest.name + ".part") if staged else dest

    def discard_partial() -> None:
        """Remove the partial write, but never a caller-owned `dest`.

        When the write was staged (`atomic`, or a magic check forcing it)
        the partial lives at a private `<dest>.part`, so removing it is
        always safe. Otherwise the stream wrote straight to `dest`, which
        `_stream_to_file` has already truncated by opening it `"wb"` —
        deleting it as well would only turn a truncated file into a missing
        one, and would destroy a file this call never owned.
        """
        if staged:
            tmp.unlink(missing_ok=True)

    merged = self._merge_headers(headers)
    effective_timeout = self.timeout if timeout is None else timeout
    attempt = 0
    while True:
        self._throttle()
        try:
            response = self._send(
                "GET",
                url,
                stream=True,
                headers=merged,
                timeout=effective_timeout,
                **kwargs,
            )
            try:
                retryable = response.status_code in self.status_forcelist or (
                    self._retry_predicate is not None
                    and self._retry_predicate(response)
                )
                if retryable and attempt < self.max_retries:
                    retry_after = _parse_retry_after(
                        response.headers.get("Retry-After")
                    )
                    wait = self._backoff_wait(retry_after, attempt)
                    logger.warning(
                        f"HTTP {response.status_code} on {redact_url(url)}; retry "
                        f"{attempt + 1}/{self.max_retries} after {wait:.1f}s"
                    )
                    self._sleep(wait)
                    attempt += 1
                    continue
                response.raise_for_status()
                self._stream_to_file(
                    response, tmp, chunk=chunk, progress=progress, desc=dest.name
                )
                if expect_magic is not None:
                    _check_magic(tmp, expect_magic, url)
            finally:
                response.close()
        except self.retry_on_exceptions as exc:
            discard_partial()
            if attempt >= self.max_retries:
                raise
            wait = self._backoff_wait(None, attempt)
            logger.warning(
                f"{type(exc).__name__} on {redact_url(url)}; retry "
                f"{attempt + 1}/{self.max_retries} after {wait:.1f}s"
            )
            self._sleep(wait)
            attempt += 1
            continue
        except BaseException:
            discard_partial()
            raise
        if staged:
            # Guard the rename too, so the "removes the temp on any
            # failure" promise holds if the final replace fails.
            try:
                tmp.replace(dest)
            except BaseException:
                discard_partial()
                raise
        return dest

get(url, **kwargs)

Send a GET request. See :meth:request for arguments.

Source code in libs/core/src/earthlens/base/http.py

def get(self, url: str, **kwargs: Any) -> requests.Response:
    """Send a `GET` request. See :meth:`request` for arguments."""
    return self.request("GET", url, **kwargs)

get_json(url, **kwargs)

Send a GET request and decode the JSON response body.

Convenience over :meth:get for the REST endpoints that return JSON envelopes.

Parameters:

Name	Type	Description	Default
url	str	Absolute request URL.	required
**kwargs	Any	Keyword arguments forwarded to :meth:get (params, headers, timeout, ...).	{}

Returns:

Type	Description
Any	The parsed JSON body (typically a dict or list).

Raises:

Type	Description
HTTPError	On a non-retryable error status, or after the retryable status is exhausted.

Source code in libs/core/src/earthlens/base/http.py

def get_json(self, url: str, **kwargs: Any) -> Any:
    """Send a `GET` request and decode the JSON response body.

    Convenience over :meth:`get` for the REST endpoints that return
    JSON envelopes.

    Args:
        url: Absolute request URL.
        **kwargs: Keyword arguments forwarded to :meth:`get`
            (`params`, `headers`, `timeout`, ...).

    Returns:
        The parsed JSON body (typically a `dict` or `list`).

    Raises:
        requests.HTTPError: On a non-retryable error status, or after
            the retryable status is exhausted.
    """
    return self.get(url, **kwargs).json()

post(url, **kwargs)

Send a POST request. See :meth:request for arguments.

Source code in libs/core/src/earthlens/base/http.py

def post(self, url: str, **kwargs: Any) -> requests.Response:
    """Send a `POST` request. See :meth:`request` for arguments."""
    return self.request("POST", url, **kwargs)

request(method, url, *, headers=None, timeout=None, raise_for_status=None, **kwargs)

Send one request with the default headers, timeout, and retry.

Parameters:

Name	Type	Description	Default
method	str	HTTP verb ("GET", "POST", ...).	required
url	str	Absolute request URL.	required
headers	dict[str, str] | None	Per-request headers merged over the client defaults.	None
timeout	Timeout | None	Per-request timeout override (seconds), as a single float or a (connect, read) pair. Defaults to the client's timeout.	None
raise_for_status	bool | None	Per-request override of the client's raise_for_status policy. None (default) uses the client setting.	None
**kwargs	Any	Extra keyword arguments forwarded to requests (params, data, json, stream, ...).	{}

Returns:

Type	Description
Response	requests.Response: The response (after raise_for_status unless it is disabled).

Raises:

Type	Description
HTTPError	On a non-retryable error status, or after the retryable status is exhausted (when raise_for_status is on).

Source code in libs/core/src/earthlens/base/http.py

def request(
    self,
    method: str,
    url: str,
    *,
    headers: dict[str, str] | None = None,
    timeout: Timeout | None = None,
    raise_for_status: bool | None = None,
    **kwargs: Any,
) -> requests.Response:
    """Send one request with the default headers, timeout, and retry.

    Args:
        method: HTTP verb (`"GET"`, `"POST"`, ...).
        url: Absolute request URL.
        headers: Per-request headers merged over the client defaults.
        timeout: Per-request timeout override (seconds), as a single
            float or a `(connect, read)` pair. Defaults to the client's
            `timeout`.
        raise_for_status: Per-request override of the client's
            `raise_for_status` policy. `None` (default) uses the
            client setting.
        **kwargs: Extra keyword arguments forwarded to `requests`
            (`params`, `data`, `json`, `stream`, ...).

    Returns:
        requests.Response: The response (after `raise_for_status`
            unless it is disabled).

    Raises:
        requests.HTTPError: On a non-retryable error status, or after
            the retryable status is exhausted (when `raise_for_status`
            is on).
    """
    merged = self._merge_headers(headers)
    effective_timeout = self.timeout if timeout is None else timeout
    return self._request_with_retry(
        method,
        url,
        headers=merged,
        timeout=effective_timeout,
        raise_for_status=raise_for_status,
        **kwargs,
    )

stream(url, **kwargs)

Send a streaming GET (stream=True), retry-wrapped.

Returns the open response without consuming its body, so the caller can iterate iter_content. Retries follow the same Retry-After/back-off policy as the other verbs; the retry decision reads only the status line, never the body.

Parameters:

Name	Type	Description	Default
url	str	Absolute request URL.	required
**kwargs	Any	Keyword arguments forwarded to :meth:get.	{}

Returns:

Type	Description
Response	requests.Response: The open streaming response.

Source code in libs/core/src/earthlens/base/http.py

def stream(self, url: str, **kwargs: Any) -> requests.Response:
    """Send a streaming `GET` (`stream=True`), retry-wrapped.

    Returns the open response without consuming its body, so the
    caller can iterate `iter_content`. Retries follow the same
    `Retry-After`/back-off policy as the other verbs; the retry
    decision reads only the status line, never the body.

    Args:
        url: Absolute request URL.
        **kwargs: Keyword arguments forwarded to :meth:`get`.

    Returns:
        requests.Response: The open streaming response.
    """
    return self.get(url, stream=True, **kwargs)