tempestweb.query

The read side of remote data: a cache with hierarchical keys, prefix invalidation, single-flight, both pagination shapes, and optimistic mutation with an exact rollback. Modes A and B (Mode C refuses the import at build time). It does not replace native.sync, which is still the way to reconcile a large collection.

Tutorial with examples: Reading remote data.

tempestweb.query

The read side of remote data: cache, keys, pagination, optimistic updates.

tempestweb had both hard ends and nothing in between. native.http retries with backoff and idempotency, native.offline holds a durable FIFO of mutations, and native.sync reconciles a collection by watermark. Reading had nothing: nowhere to keep the answer to a GET under a key, invalidate it when a mutation lands, paginate, or put a change on screen before the server agreed to it.

Every app wrote that as a dict inside its own State, and the part that always came out wrong was the invalidation.

Modules

* :mod:`keys` — hierarchical keys, so invalidation is by prefix.
* :mod:`cache` — :class:`QueryCache`: staleness, single-flight, rollback.
* :mod:`pagination` — the offset and cursor shapes, typed.
* :mod:`optimistic` — `upsert_by_id` / `remove_by_id` over a cached list.
* :mod:`policy` — how long an answer is fresh, and what is worth retrying.
* :mod:`persistence` — writing the cache to the store the app already has.

Example

from tempestweb import native
from tempestweb.query import QueryCache, keys, offset_page, upsert_by_id

USERS = keys("users")
CACHE = QueryCache()

response = await CACHE.fetch(
    USERS.list(page=1),
    lambda: native.http.request("GET", "/api/users?page=1"),
)
page = offset_page(response.json)

with CACHE.optimistic(USERS.all(), lambda rows: upsert_by_id(rows, edited)):
    await native.http.request("PATCH", "/api/users/7", json=edited)

If the PATCH raises, the block's rollback puts back exactly the entries it replaced — no refetch needed to undo something the server never accepted.

The cache is app state, not a hidden singleton

A QueryCache is created by the app and kept in its State. There is no module-level instance and no implicit context: the view reads from the cache it was handed, and a test builds its own with a fake clock.

Modes A and B only

Mode C transpiles the app's own Python into JavaScript and serves a fixed set of modules — tempest_core, tempestweb.components and tempestweb.native. Importing this package from a Mode C app is refused at build time with a named error.

This does not replace native.sync

Delta-sync is still the way to reconcile a large collection against a watermark. This cache is for reading a screen.

Import everything from this package level rather than from submodules.

QueryCache

Keyed cache of read answers, with staleness, single-flight and rollback.

The second read below never runs its loader: the first answer is still inside the staleness window, so the cache answers it.

Example

cache = QueryCache()
rows = await cache.fetch(USERS.list(), load_from_network)
again = await cache.fetch(USERS.list(), load_from_network)  # no request

Source code in tempestweb/query/cache.py

