repomatic.http module

Shared HTTP fetch for the API clients.

The single implementation of the GET loop used by the PyPI (repomatic.pypi), npm (repomatic.npm), and GitHub Releases (repomatic.github.releases) clients, so every datasource shares the same timeout, User-Agent and truncated-body retry semantics. Caching policy stays with the callers — each client owns its cache namespace, TTL, and serialization — while get_cached_json() shares the raw-response caching mechanics for the clients that store verbatim bodies.

Most responses are JSON (get_json()); get_text() serves the datasources that are text rather than an API payload (the astral-sh/setup-uv checksum table repomatic.release.version_sync reads, the gitignore.io template repomatic.gitignore fetches), and get_bytes() the ones written back verbatim (a downloaded label-definition file).

repomatic.http.DEFAULT_TIMEOUT = 10

Socket timeout in seconds for every HTTP fetch repomatic makes.

A stalled connection must fail the operation, not hang it.

repomatic.http.USER_AGENT = 'repomatic/7.14.1.dev0'

User-Agent header sent with every fetch this module makes.

Identifying the client beats urllib’s anonymous default on every count: some registries throttle unidentified agents harder, and an operator reading an upstream access log can tell which release of this tool hit them.

exception repomatic.http.FetchError[source]

Bases: RuntimeError

Raised when a JSON fetch could not complete cleanly.

Wraps every failure mode of get_json(): HTTP 4xx/5xx, network error, timeout, truncated body (after its one retry), and JSON parse error. Callers decide whether a failure is fatal (GitHub pagination, where a missing page corrupts the result) or a soft miss (PyPI/npm lookups, logged and treated as “no data”).

repomatic.http.get_json(url, *, headers=None, timeout=10)[source]

GET url and parse the body as JSON, retrying once on truncation.

Parameters:

  • url (str) – The URL to fetch.

  • headers (Mapping[str, str] | None) – Extra request headers, merged over the JSON Accept default (caller wins on conflict).

  • timeout (float) – Socket timeout in seconds.

Return type:

tuple[Any, bytes]

Returns:

(parsed, raw_bytes): the decoded JSON value and the raw body (for callers that cache the verbatim response).

Raises:

FetchError – On any failure (see the class docstring).

repomatic.http.get_text(url, *, headers=None, timeout=10)[source]

GET url and decode the body as UTF-8 text.

An undecodable body is a failed fetch rather than something to paper over with replacement characters: every caller parses what it reads, and parsing mojibake yields a wrong answer instead of a missing one.

Parameters:

  • url (str) – The URL to fetch.

  • headers (Mapping[str, str] | None) – Extra request headers, merged over the plain-text Accept default (caller wins on conflict).

  • timeout (float) – Socket timeout in seconds.

Return type:

str

Returns:

The decoded response body.

Raises:

FetchError – On any failure (see the class docstring), including a body that is not valid UTF-8.

repomatic.http.get_bytes(url, *, headers=None, timeout=10)[source]

GET url and return its raw body.

For payloads written back verbatim (a downloaded label-definition file), where decoding would only risk corrupting bytes nothing here reads.

Parameters:

  • url (str) – The URL to fetch.

  • headers (Mapping[str, str] | None) – Extra request headers, merged over the wildcard Accept default (caller wins on conflict).

  • timeout (float) – Socket timeout in seconds.

Return type:

bytes

Returns:

The raw response body.

Raises:

FetchError – On any failure (see the class docstring).

repomatic.http.get_json_soft(url, log_label)[source]

GET url as JSON, logging any failure as a soft miss.

Parameters:

  • url (str) – The URL to fetch.

  • log_label (str) – Human-readable label for the debug log on failure.

Return type:

tuple[Any, bytes] | None

Returns:

(parsed, raw_bytes), or None on any failure (HTTP error, network error, timeout, JSON parse error).

repomatic.http.get_cached_json(namespace, key, url, *, ttl, log_label, force_refresh=False)[source]

GET url as JSON through the raw-response cache.

A fresh cached body under namespace/key short-circuits the network; otherwise the response is fetched, cached verbatim (when ttl is positive), and returned parsed. The caller keeps the caching policy: it picks the namespace, the cache key, and the TTL.

Note

force_refresh skips the cache read but keeps the write, which is what separates it from ttl=0: the latter also skips the store, so a caller using it to bypass a stale entry would leave that entry in place for the next reader. A forced refresh replaces it.

Parameters:

  • namespace (str) – Cache namespace (like "pypi" or "npm").

  • key (str) – Cache key within the namespace, usually the package name.

  • url (str) – The URL to fetch on a cache miss.

  • ttl (int) – Freshness TTL in seconds; 0 disables caching.

  • log_label (str) – Human-readable label for the debug log on failure.

  • force_refresh (bool) – Ignore any cached body and re-fetch, then store the fresh response.

Return type:

Any | None

Returns:

The parsed JSON value, or None on any fetch failure.