Payload Loader API

The payload_loader module is the single owner of the payload pipeline. One JSON payload moves through this order: session memo, persistent cache, CDN fetch, validation, then memo and cache write-back. Session and the async fetch layer route their fetches through this module.

Overview

The module contains one protocol, two transport implementations, one loader class, and one factory function:

HttpTransport — the HTTP seam, defined as a typing.Protocol.
NiquestsTransport — the production transport, built on the shared niquests session.
InMemoryTransport — a test transport that serves payloads from an in-memory mapping.
PayloadLoader — the two-method interface (get for one path, get_many for async fan-out) that hides the whole pipeline.
get_url_loader() — a thread-safe factory for the shared loader used for absolute-URL payloads.

Session constructs one PayloadLoader per session and injects its own memo, cache callables, and overridable fetch delegates (Session._fetch_from_cdn and Session._fetch_from_cdn_fast). CDN source ordering and fallback live in tif1.cdn.CDNManager, not in this module. The loader supplies only the per-URL fetch callable built on the transport. Validation failures always raise. There is no patched-callable escape hatch.

from tif1.payload_loader import (
    HttpTransport,
    InMemoryTransport,
    NiquestsTransport,
    PayloadLoader,
    get_url_loader,
)

HttpTransport

HttpTransport is a runtime-checkable typing.Protocol. Implementations fetch one URL and return the parsed JSON object.

class HttpTransport(Protocol):
    def get_json(self, url: str, *, timeout: float | None = None) -> dict[str, Any]: ...

Implementations must raise these exceptions:

DataNotFoundError — the resource does not exist (HTTP 404).
NetworkError — the request failed at the transport level.
InvalidDataError — the response body was not a JSON object.

Write a custom transport to route fetches through a proxy, a recording layer, or a local fixture set.

NiquestsTransport

NiquestsTransport is the production transport. It sends every request through the pooled niquests session owned by tif1.http_session.

from tif1.payload_loader import NiquestsTransport

transport = NiquestsTransport()
data = transport.get_json("https://example.com/payload.json", timeout=10.0)

Methods

`get_json`

get_json(self, url: str, *, timeout: float | None = None) -> dict[str, Any]

Fetch url with the pooled niquests session and parse the JSON body.

str

required

Absolute URL to fetch.

float | None

default:"None"

Request timeout in seconds. When None, the transport reads the timeout value from the configuration. The configuration default is 30.

Behavior:

A transport-level failure raises NetworkError with the URL and no status code.
HTTP 404 raises DataNotFoundError with the URL.
Any other HTTP error status raises NetworkError with the URL and the status code.
A response body that decodes to a non-dict JSON value raises InvalidDataError.

InMemoryTransport

InMemoryTransport serves payloads from an in-memory mapping. Use it in tests to run the full pipeline without network access. Every requested URL is recorded in the calls attribute.

from tif1.payload_loader import InMemoryTransport

transport = InMemoryTransport({"drivers.json": {"drivers": []}})
data = transport.get_json("https://cdn.example.com/2025/main/GP/R/drivers.json")

Constructor

InMemoryTransport(self, payloads: Mapping[str, Any] | None = None, *, default: Any = <unset>)

Mapping[str, Any] | None

default:"None"

Mapping of URL fragment to payload. A fragment can be a path, a full URL, or any substring of a URL. A value can be a dict, any other payload value, an Exception class or instance to raise, or a callable. A callable takes the URL and returns any of those value types.

Any

default:"sentinel"

Fallback entry used when no mapping matches. When unset, an unmatched URL raises DataNotFoundError.

Matching prefers an exact URL match, then the longest fragment contained in the requested URL.

Methods

`add`

add(self, key: str, value: Any) -> None

`get_json`

get_json(self, url: str, *, timeout: float | None = None) -> dict[str, Any]

Resolve url against the in-memory mapping and record url in calls. Raises DataNotFoundError when no mapping matched and no default was provided. Raises the matched exception when the matched entry specifies one.

PayloadLoader