class QueryCache:
    """Keyed cache of read answers, with staleness, single-flight and rollback.

    The second read below never runs its loader: the first answer is still
    inside the staleness window, so the cache answers it.

    Example:
        ```python
        cache = QueryCache()
        rows = await cache.fetch(USERS.list(), load_from_network)
        again = await cache.fetch(USERS.list(), load_from_network)  # no request
        ```
    """

    def __init__(
        self,
        *,
        clock: Clock = _monotonic_ms,
        stale_ms: float = STALE_TIME_MS,
        cache_ms: float = CACHE_TIME_MS,
    ) -> None:
        """Build an empty cache.

        Args:
            clock: Reads the current time in milliseconds.
            stale_ms: How long an answer is served without going back to the
                network.
            cache_ms: How long an entry survives at all. Longer than
                ``stale_ms``, so a stale answer is still on screen while its
                refetch is in flight.
        """
        self._entries: dict[QueryKey, QueryEntry] = {}
        self._inflight: dict[QueryKey, asyncio.Future[object]] = {}
        self._listeners: list[Listener] = []
        self._clock = clock
        self._stale_ms = stale_ms
        self._cache_ms = cache_ms

    # -- reading ---------------------------------------------------------

    async def fetch(
        self,
        key: QueryKey,
        loader: Callable[[], Awaitable[T]],
        *,
        stale_ms: float | None = None,
        force: bool = False,
    ) -> T:
        """Answer from cache when fresh, otherwise run the loader once.

        Concurrent calls for the same key **share one loader run**: the second
        caller awaits the first one's result rather than issuing a second
        request. That is single-flight, and it is the behaviour a screen with
        three widgets reading the same query needs.

        Args:
            key: The cache key, from :func:`~tempestweb.query.keys`.
            loader: Called to produce the value when the cache cannot answer.
            stale_ms: Override the cache's staleness window for this read.
            force: Skip the freshness check and load anyway. The in-flight share
                still applies, so forcing twice concurrently still loads once.

        Returns:
            The value, cached or freshly loaded.

        Raises:
            Exception: Whatever the loader raises, to every caller sharing the
                run. A failed load leaves the previous entry alone — showing the
                last good answer beats blanking the screen because a refetch
                failed.
        """
        window = self._stale_ms if stale_ms is None else stale_ms
        now = self._clock()
        self._collect(now)

        if not force:
            entry = self._entries.get(key)
            if entry is not None and now - entry.updated_at < window:
                return cast(T, entry.value)

        task = self._inflight.get(key)
        if task is None:
            task = asyncio.ensure_future(_awaited(loader()))
            self._inflight[key] = task
            task.add_done_callback(lambda done: self._settle(key, done))
        return cast(T, await asyncio.shield(task))

    def get(self, key: QueryKey) -> object | None:
        """Read a cached value without loading anything.

        Args:
            key: The cache key.

        Returns:
            The value, or ``None`` when nothing is cached under that key or the
            entry has aged past the cache window.
        """
        self._collect(self._clock())
        entry = self._entries.get(key)
        return None if entry is None else entry.value

    def is_stale(self, key: QueryKey, *, stale_ms: float | None = None) -> bool:
        """Report whether a key needs a trip to the network.

        Args:
            key: The cache key.
            stale_ms: Override the cache's staleness window.

        Returns:
            ``True`` when nothing is cached, or when the entry is older than the
            window.
        """
        window = self._stale_ms if stale_ms is None else stale_ms
        entry = self._entries.get(key)
        if entry is None:
            return True
        return self._clock() - entry.updated_at >= window

    @property
    def keys(self) -> tuple[QueryKey, ...]:
        """Every key currently held, in insertion order.

        Returns:
            The keys.
        """
        return tuple(self._entries)

    # -- writing ---------------------------------------------------------

    def set(self, key: QueryKey, value: object) -> None:
        """Store a value, stamping it fresh.

        Args:
            key: The cache key.
            value: The value to store.
        """
        self._entries[key] = QueryEntry(value=value, updated_at=self._clock())
        self._notify()

    def invalidate(self, prefix: QueryKey) -> int:
        """Mark everything under a prefix stale, keeping the values on screen.

        This is the operation the hierarchy exists for: ``invalidate(("users",))``
        reaches ``("users", "list", "page=1")``, ``("users", "detail", "7")`` and
        everything else about users, without the caller keeping a second registry
        of which keys mean users.

        The values stay, so a screen keeps showing the last good answer while the
        refetch is in flight. Use :meth:`drop` when the value is known to be
        wrong rather than merely old.

        Args:
            prefix: The prefix to invalidate. The empty tuple reaches everything.

        Returns:
            How many entries were marked.
        """
        stamp = self._clock() - self._stale_ms
        marked = 0
        for key, entry in self._entries.items():
            if is_under(prefix, key):
                self._entries[key] = QueryEntry(entry.value, stamp)
                marked += 1
        if marked:
            self._notify()
        return marked

    def drop(self, prefix: QueryKey) -> int:
        """Remove everything under a prefix.

        Args:
            prefix: The prefix to drop. The empty tuple clears the cache.

        Returns:
            How many entries were removed.
        """
        doomed = [key for key in self._entries if is_under(prefix, key)]
        for key in doomed:
            del self._entries[key]
        if doomed:
            self._notify()
        return len(doomed)

    def patch(self, prefix: QueryKey, patcher: Patcher) -> Rollback:
        """Apply an optimistic change to every entry under a prefix.

        A prefix rather than one key, because a rename has to reach every cached
        page the row appears on — patching only ``("users", "list", "page=1")``
        leaves page 2 showing the old name until something else invalidates it.

        Args:
            prefix: The prefix whose entries are patched.
            patcher: Turns each entry's value into its replacement. Must not
                mutate the value it is handed.

        Returns:
            A callable restoring exactly the entries this patch replaced,
            timestamps included. Calling it twice is harmless.

        Raises:
            Exception: Whatever ``patcher`` raises. Entries already patched are
                restored first, so a patch either lands everywhere it applies or
                nowhere — a half-applied optimistic update is a screen showing
                two different truths.
        """
        touched: dict[QueryKey, QueryEntry] = {}
        try:
            for key in tuple(self._entries):
                if not is_under(prefix, key):
                    continue
                entry = self._entries[key]
                touched[key] = entry
                self._entries[key] = QueryEntry(patcher(entry.value), entry.updated_at)
        except BaseException:
            self._restore(touched)
            raise
        if touched:
            self._notify()
        return lambda: self._restore(touched)

    @contextmanager
    def optimistic(self, prefix: QueryKey, patcher: Patcher) -> Iterator[Rollback]:
        """Apply a patch, and undo it if the block raises.

        The shape a mutation wants, because the rollback cannot be forgotten:

        ```python
        with cache.optimistic(USERS.all(), rename) as rollback:
            await native.http.request("PATCH", f"/api/users/{user_id}", json=body)
        ```

        Args:
            prefix: The prefix whose entries are patched.
            patcher: Turns each entry's value into its replacement.

        Yields:
            The rollback, for a block that decides to undo without raising.
        """
        rollback = self.patch(prefix, patcher)
        try:
            yield rollback
        except BaseException:
            rollback()
            raise

    def clear(self) -> None:
        """Drop every entry."""
        if self._entries:
            self._entries.clear()
            self._notify()

    # -- observing -------------------------------------------------------

    def on_change(self, listener: Listener) -> Callable[[], None]:
        """Register a callback fired after any change to the cache.

        This is how a cached read reaches the screen: the app subscribes once
        and asks for a rebuild.

        Args:
            listener: Called with no arguments after each change.

        Returns:
            A callable that unsubscribes.
        """
        self._listeners.append(listener)

        def unsubscribe() -> None:
            """Stop calling this listener."""
            if listener in self._listeners:
                self._listeners.remove(listener)

        return unsubscribe

    # -- internals -------------------------------------------------------

    def _settle(self, key: QueryKey, task: asyncio.Future[object]) -> None:
        """Store a completed load and release its in-flight slot.

        Storing here rather than in :meth:`fetch` means the value lands even
        when the caller that started the load was cancelled while awaiting —
        the request was paid for either way, so throwing the answer away would
        make the next read pay again.

        Args:
            key: The key that was loaded.
            task: The completed load.
        """
        self._inflight.pop(key, None)
        if task.cancelled() or task.exception() is not None:
            return
        self.set(key, task.result())

    def _restore(self, entries: dict[QueryKey, QueryEntry]) -> None:
        """Put back a set of entries exactly as they were.

        Args:
            entries: The entries to restore, keyed as they were stored.
        """
        if not entries:
            return
        for key, entry in entries.items():
            self._entries[key] = entry
        self._notify()

    def _collect(self, now: float) -> None:
        """Drop entries that have aged past the cache window.

        Args:
            now: The current time in milliseconds.
        """
        expired = [
            key
            for key, entry in self._entries.items()
            if now - entry.updated_at >= self._cache_ms
        ]
        for key in expired:
            del self._entries[key]

    def _notify(self) -> None:
        """Fire every listener."""
        for listener in tuple(self._listeners):
            listener()

keys property

keys: tuple[QueryKey, ...]

Every key currently held, in insertion order.

Returns:

Type	Description
tuple[QueryKey, ...]	The keys.

fetch async

fetch(key: QueryKey, loader: Callable[[], Awaitable[T]], *, stale_ms: float | None = None, force: bool = False) -> T

Answer from cache when fresh, otherwise run the loader once.

Concurrent calls for the same key share one loader run: the second caller awaits the first one's result rather than issuing a second request. That is single-flight, and it is the behaviour a screen with three widgets reading the same query needs.

Parameters:

Name	Type	Description	Default
key	QueryKey	The cache key, from :func:~tempestweb.query.keys.	required
loader	Callable[[], Awaitable[T]]	Called to produce the value when the cache cannot answer.	required
stale_ms	float | None	Override the cache's staleness window for this read.	None
force	bool	Skip the freshness check and load anyway. The in-flight share still applies, so forcing twice concurrently still loads once.	False

Returns:

Type	Description
T	The value, cached or freshly loaded.

Raises:

Type	Description
Exception	Whatever the loader raises, to every caller sharing the run. A failed load leaves the previous entry alone — showing the last good answer beats blanking the screen because a refetch failed.

