Pagination¶

This guide explains the platform's limit / offset pagination protocol, how to walk one page or all pages, and how to use iterate_all() for streaming iteration across large result sets.

What "pagination" means in kneo-client¶

Every list endpoint the platform exposes — runs, audit events, human tasks, environment policies, traces, checkpoints — uses the same limit / offset pagination protocol. The client surfaces that protocol through:

Uniform keyword arguments on every list method (limit, offset, sort_by, sort_order).
A PaginatedResult[T] wrapper class in kneo_client.core.pagination for typed page-by-page handling.
An iterate_all() async iterator that walks pages transparently given a fetch_page(limit, offset) callable.

The protocol¶

Platform list endpoints accept:

Query parameter	Type	Meaning
`limit`	int (1–1000)	Page size. Default 100.
`offset`	int	Skip this many items.
`sort_by`	str	Field to sort by. Defaults vary per endpoint (`updated_at` for runs, `timestamp` for audit, etc.).
`sort_order`	`"asc"` or `"desc"`	Sort direction. Defaults vary per endpoint.

Responses include:

Response field	Meaning
`count`	Items on this page (`len(items)`).
`total`	Total items across all pages, as reported by the server.
`limit`	The page size the server actually applied (may equal the requested value).
`offset`	The offset of the first item on this page.
`sort_by`, `sort_order`	Echo of the sort parameters in effect.
Items array	Endpoint-specific name (`runs`, `events`, `tasks`, `checkpoints`, …).

Concrete shapes vary per endpoint — the items array is named after the resource (page.runs, page.events, etc.). See the API Reference for each endpoint's exact response model.

Walking one page¶

Every platform list method exposes the same kwargs:

page = await client.platform.runs.list(
    status="running",
    limit=50,
    offset=0,
    sort_by="updated_at",
    sort_order="desc",
)
print(f"got {page.count} of {page.total} runs")
for run in page.runs:
    print(run)

None for any keyword argument means omit — the platform's default kicks in. The client passes through whatever the server returns; it doesn't second-guess the page size or impose its own defaults beyond passing through your input.

Walking all pages manually¶

The straightforward pattern works for any list endpoint:

async def all_runs(client, **filters):
    offset = 0
    page_size = 200
    while True:
        page = await client.platform.runs.list(
            limit=page_size, offset=offset, **filters
        )
        for run in page.runs:
            yield run
        if page.count < page_size:
            break
        offset += page.count

This pattern uses count < page_size as the end-of-iteration signal, which works whether or not the server returns a total. It's also robust to a server that returns fewer items than requested (e.g., quota / rate limiting on a per-page basis).

examples/03_paginate_audit.py does exactly this for audit events.

The `iterate_all()` helper¶

kneo_client.core.pagination.iterate_all() is an async iterator that walks pages given a fetch_page(limit, offset) callable returning a PaginatedResult:

from kneo_client.core.pagination import PaginatedResult, iterate_all

async for item in iterate_all(fetch_page, page_size=200):
    process(item)

The platform list methods don't yet return PaginatedResult directly — that integration is a known follow-up — so today you adapt the call site:

from kneo_client.core.pagination import PaginatedResult, iterate_all

async def all_audit_events(client, **filters):
    async def fetch_page(limit: int, offset: int) -> PaginatedResult:
        resp = await client.platform.audit.list(limit=limit, offset=offset, **filters)
        return PaginatedResult(
            items=resp.events,
            total=getattr(resp, "total", 0) or 0,
            limit=limit,
            offset=offset,
        )

    async for event in iterate_all(fetch_page, page_size=200):
        yield event

# Use it:
async for event in all_audit_events(client, event_type="run.created"):
    print(event)

iterate_all() does three things on your behalf:

Clamps page_size to MAX_PAGE_SIZE = 1000 — the platform's hard upper bound. Asking for more than 1000 silently downsizes.
Walks offset automatically — each page's offset becomes the next page's starting position.
Stops when has_more is false — the PaginatedResult.has_more property is offset + count < total, which works whenever the response includes a total. For responses that don't, build the PaginatedResult with total=count and the iteration stops after one page.

`PaginatedResult[T]` — the typed wrapper¶

@dataclass(frozen=True)
class PaginatedResult(Generic[T]):
    items: list[T]
    total: int
    limit: int
    offset: int
    sort_by: str | None = None
    sort_order: str | None = None

    @property
    def count(self) -> int:
        return len(self.items)

    @property
    def has_more(self) -> bool:
        return self.offset + self.count < self.total

Use it when building your own page-walker (as above) or when you want a uniform shape across endpoints regardless of what the underlying response model is named.

Choosing a page size¶

A few rules of thumb:

100–200 for most interactive flows. Small enough to keep latency snappy; large enough to amortize request overhead.
500–1000 for back-end export jobs that walk the whole list and don't care about first-byte latency. Larger pages reduce request count.
< 50 if downstream processing per item is slow and you want to start producing output sooner.

Larger pages reduce per-request overhead but increase the cost of a failed page — everything in flight has to be re-fetched. If the platform deployment you're talking to is on a flaky network, prefer smaller pages.

The maximum is MAX_PAGE_SIZE = 1000 (the platform's enforced upper bound). Anything larger is silently clamped by iterate_all() and by the platform itself.

Pagination + filters¶

All list methods accept resource-specific filter kwargs alongside the pagination args. Common patterns:

# Just the failures
failed_runs = await client.platform.runs.list(status="failed")

# Just audit events for one run
audit_for_run = await client.platform.audit.list(run_id="r1")

# Just pending human tasks
pending = await client.platform.human_tasks.list(status="pending")

Filters compose with pagination — runs.list(status="failed", limit=50) filters then paginates the filtered result.

What's not on the roadmap¶

Auto-pagination at the adapter layer (e.g., client.platform.runs.list_all() returning an iterator) is not planned. The current shape — list methods return one page; iterate_all() walks pages explicitly — keeps the per-call cost transparent and lets callers decide when to stop. Auto-walking can mask runaway iteration if a filter accidentally matches a huge result set.

If you want a one-liner for the common case, write a small helper in your application like all_runs() above. The pattern is identical for every endpoint.