PayloadLoader owns the pipeline order: memo, cache, CDN fetch, validation, write-back. The loader is parameterized by session coordinates (year, gp, session) that build cache keys and CDN URLs. All storage and the fetch step are injectable as callables.

Constructor

PayloadLoader(
    self,
    year: int | None = None,
    gp: str | None = None,
    session: str | None = None,
    *,
    transport: HttpTransport | None = None,
    cdn_manager: CDNManager | None = None,
    memo: SessionMemo | None = None,
    enable_cache: bool = True,
    fetch: Callable[..., Any] | None = None,
    memo_get: Callable[[str], dict[str, Any] | None] | None = None,
    memo_set: Callable[[str, dict[str, Any]], None] | None = None,
    cache_get: Callable[[str], Any | None] | None = None,
    cache_set: Callable[[str, dict[str, Any]], None] | None = None,
) -> None

int | None

default:"None"

Season year. None is valid only for absolute-URL loaders.

str | None

default:"None"

Grand Prix identifier, URL-encoded.

str | None

default:"None"

Session identifier, URL-encoded.

HttpTransport | None

default:"None"

HTTP seam implementation. Defaults to NiquestsTransport().

CDNManager | None

default:"None"

CDN fallback manager. Defaults to the process-global tif1.cdn.get_cdn_manager() instance, resolved at call time. Also settable through the cdn_manager property.

SessionMemo | None

default:"None"

Per-session memo tier used by the default memo callables.

bool

default:"True"

Gate for the default persistent-cache callables.

Callable[..., Any] | None

default:"None"

Override for the fetch step, with signature fetch(path, *, fast: bool). Defaults to PayloadLoader.fetch_from_cdn.

Callable[[str], dict[str, Any] | None] | None

default:"None"

Override for the memo read, path -> dict or None.

Callable[[str, dict[str, Any]], None] | None

default:"None"

Override for the memo write, (path, dict).

Callable[[str], Any | None] | None

default:"None"

Override for the persistent-cache read, cache key -> data or None.

Callable[[str, dict[str, Any]], None] | None

default:"None"

Override for the persistent-cache write, (cache key, dict).

Methods

`get`

get(
    self,
    path: str,
    *,
    validate: bool = True,
    use_cache: bool = True,
    write_cache: bool = True,
    fast: bool = False,
) -> dict[str, Any]

Fetch one session-relative JSON payload through the full pipeline. The order is: session memo, persistent cache, CDN fetch, validation, memo and cache write-back.

str

required

Session-relative payload path, for example "drivers.json".

bool

default:"True"

Run payload validation before returning. Validation is path-based and pydantic-backed; it is gated by the validate_data, validate_lap_times, and validate_telemetry configuration values.

bool

default:"True"

Read the persistent cache before a network fetch.

bool

default:"True"

Write fetched payloads to the persistent cache.

bool

default:"False"

Skip per-source retry and backoff delays on the CDN fetch.

Raises DataNotFoundError when the payload does not exist. Raises InvalidDataError when the payload is not a JSON object or failed validation. Raises NetworkError when all CDN sources failed. Validation failures always raise. A payload found in the memo or cache returns without a network request. A cached dict is also written to the memo.

`fetch_from_cdn`

fetch_from_cdn(self, path: str, *, fast: bool = False) -> dict[str, Any]

Fetch a payload over HTTP with CDN fallback. CDN source ordering and fallback live in CDNManager.try_sources; this method only builds the per-URL fetch callable on the configured transport. When fast is False, the fetch callable is wrapped in retry_with_backoff using the max_retries, retry_backoff_factor, and retry_jitter configuration values. When fast is True, the zero-retry path is used. Raises DataNotFoundError when the payload does not exist (HTTP 404) and NetworkError when every CDN source failed.

`get_url`

get_url(self, url: str, *, timeout: float | None = None) -> dict[str, Any]

Fetch a JSON object from an absolute URL. This method bypasses the memo, the cache, and CDN fallback; it calls the transport directly. Use it for resources outside the per-session CDN layout. The timeout default comes from the configuration.