Source code in tempestweb/query/cache.py

async def fetch(
    self,
    key: QueryKey,
    loader: Callable[[], Awaitable[T]],
    *,
    stale_ms: float | None = None,
    force: bool = False,
) -> T:
    """Answer from cache when fresh, otherwise run the loader once.

    Concurrent calls for the same key **share one loader run**: the second
    caller awaits the first one's result rather than issuing a second
    request. That is single-flight, and it is the behaviour a screen with
    three widgets reading the same query needs.

    Args:
        key: The cache key, from :func:`~tempestweb.query.keys`.
        loader: Called to produce the value when the cache cannot answer.
        stale_ms: Override the cache's staleness window for this read.
        force: Skip the freshness check and load anyway. The in-flight share
            still applies, so forcing twice concurrently still loads once.

    Returns:
        The value, cached or freshly loaded.

    Raises:
        Exception: Whatever the loader raises, to every caller sharing the
            run. A failed load leaves the previous entry alone — showing the
            last good answer beats blanking the screen because a refetch
            failed.
    """
    window = self._stale_ms if stale_ms is None else stale_ms
    now = self._clock()
    self._collect(now)

    if not force:
        entry = self._entries.get(key)
        if entry is not None and now - entry.updated_at < window:
            return cast(T, entry.value)

    task = self._inflight.get(key)
    if task is None:
        task = asyncio.ensure_future(_awaited(loader()))
        self._inflight[key] = task
        task.add_done_callback(lambda done: self._settle(key, done))
    return cast(T, await asyncio.shield(task))

get

get(key: QueryKey) -> object | None

Read a cached value without loading anything.

Parameters:

Name	Type	Description	Default
key	QueryKey	The cache key.	required

Returns:

Type	Description
object | None	The value, or None when nothing is cached under that key or the
object | None	entry has aged past the cache window.

Source code in tempestweb/query/cache.py

def get(self, key: QueryKey) -> object | None:
    """Read a cached value without loading anything.

    Args:
        key: The cache key.

    Returns:
        The value, or ``None`` when nothing is cached under that key or the
        entry has aged past the cache window.
    """
    self._collect(self._clock())
    entry = self._entries.get(key)
    return None if entry is None else entry.value

is_stale

is_stale(key: QueryKey, *, stale_ms: float | None = None) -> bool

Report whether a key needs a trip to the network.

Parameters:

Name	Type	Description	Default
key	QueryKey	The cache key.	required
stale_ms	float | None	Override the cache's staleness window.	None

Returns:

Type	Description
bool	True when nothing is cached, or when the entry is older than the
bool	window.

Source code in tempestweb/query/cache.py

def is_stale(self, key: QueryKey, *, stale_ms: float | None = None) -> bool:
    """Report whether a key needs a trip to the network.

    Args:
        key: The cache key.
        stale_ms: Override the cache's staleness window.

    Returns:
        ``True`` when nothing is cached, or when the entry is older than the
        window.
    """
    window = self._stale_ms if stale_ms is None else stale_ms
    entry = self._entries.get(key)
    if entry is None:
        return True
    return self._clock() - entry.updated_at >= window

set

set(key: QueryKey, value: object) -> None

Store a value, stamping it fresh.

Parameters:

Name	Type	Description	Default
key	QueryKey	The cache key.	required
value	object	The value to store.	required

Source code in tempestweb/query/cache.py

def set(self, key: QueryKey, value: object) -> None:
    """Store a value, stamping it fresh.

    Args:
        key: The cache key.
        value: The value to store.
    """
    self._entries[key] = QueryEntry(value=value, updated_at=self._clock())
    self._notify()

invalidate

invalidate(prefix: QueryKey) -> int

Mark everything under a prefix stale, keeping the values on screen.

This is the operation the hierarchy exists for: invalidate(("users",)) reaches ("users", "list", "page=1"), ("users", "detail", "7") and everything else about users, without the caller keeping a second registry of which keys mean users.

The values stay, so a screen keeps showing the last good answer while the refetch is in flight. Use :meth:drop when the value is known to be wrong rather than merely old.

Parameters:

Name	Type	Description	Default
prefix	QueryKey	The prefix to invalidate. The empty tuple reaches everything.	required

Returns:

Type	Description
int	How many entries were marked.

Source code in tempestweb/query/cache.py

def invalidate(self, prefix: QueryKey) -> int:
    """Mark everything under a prefix stale, keeping the values on screen.

    This is the operation the hierarchy exists for: ``invalidate(("users",))``
    reaches ``("users", "list", "page=1")``, ``("users", "detail", "7")`` and
    everything else about users, without the caller keeping a second registry
    of which keys mean users.

    The values stay, so a screen keeps showing the last good answer while the
    refetch is in flight. Use :meth:`drop` when the value is known to be
    wrong rather than merely old.

    Args:
        prefix: The prefix to invalidate. The empty tuple reaches everything.

    Returns:
        How many entries were marked.
    """
    stamp = self._clock() - self._stale_ms
    marked = 0
    for key, entry in self._entries.items():
        if is_under(prefix, key):
            self._entries[key] = QueryEntry(entry.value, stamp)
            marked += 1
    if marked:
        self._notify()
    return marked

drop

drop(prefix: QueryKey) -> int

Remove everything under a prefix.

Parameters:

Name	Type	Description	Default
prefix	QueryKey	The prefix to drop. The empty tuple clears the cache.	required

Returns:

Type	Description
int	How many entries were removed.

Source code in tempestweb/query/cache.py

def drop(self, prefix: QueryKey) -> int:
    """Remove everything under a prefix.

    Args:
        prefix: The prefix to drop. The empty tuple clears the cache.

    Returns:
        How many entries were removed.
    """
    doomed = [key for key in self._entries if is_under(prefix, key)]
    for key in doomed:
        del self._entries[key]
    if doomed:
        self._notify()
    return len(doomed)

patch

patch(prefix: QueryKey, patcher: Patcher) -> Rollback

Apply an optimistic change to every entry under a prefix.

A prefix rather than one key, because a rename has to reach every cached page the row appears on — patching only ("users", "list", "page=1") leaves page 2 showing the old name until something else invalidates it.

Parameters:

Name	Type	Description	Default
prefix	QueryKey	The prefix whose entries are patched.	required
patcher	Patcher	Turns each entry's value into its replacement. Must not mutate the value it is handed.	required

Returns:

Type	Description
Rollback	A callable restoring exactly the entries this patch replaced,
Rollback	timestamps included. Calling it twice is harmless.

