Skip to content

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.

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 src/earthlens/base/http.py
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
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.
        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: float = 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.
            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 requests.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._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: float | 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). 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,
        headers: dict[str, str] | None = None,
        timeout: float | 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. `False` writes straight to `dest`.
            headers: Per-request headers merged over the client defaults.
            timeout: Per-request timeout override (seconds).
            **kwargs: Extra keyword arguments forwarded to `requests`.

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

        Raises:
            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)
        tmp = dest.with_name(dest.name + ".part") if atomic else dest
        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
                    )
                finally:
                    response.close()
            except self.retry_on_exceptions as exc:
                tmp.unlink(missing_ok=True)
                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:
                tmp.unlink(missing_ok=True)
                raise
            if atomic:
                # 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:
                    tmp.unlink(missing_ok=True)
                    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.
        """
        if self.min_interval <= 0:
            return
        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 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 float

Per-request timeout in seconds.

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 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: float = 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.
        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 requests.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._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, 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. False writes straight to dest.

True
headers dict[str, str] | None

Per-request headers merged over the client defaults.

None
timeout float | None

Per-request timeout override (seconds).

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
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 src/earthlens/base/http.py
def download(
    self,
    url: str,
    dest: str | Path,
    *,
    chunk: int = DEFAULT_CHUNK_SIZE,
    progress: bool = True,
    atomic: bool = True,
    headers: dict[str, str] | None = None,
    timeout: float | 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. `False` writes straight to `dest`.
        headers: Per-request headers merged over the client defaults.
        timeout: Per-request timeout override (seconds).
        **kwargs: Extra keyword arguments forwarded to `requests`.

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

    Raises:
        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)
    tmp = dest.with_name(dest.name + ".part") if atomic else dest
    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
                )
            finally:
                response.close()
        except self.retry_on_exceptions as exc:
            tmp.unlink(missing_ok=True)
            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:
            tmp.unlink(missing_ok=True)
            raise
        if atomic:
            # 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:
                tmp.unlink(missing_ok=True)
                raise
        return dest

get(url, **kwargs) #

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

Source code in 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 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 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 float | None

Per-request timeout override (seconds). 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 src/earthlens/base/http.py
def request(
    self,
    method: str,
    url: str,
    *,
    headers: dict[str, str] | None = None,
    timeout: float | 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). 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 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)