`get_many`

async get_many(
    self,
    paths: list[str],
    *,
    use_cache: bool = True,
    write_cache: bool = True,
    validate: bool = True,
    max_retries: int | None = None,
    timeout: int | None = None,
    max_concurrent_requests: int | None = None,
) -> list[dict[str, Any] | None]

Fetch many session-relative payloads concurrently. This coroutine delegates to tif1.async_fetch.fetch_multiple_async. Every keyword argument defaults to the configured value when None. The returned list has one entry per requested path; a failed fetch produces None at its position.

`cache_key`

cache_key(self, path: str) -> str

Build the persistent-cache key for a session-relative path. The key format is "{year}/{gp}/{session}/{path}".

`cdn_manager`

cdn_manager(self) -> CDNManager   # property, settable

The CDN manager used for fallback. When no manager was injected, the property returns the process-global instance from tif1.cdn.get_cdn_manager().

get_url_loader

get_url_loader() -> PayloadLoader

Return the shared PayloadLoader for absolute-URL payloads. The loader is created on first call under a lock, so the function is safe to call from multiple threads. tif1.events uses this loader to fetch f1schedule year payloads from their CDN URLs.

Examples

Offline pipeline run with InMemoryTransport

This example runs the full get pipeline without network access. The enable_cache=False argument keeps the run off the persistent cache. Validation is disabled because the fixture payload is not a real drivers.json document.

from tif1.payload_loader import InMemoryTransport, PayloadLoader

transport = InMemoryTransport({"drivers.json": {"drivers": []}})
loader = PayloadLoader(
    2025,
    "Abu%20Dhabi%20Grand%20Prix",
    "Race",
    transport=transport,
    enable_cache=False,
)

data = loader.get("drivers.json", validate=False)
data = loader.get("drivers.json", validate=False)  # served from the memo

print(transport.calls[0])
# https://cdn.staticdelivr.com/gh/TracingInsights/2025/main/Abu%20Dhabi%20Grand%20Prix/Race/drivers.json

Absolute-URL fetch

This example requires network access.

from tif1.payload_loader import get_url_loader

data = get_url_loader().get_url("https://example.com/data.json", timeout=15.0)

Core API: Session and its overridable _fetch_from_cdn delegates
CDN API: CDNManager source ordering and fallback
Cache API: SessionMemo and the persistent SQLite cache
Async Fetch API: fetch_multiple_async used by get_many
Config API: timeout and retry configuration values
Validation API: the pydantic validators used by the pipeline

Getting Started

Core API

Data Pipeline

Configuration & Cache

Visualization & Tools

Utilities & Helpers

Compatibility & Errors

Payload Loader API

Overview

HttpTransport

NiquestsTransport

Methods

`get_json`

InMemoryTransport

Constructor

Methods

`add`

`get_json`

PayloadLoader

Constructor

Methods

`get`

`fetch_from_cdn`

`get_url`

`get_many`

`cache_key`

`cdn_manager`

get_url_loader

Examples

Offline pipeline run with InMemoryTransport

Absolute-URL fetch

​Overview

​HttpTransport

​NiquestsTransport

​Methods

​get_json

​InMemoryTransport

​Constructor

​Methods

​add

​get_json

​PayloadLoader

​Constructor

​Methods

​get

​fetch_from_cdn

​get_url

​get_many

​cache_key

​cdn_manager

​get_url_loader

​Examples

​Offline pipeline run with InMemoryTransport

​Absolute-URL fetch

​Related APIs

Overview

HttpTransport

NiquestsTransport

Methods

`get_json`

InMemoryTransport

Constructor

Methods

`add`

`get_json`

PayloadLoader

Constructor

Methods

`get`

`fetch_from_cdn`

`get_url`

`get_many`

`cache_key`

`cdn_manager`

get_url_loader

Examples

Offline pipeline run with InMemoryTransport

Absolute-URL fetch

Related APIs