Raises:

Type	Description
Exception	Whatever patcher raises. Entries already patched are restored first, so a patch either lands everywhere it applies or nowhere — a half-applied optimistic update is a screen showing two different truths.

Source code in tempestweb/query/cache.py

def patch(self, prefix: QueryKey, patcher: Patcher) -> Rollback:
    """Apply an optimistic change to every entry under a prefix.

    A prefix rather than one key, because a rename has to reach every cached
    page the row appears on — patching only ``("users", "list", "page=1")``
    leaves page 2 showing the old name until something else invalidates it.

    Args:
        prefix: The prefix whose entries are patched.
        patcher: Turns each entry's value into its replacement. Must not
            mutate the value it is handed.

    Returns:
        A callable restoring exactly the entries this patch replaced,
        timestamps included. Calling it twice is harmless.

    Raises:
        Exception: Whatever ``patcher`` raises. Entries already patched are
            restored first, so a patch either lands everywhere it applies or
            nowhere — a half-applied optimistic update is a screen showing
            two different truths.
    """
    touched: dict[QueryKey, QueryEntry] = {}
    try:
        for key in tuple(self._entries):
            if not is_under(prefix, key):
                continue
            entry = self._entries[key]
            touched[key] = entry
            self._entries[key] = QueryEntry(patcher(entry.value), entry.updated_at)
    except BaseException:
        self._restore(touched)
        raise
    if touched:
        self._notify()
    return lambda: self._restore(touched)

optimistic

optimistic(prefix: QueryKey, patcher: Patcher) -> Iterator[Rollback]

Apply a patch, and undo it if the block raises.

The shape a mutation wants, because the rollback cannot be forgotten:

with cache.optimistic(USERS.all(), rename) as rollback:
    await native.http.request("PATCH", f"/api/users/{user_id}", json=body)

Parameters:

Name	Type	Description	Default
prefix	QueryKey	The prefix whose entries are patched.	required
patcher	Patcher	Turns each entry's value into its replacement.	required

Yields:

Type	Description
Rollback	The rollback, for a block that decides to undo without raising.

Source code in tempestweb/query/cache.py

@contextmanager
def optimistic(self, prefix: QueryKey, patcher: Patcher) -> Iterator[Rollback]:
    """Apply a patch, and undo it if the block raises.

    The shape a mutation wants, because the rollback cannot be forgotten:

    ```python
    with cache.optimistic(USERS.all(), rename) as rollback:
        await native.http.request("PATCH", f"/api/users/{user_id}", json=body)
    ```

    Args:
        prefix: The prefix whose entries are patched.
        patcher: Turns each entry's value into its replacement.

    Yields:
        The rollback, for a block that decides to undo without raising.
    """
    rollback = self.patch(prefix, patcher)
    try:
        yield rollback
    except BaseException:
        rollback()
        raise

clear

clear() -> None

Drop every entry.

Source code in tempestweb/query/cache.py

def clear(self) -> None:
    """Drop every entry."""
    if self._entries:
        self._entries.clear()
        self._notify()

on_change

on_change(listener: Listener) -> Callable[[], None]

Register a callback fired after any change to the cache.

This is how a cached read reaches the screen: the app subscribes once and asks for a rebuild.

Parameters:

Name	Type	Description	Default
listener	Listener	Called with no arguments after each change.	required

Returns:

Type	Description
Callable[[], None]	A callable that unsubscribes.

Source code in tempestweb/query/cache.py

def on_change(self, listener: Listener) -> Callable[[], None]:
    """Register a callback fired after any change to the cache.

    This is how a cached read reaches the screen: the app subscribes once
    and asks for a rebuild.

    Args:
        listener: Called with no arguments after each change.

    Returns:
        A callable that unsubscribes.
    """
    self._listeners.append(listener)

    def unsubscribe() -> None:
        """Stop calling this listener."""
        if listener in self._listeners:
            self._listeners.remove(listener)

    return unsubscribe

QueryEntry dataclass

One cached answer.

Attributes:

Name	Type	Description
value	object	Whatever the loader returned.
updated_at	float	When it was stored, in milliseconds from :data:Clock.

Source code in tempestweb/query/cache.py

@dataclass(frozen=True)
class QueryEntry:
    """One cached answer.

    Attributes:
        value: Whatever the loader returned.
        updated_at: When it was stored, in milliseconds from :data:`Clock`.
    """

    value: object
    updated_at: float

QueryKeys dataclass

A key factory rooted at one resource.

Attributes:

Name	Type	Description
root	QueryKey	The segments every key from this factory starts with.

Source code in tempestweb/query/keys.py

@dataclass(frozen=True)
class QueryKeys:
    """A key factory rooted at one resource.

    Attributes:
        root: The segments every key from this factory starts with.
    """

    root: QueryKey

    def all(self) -> QueryKey:
        """The root key, which is a prefix of every other key from here.

        Returns:
            The root segments — pass this to
            :meth:`~tempestweb.query.QueryCache.invalidate` to reach everything
            about this resource.
        """
        return self.root

    def list(self, **params: object) -> QueryKey:
        """A key for a listing, parameterized.

        Args:
            **params: Query parameters — page, filters, sort. Sorted by name
                before joining, so argument order never splits the cache.

        Returns:
            The key, under :meth:`all`.
        """
        return (*self.root, "list", *_params(params))

    def detail(self, identifier: object, **params: object) -> QueryKey:
        """A key for a single record.

        Args:
            identifier: The record's id, rendered with ``str``.
            **params: Any extra parameters, sorted as in :meth:`list`.

        Returns:
            The key, under :meth:`all`.
        """
        return (*self.root, "detail", str(identifier), *_params(params))

    def sub(self, *segments: object, **params: object) -> QueryKey:
        """A key for anything the other two do not name.

        Args:
            *segments: Extra segments, rendered with ``str``.
            **params: Any parameters, sorted as in :meth:`list`.

        Returns:
            The key, under :meth:`all`.
        """
        return (*self.root, *(str(part) for part in segments), *_params(params))

all

all() -> QueryKey

The root key, which is a prefix of every other key from here.

Returns:

Type	Description
QueryKey	The root segments — pass this to
QueryKey	meth:~tempestweb.query.QueryCache.invalidate to reach everything
QueryKey	about this resource.

Source code in tempestweb/query/keys.py

def all(self) -> QueryKey:
    """The root key, which is a prefix of every other key from here.

    Returns:
        The root segments — pass this to
        :meth:`~tempestweb.query.QueryCache.invalidate` to reach everything
        about this resource.
    """
    return self.root

list

list(**params: object) -> QueryKey

A key for a listing, parameterized.

Parameters:

Name	Type	Description	Default
**params	object	Query parameters — page, filters, sort. Sorted by name before joining, so argument order never splits the cache.	{}

Returns:

Type	Description
QueryKey	The key, under :meth:all.

Source code in tempestweb/query/keys.py

def list(self, **params: object) -> QueryKey:
    """A key for a listing, parameterized.

    Args:
        **params: Query parameters — page, filters, sort. Sorted by name
            before joining, so argument order never splits the cache.

    Returns:
        The key, under :meth:`all`.
    """
    return (*self.root, "list", *_params(params))

detail

detail(identifier: object, **params: object) -> QueryKey

A key for a single record.

Parameters:

Name	Type	Description	Default
identifier	object	The record's id, rendered with str.	required
**params	object	Any extra parameters, sorted as in :meth:list.	{}

Returns:

Type	Description
QueryKey	The key, under :meth:all.

Source code in tempestweb/query/keys.py

def detail(self, identifier: object, **params: object) -> QueryKey:
    """A key for a single record.

    Args:
        identifier: The record's id, rendered with ``str``.
        **params: Any extra parameters, sorted as in :meth:`list`.

    Returns:
        The key, under :meth:`all`.
    """
    return (*self.root, "detail", str(identifier), *_params(params))

sub

sub(*segments: object, **params: object) -> QueryKey

A key for anything the other two do not name.

Parameters:

Name	Type	Description	Default
*segments	object	Extra segments, rendered with str.	()
**params	object	Any parameters, sorted as in :meth:list.	{}

Returns:

Type	Description
QueryKey	The key, under :meth:all.

Source code in tempestweb/query/keys.py

def sub(self, *segments: object, **params: object) -> QueryKey:
    """A key for anything the other two do not name.

    Args:
        *segments: Extra segments, rendered with ``str``.
        **params: Any parameters, sorted as in :meth:`list`.

    Returns:
        The key, under :meth:`all`.
    """
    return (*self.root, *(str(part) for part in segments), *_params(params))

CursorPage dataclass

One page of a cursor-paginated answer.

Attributes:

Name	Type	Description
items	tuple[object, ...]	The rows on this page.
next_cursor	str | None	The cursor to ask for the next page, or None when this is the last one.

Source code in tempestweb/query/pagination.py

@dataclass(frozen=True)
class CursorPage:
    """One page of a cursor-paginated answer.

    Attributes:
        items: The rows on this page.
        next_cursor: The cursor to ask for the next page, or ``None`` when this
            is the last one.
    """

    items: tuple[object, ...] = ()
    next_cursor: str | None = None

    @property
    def has_next(self) -> bool:
        """Whether a page follows this one.

        Returns:
            Whether a cursor was handed back.
        """
        return self.next_cursor is not None

has_next property

has_next: bool

Whether a page follows this one.

Returns:

Type	Description
bool	Whether a cursor was handed back.

OffsetPage dataclass

One page of an offset-paginated answer.

Attributes:

Name	Type	Description
items	tuple[object, ...]	The rows on this page.
total	int	How many rows exist across every page.
page	int	This page's 1-based number.
page_size	int	How many rows a full page holds.

Source code in tempestweb/query/pagination.py

@dataclass(frozen=True)
class OffsetPage:
    """One page of an offset-paginated answer.

    Attributes:
        items: The rows on this page.
        total: How many rows exist across every page.
        page: This page's 1-based number.
        page_size: How many rows a full page holds.
    """

    items: tuple[object, ...] = ()
    total: int = 0
    page: int = 1
    page_size: int = 0

    @property
    def pages(self) -> int:
        """How many pages the total spans.

        Returns:
            The page count, or ``0`` when the page size is unknown — dividing by
            it would raise, and a screen asking "how many pages" before the first
            answer arrives is normal, not exceptional.
        """
        if self.page_size <= 0:
            return 0
        return -(-self.total // self.page_size)

    @property
    def has_next(self) -> bool:
        """Whether a page follows this one.

        Returns:
            Whether :attr:`page` is below :attr:`pages`.
        """
        return self.page < self.pages

    @property
    def has_previous(self) -> bool:
        """Whether a page precedes this one.

        Returns:
            Whether :attr:`page` is above the first.
        """
        return self.page > 1

pages property

pages: int

How many pages the total spans.

Returns:

Type	Description
int	The page count, or 0 when the page size is unknown — dividing by
int	it would raise, and a screen asking "how many pages" before the first
int	answer arrives is normal, not exceptional.

has_next property

has_next: bool

Whether a page follows this one.

Returns:

Name	Type	Description
Whether	bool	attr:page is below :attr:pages.

has_previous property

has_previous: bool

Whether a page precedes this one.

Returns:

Name	Type	Description
Whether	bool	attr:page is above the first.

PageKeys dataclass

Which payload keys to read, for a server naming them differently.

Attributes:

Name	Type	Description
items	str	The key holding the rows.
total	str	The key holding the overall count.
page	str	The key holding the page number.
page_size	str	The key holding the page size.
cursor	str	The key holding the next cursor.

Source code in tempestweb/query/pagination.py

@dataclass(frozen=True)
class PageKeys:
    """Which payload keys to read, for a server naming them differently.

    Attributes:
        items: The key holding the rows.
        total: The key holding the overall count.
        page: The key holding the page number.
        page_size: The key holding the page size.
        cursor: The key holding the next cursor.
    """

    items: str = ITEMS_KEY
    total: str = TOTAL_KEY
    page: str = PAGE_KEY
    page_size: str = PAGE_SIZE_KEY
    cursor: str = CURSOR_KEY

PersistResult dataclass

What :func:persist did.

Attributes:

Name	Type	Description
written	int	How many entries reached the store.
skipped	int	How many were left behind because their value is not JSON-able.

Source code in tempestweb/query/persistence.py

@dataclass(frozen=True)
class PersistResult:
    """What :func:`persist` did.

    Attributes:
        written: How many entries reached the store.
        skipped: How many were left behind because their value is not JSON-able.
    """

    written: int = 0
    skipped: int = 0

QueryStorage

Bases: Protocol

The slice of native.storage this module needs.

Declared as a Protocol rather than importing native.storage directly so the module runs without a browser bridge — a test passes a fake, and the dependency arrow never points from query into native.

Source code in tempestweb/query/persistence.py

@runtime_checkable
class QueryStorage(Protocol):
    """The slice of ``native.storage`` this module needs.

    Declared as a Protocol rather than importing ``native.storage`` directly so
    the module runs without a browser bridge — a test passes a fake, and the
    dependency arrow never points from ``query`` into ``native``.
    """

    def put(self, name: str, content: str) -> Awaitable[None]:
        """Store a string under a key.

        Args:
            name: The storage key.
            content: The string to store.

        Returns:
            An awaitable completing when the write lands.
        """
        ...

    def get(self, name: str) -> Awaitable[str]:
        """Read the string stored under a key.

        Args:
            name: The storage key.

        Returns:
            An awaitable resolving to the stored string.
        """
        ...

    def remove(self, name: str) -> Awaitable[None]:
        """Delete the value stored under a key.

        Args:
            name: The storage key.

        Returns:
            An awaitable completing when the delete lands.
        """
        ...

    def list_keys(self) -> Awaitable[list[str]]:
        """List every key the store holds.

        Returns:
            An awaitable resolving to the keys.
        """
        ...

put

put(name: str, content: str) -> Awaitable[None]

Store a string under a key.

Parameters:

Name	Type	Description	Default
name	str	The storage key.	required
content	str	The string to store.	required

Returns:

Type	Description
Awaitable[None]	An awaitable completing when the write lands.

Source code in tempestweb/query/persistence.py

def put(self, name: str, content: str) -> Awaitable[None]:
    """Store a string under a key.

    Args:
        name: The storage key.
        content: The string to store.

    Returns:
        An awaitable completing when the write lands.
    """
    ...

get

get(name: str) -> Awaitable[str]

Read the string stored under a key.

Parameters:

Name	Type	Description	Default
name	str	The storage key.	required

Returns:

Type	Description
Awaitable[str]	An awaitable resolving to the stored string.

Source code in tempestweb/query/persistence.py

def get(self, name: str) -> Awaitable[str]:
    """Read the string stored under a key.

    Args:
        name: The storage key.

    Returns:
        An awaitable resolving to the stored string.
    """
    ...

remove

remove(name: str) -> Awaitable[None]

Delete the value stored under a key.

Parameters:

Name	Type	Description	Default
name	str	The storage key.	required

Returns:

Type	Description
Awaitable[None]	An awaitable completing when the delete lands.

Source code in tempestweb/query/persistence.py

def remove(self, name: str) -> Awaitable[None]:
    """Delete the value stored under a key.

    Args:
        name: The storage key.

    Returns:
        An awaitable completing when the delete lands.
    """
    ...

list_keys

list_keys() -> Awaitable[list[str]]

List every key the store holds.

Returns:

Type	Description
Awaitable[list[str]]	An awaitable resolving to the keys.

Source code in tempestweb/query/persistence.py

def list_keys(self) -> Awaitable[list[str]]:
    """List every key the store holds.

    Returns:
        An awaitable resolving to the keys.
    """
    ...

RestoreResult dataclass

What :func:restore did.

Attributes:

Name	Type	Description
restored	int	How many entries were read back into the cache.
discarded	int	How many stored records could not be read and were deleted.

Source code in tempestweb/query/persistence.py

@dataclass(frozen=True)
class RestoreResult:
    """What :func:`restore` did.

    Attributes:
        restored: How many entries were read back into the cache.
        discarded: How many stored records could not be read and were deleted.
    """

    restored: int = 0
    discarded: int = 0

is_under

is_under(prefix: QueryKey, key: QueryKey) -> bool

Report whether a key lives under a prefix.

Segment-wise, never character-wise: ("users",) is a prefix of ("users", "list") and is not a prefix of ("users-archive",). A startswith on joined strings would get that second one wrong, and it would get it wrong silently — invalidating a resource that merely shares a name.

Parameters:

Name	Type	Description	Default
prefix	QueryKey	The prefix to test against.	required
key	QueryKey	The key to test.	required

Returns:

Type	Description
bool	Whether key starts with prefix. The empty prefix matches
bool	everything, which is how "invalidate the whole cache" is spelled.

Source code in tempestweb/query/keys.py

def is_under(prefix: QueryKey, key: QueryKey) -> bool:
    """Report whether a key lives under a prefix.

    Segment-wise, never character-wise: ``("users",)`` is a prefix of
    ``("users", "list")`` and is **not** a prefix of ``("users-archive",)``.
    A ``startswith`` on joined strings would get that second one wrong, and it
    would get it wrong silently — invalidating a resource that merely shares a
    name.

    Args:
        prefix: The prefix to test against.
        key: The key to test.

    Returns:
        Whether ``key`` starts with ``prefix``. The empty prefix matches
        everything, which is how "invalidate the whole cache" is spelled.
    """
    return key[: len(prefix)] == prefix

remove_by_id

remove_by_id(rows: Iterable[object], identifier: object, *, id_field: str = ID_FIELD) -> tuple[object, ...]

Drop every row carrying an id.

Parameters:

Name	Type	Description	Default
rows	Iterable[object]	The rows currently cached.	required
identifier	object	The id to drop.	required
id_field	str	The field the rows are identified by.	ID_FIELD

Returns:

Type	Description
object	A new tuple without those rows. Removing an id that is not there is not
...	an error — it answers the same rows back, which is what a double-click on
tuple[object, ...]	Delete should do.

Source code in tempestweb/query/optimistic.py

def remove_by_id(
    rows: Iterable[object],
    identifier: object,
    *,
    id_field: str = ID_FIELD,
) -> tuple[object, ...]:
    """Drop every row carrying an id.

    Args:
        rows: The rows currently cached.
        identifier: The id to drop.
        id_field: The field the rows are identified by.

    Returns:
        A new tuple without those rows. Removing an id that is not there is not
        an error — it answers the same rows back, which is what a double-click on
        Delete should do.
    """
    return tuple(
        row
        for row in rows
        if (found := _identifier(row, id_field)) is None or found != identifier
    )

replace_where

replace_where(rows: Iterable[object], matches: object, row: object) -> tuple[object, ...]

Replace every row a predicate accepts.

For the cases upsert_by_id does not cover — a composite key, a row identified by something other than a field.

Parameters:

Name	Type	Description	Default
rows	Iterable[object]	The rows currently cached.	required
matches	object	A callable answering whether a row should be replaced.	required
row	object	The replacement.	required

Returns:

Type	Description
tuple[object, ...]	A new tuple.

Raises:

Type	Description
TypeError	If matches is not callable.

Source code in tempestweb/query/optimistic.py

def replace_where(
    rows: Iterable[object],
    matches: object,
    row: object,
) -> tuple[object, ...]:
    """Replace every row a predicate accepts.

    For the cases ``upsert_by_id`` does not cover — a composite key, a row
    identified by something other than a field.

    Args:
        rows: The rows currently cached.
        matches: A callable answering whether a row should be replaced.
        row: The replacement.

    Returns:
        A new tuple.

    Raises:
        TypeError: If ``matches`` is not callable.
    """
    if not callable(matches):
        raise TypeError("replace_where needs a callable predicate")
    return tuple(row if matches(existing) else existing for existing in rows)

upsert_by_id

upsert_by_id(rows: Iterable[object], row: object, *, id_field: str = ID_FIELD) -> tuple[object, ...]

Replace a row with the same id, or append it when there is none.

Parameters:

Name	Type	Description	Default
rows	Iterable[object]	The rows currently cached.	required
row	object	The row to put in.	required
id_field	str	The field the rows are identified by.	ID_FIELD

Returns:

Type	Description
object	A new tuple. The replaced row keeps its position; a new row goes last.
...	When row carries no id, it is appended — an unidentified row cannot
tuple[object, ...]	replace anything, and dropping it silently would lose the user's edit.

Source code in tempestweb/query/optimistic.py

def upsert_by_id(
    rows: Iterable[object],
    row: object,
    *,
    id_field: str = ID_FIELD,
) -> tuple[object, ...]:
    """Replace a row with the same id, or append it when there is none.

    Args:
        rows: The rows currently cached.
        row: The row to put in.
        id_field: The field the rows are identified by.

    Returns:
        A new tuple. The replaced row keeps its position; a new row goes last.
        When ``row`` carries no id, it is appended — an unidentified row cannot
        replace anything, and dropping it silently would lose the user's edit.
    """
    identifier = _identifier(row, id_field)
    if identifier is None:
        return (*rows, row)

    result: list[object] = []
    replaced = False
    for existing in rows:
        found = _identifier(existing, id_field)
        if not replaced and found is not None and found == identifier:
            result.append(row)
            replaced = True
        else:
            result.append(existing)
    if not replaced:
        result.append(row)
    return tuple(result)

cursor_page

cursor_page(payload: Mapping[str, object], *, page_keys: PageKeys = DEFAULT_PAGE_KEYS) -> CursorPage

Read a cursor-paginated payload.

Parameters:

Name	Type	Description	Default
payload	Mapping[str, object]	The decoded JSON body.	required
page_keys	PageKeys	Which keys to read, for a server naming them differently.	DEFAULT_PAGE_KEYS

Returns:

Name	Type	Description
The	CursorPage	class:CursorPage.

Source code in tempestweb/query/pagination.py

def cursor_page(
    payload: Mapping[str, object],
    *,
    page_keys: PageKeys = DEFAULT_PAGE_KEYS,
) -> CursorPage:
    """Read a cursor-paginated payload.

    Args:
        payload: The decoded JSON body.
        page_keys: Which keys to read, for a server naming them differently.

    Returns:
        The :class:`CursorPage`.
    """
    cursor = payload.get(page_keys.cursor)
    return CursorPage(
        items=_items(payload.get(page_keys.items)),
        next_cursor=cursor if isinstance(cursor, str) and cursor else None,
    )

empty_cursor_page

empty_cursor_page() -> CursorPage

A cursor page with no rows, for the state before the first answer.

Returns:

Type	Description
CursorPage	The empty page.

Source code in tempestweb/query/pagination.py

def empty_cursor_page() -> CursorPage:
    """A cursor page with no rows, for the state before the first answer.

    Returns:
        The empty page.
    """
    return CursorPage()

empty_offset_page

empty_offset_page(*, page: int = 1, page_size: int = 0) -> OffsetPage

An offset page with no rows, for the state before the first answer.

Parameters:

Name	Type	Description	Default
page	int	The page number the screen is on.	1
page_size	int	The page size the screen asked for.	0

Returns:

Type	Description
OffsetPage	The empty page. Preferred over None in state: a view reading
OffsetPage	page.items never has to check first.

Source code in tempestweb/query/pagination.py

def empty_offset_page(*, page: int = 1, page_size: int = 0) -> OffsetPage:
    """An offset page with no rows, for the state before the first answer.

    Args:
        page: The page number the screen is on.
        page_size: The page size the screen asked for.

    Returns:
        The empty page. Preferred over ``None`` in state: a view reading
        ``page.items`` never has to check first.
    """
    return OffsetPage(page=page, page_size=page_size)

is_cursor_page

is_cursor_page(payload: object, *, page_keys: PageKeys = DEFAULT_PAGE_KEYS) -> bool

Report whether a payload looks cursor-paginated.

Parameters:

Name	Type	Description	Default
payload	object	The decoded JSON body.	required
page_keys	PageKeys	Which keys identify the shape.	DEFAULT_PAGE_KEYS

Returns:

Type	Description
bool	Whether it carries both the rows key and the cursor key. The cursor key
bool	being present with a null value still counts — that is how the last
bool	page announces itself.

Source code in tempestweb/query/pagination.py

def is_cursor_page(
    payload: object,
    *,
    page_keys: PageKeys = DEFAULT_PAGE_KEYS,
) -> bool:
    """Report whether a payload looks cursor-paginated.

    Args:
        payload: The decoded JSON body.
        page_keys: Which keys identify the shape.

    Returns:
        Whether it carries both the rows key and the cursor key. The cursor key
        being present with a ``null`` value still counts — that is how the last
        page announces itself.
    """
    return (
        isinstance(payload, Mapping)
        and page_keys.items in payload
        and page_keys.cursor in payload
    )

is_offset_page

is_offset_page(payload: object, *, page_keys: PageKeys = DEFAULT_PAGE_KEYS) -> bool

Report whether a payload looks offset-paginated.

Parameters:

Name	Type	Description	Default
payload	object	The decoded JSON body.	required
page_keys	PageKeys	Which keys identify the shape.	DEFAULT_PAGE_KEYS

Returns:

Type	Description
bool	Whether it carries both the rows key and the total key.

Source code in tempestweb/query/pagination.py

def is_offset_page(
    payload: object,
    *,
    page_keys: PageKeys = DEFAULT_PAGE_KEYS,
) -> bool:
    """Report whether a payload looks offset-paginated.

    Args:
        payload: The decoded JSON body.
        page_keys: Which keys identify the shape.

    Returns:
        Whether it carries both the rows key and the total key.
    """
    return (
        isinstance(payload, Mapping)
        and page_keys.items in payload
        and page_keys.total in payload
    )

offset_page

offset_page(payload: Mapping[str, object], *, page_keys: PageKeys = DEFAULT_PAGE_KEYS) -> OffsetPage

Read an offset-paginated payload.

Parameters:

Name	Type	Description	Default
payload	Mapping[str, object]	The decoded JSON body.	required
page_keys	PageKeys	Which keys to read, for a server naming them differently.	DEFAULT_PAGE_KEYS

Returns:

Name	Type	Description
The	OffsetPage	class:OffsetPage. Missing or wrongly-typed keys fall back to the
	OffsetPage	dataclass defaults rather than raising — a listing that renders empty is
	OffsetPage	recoverable; a screen that raised on the way to rendering is not.

Source code in tempestweb/query/pagination.py

def offset_page(
    payload: Mapping[str, object],
    *,
    page_keys: PageKeys = DEFAULT_PAGE_KEYS,
) -> OffsetPage:
    """Read an offset-paginated payload.

    Args:
        payload: The decoded JSON body.
        page_keys: Which keys to read, for a server naming them differently.

    Returns:
        The :class:`OffsetPage`. Missing or wrongly-typed keys fall back to the
        dataclass defaults rather than raising — a listing that renders empty is
        recoverable; a screen that raised on the way to rendering is not.
    """
    return OffsetPage(
        items=_items(payload.get(page_keys.items)),
        total=_int(payload.get(page_keys.total), 0),
        page=_int(payload.get(page_keys.page), 1),
        page_size=_int(payload.get(page_keys.page_size), 0),
    )

persist async

persist(cache: QueryCache, storage: QueryStorage, *, prefix: str = STORAGE_PREFIX) -> PersistResult

Write every JSON-able cache entry to the store.

Parameters:

Name	Type	Description	Default
cache	QueryCache	The cache to write out.	required
storage	QueryStorage	Where to write — native.storage, or a fake in a test.	required
prefix	str	The storage-key prefix, so :func:restore finds these and nothing else.	STORAGE_PREFIX

Returns:

Name	Type	Description
A	PersistResult	class:PersistResult counting what was written and what was skipped.

Source code in tempestweb/query/persistence.py

async def persist(
    cache: QueryCache,
    storage: QueryStorage,
    *,
    prefix: str = STORAGE_PREFIX,
) -> PersistResult:
    """Write every JSON-able cache entry to the store.

    Args:
        cache: The cache to write out.
        storage: Where to write — ``native.storage``, or a fake in a test.
        prefix: The storage-key prefix, so :func:`restore` finds these and
            nothing else.

    Returns:
        A :class:`PersistResult` counting what was written and what was skipped.
    """
    written = 0
    skipped = 0
    for key in cache.keys:
        value = cache.get(key)
        try:
            payload = json.dumps({"key": list(key), "value": value})
        except (TypeError, ValueError):
            skipped += 1
            continue
        await storage.put(_name(prefix, key), payload)
        written += 1
    return PersistResult(written=written, skipped=skipped)

restore async

restore(cache: QueryCache, storage: QueryStorage, *, prefix: str = STORAGE_PREFIX) -> RestoreResult

Read persisted entries back into a cache.

Entries land fresh, stamped with the cache's clock at restore time. Reviving them stale would send a boot screen straight back to the network, which is the thing persisting was supposed to avoid; a screen that wants the network anyway calls :meth:~tempestweb.query.QueryCache.invalidate right after.

Parameters:

Name	Type	Description	Default
cache	QueryCache	The cache to fill.	required
storage	QueryStorage	Where to read from.	required
prefix	str	The storage-key prefix written by :func:persist.	STORAGE_PREFIX

Returns:

Name	Type	Description
A	RestoreResult	class:RestoreResult counting what came back and what was thrown
	RestoreResult	away. A record that no longer parses is deleted rather than left to
	RestoreResult	fail on every boot — the shape of a cached value changes when the app
	RestoreResult	does, and a store that cannot be read is a store that must be cleared.

Source code in tempestweb/query/persistence.py

async def restore(
    cache: QueryCache,
    storage: QueryStorage,
    *,
    prefix: str = STORAGE_PREFIX,
) -> RestoreResult:
    """Read persisted entries back into a cache.

    Entries land **fresh**, stamped with the cache's clock at restore time.
    Reviving them stale would send a boot screen straight back to the network,
    which is the thing persisting was supposed to avoid; a screen that wants the
    network anyway calls :meth:`~tempestweb.query.QueryCache.invalidate` right
    after.

    Args:
        cache: The cache to fill.
        storage: Where to read from.
        prefix: The storage-key prefix written by :func:`persist`.

    Returns:
        A :class:`RestoreResult` counting what came back and what was thrown
        away. A record that no longer parses is **deleted** rather than left to
        fail on every boot — the shape of a cached value changes when the app
        does, and a store that cannot be read is a store that must be cleared.
    """
    restored = 0
    discarded = 0
    for name in await storage.list_keys():
        if not name.startswith(prefix):
            continue
        key = await _read(storage, name, cache)
        if key is None:
            await storage.remove(name)
            discarded += 1
        else:
            restored += 1
    return RestoreResult(restored=restored, discarded=discarded)

should_retry_query

should_retry_query(attempt: int, status: int | None) -> bool

Report whether a failed read is worth attempting again.

A read — this is the query side. Retrying a GET is free; the write side is :func:tempestweb.native.http.request, which retries only what carries an idempotency key.

Parameters:

Name	Type	Description	Default
attempt	int	How many attempts have already happened, the failed one included. The first failure passes 1.	required
status	int | None	The HTTP status the server answered, or None for a network-level failure (no response at all).	required

Returns:

Type	Description
bool	Whether to try again. A network-level failure is retried; a status the
bool	server chose is retried only when it means "later" — a 404 or a 403 will
bool	answer the same way forever, and retrying it just makes the user wait
bool	three times as long for the same error.

Source code in tempestweb/query/policy.py

def should_retry_query(attempt: int, status: int | None) -> bool:
    """Report whether a failed read is worth attempting again.

    A **read** — this is the query side. Retrying a GET is free; the write side
    is :func:`tempestweb.native.http.request`, which retries only what carries
    an idempotency key.

    Args:
        attempt: How many attempts have already happened, the failed one
            included. The first failure passes ``1``.
        status: The HTTP status the server answered, or ``None`` for a
            network-level failure (no response at all).

    Returns:
        Whether to try again. A network-level failure is retried; a status the
        server chose is retried only when it means "later" — a 404 or a 403 will
        answer the same way forever, and retrying it just makes the user wait
        three times as long for the same error.
    """
    if attempt >= MAX_QUERY_ATTEMPTS:
        return False
    if status is None:
        return True
    return status in RETRYABLE_STATUS