Pagination

List operations in the Python SDK are paginated at the body level: every list input carries an optional limit and offset, and you advance through results by incrementing offset yourself.

There is no automatic page iterator. Each list operation is a single call that returns one page, so you write a short loop that requests pages until a short page comes back. This guide shows that loop with the recommended fluent Geopera client and with the low-level operation modules, covers the defaults and edge cases, and explains how offset paging differs from the cursor-based paging that catalog search uses. For the protocol-level rules these fields obey — the meaning of limit/offset, maximum page sizes, and how totals are reported — see API pagination.

How it works

Every Geopera operation is invoked the same way — POST /v1/op/{operation_id} with a JSON body — so a list operation’s paging controls live in its input body, not in query parameters. For example, the items.list body (ItemsListInput) exposes:

Field	Type	Default	Meaning
`project_id`	`str`	required	The project whose items to list
`collection_id`	`str`	unset	Restrict to one collection
`uncollected`	`bool`	`False`	Only items not yet assigned to a collection
`limit`	`int`	`100`	Maximum number of rows to return in this page
`offset`	`int`	`0`	Number of rows to skip before this page

limit and offset are optional. Omit them and you get the first 100 rows. The SDK only serializes a field into the request body when you set it, so leaving limit/offset unset sends the server defaults.

Different list operations carry different defaults and ceilings. Most list inputs default limit to 100; a few — for example alerts.events.list (AlertEventsListInput) — default to 50, and some enforce a server-side maximum. Always read the operation’s input model rather than assuming 100. See API pagination for the canonical limits and Models for how input models are constructed.

A single page with the fluent client

The recommended surface is the fluent Geopera client. Attribute access descends the resource path and calling the node invokes the operation: client.items.list({...}) calls the items.list operation. The body is a plain dict (converted to the typed input model for you) or the input model instance. The call returns the parsed 200 body — a dict for list operations — or a typed Problem on an error status.

python

from geopera import Geopera

# The token is a Bearer token: either a session token from a browser sign-in
# or a minted API key (prefix gpra_). The API key IS the bearer token.
client = Geopera(token="gpra_...")

page = client.items.list(
    {
        "project_id": "proj_123",
        "limit": 100,
        "offset": 0,
    }
)

for item in page["items"]:
    print(item["id"])

Geopera targets https://api.geopera.com by default; pass base_url= to point at another environment. Extra keyword arguments (such as timeout) flow through to the underlying client. See Configuration for the full set, and Authentication for the gpra_ token and the session-token alternative.

If you need the HTTP status and headers alongside the parsed body, pass detailed=True to get the typed Response:

python

resp = client.items.list({"project_id": "proj_123"}, detailed=True)
print(resp.status_code)        # HTTPStatus.OK
page = resp.parsed             # the dict (or Problem)

Paging through every result (fluent)

Because there is no auto-iterator, loop manually: request a page, yield its rows, then advance offset by limit. Stop when a page comes back shorter than limit — that short (or empty) page is the last one.

python

from collections.abc import Iterator
from typing import Any

from geopera import Geopera
from geopera.models.problem import Problem


def iter_items(
    client: Geopera,
    project_id: str,
    page_size: int = 100,
) -> Iterator[dict[str, Any]]:
    """Yield every item in a project, one page at a time."""
    offset = 0
    while True:
        page = client.items.list(
            {
                "project_id": project_id,
                "limit": page_size,
                "offset": offset,
            }
        )

        # Surface API problems instead of silently swallowing them.
        if isinstance(page, Problem):
            raise RuntimeError(f"{page.title}: {page.detail}")

        rows = page["items"]
        yield from rows

        # A short page means we have reached the end.
        if len(rows) < page_size:
            return

        offset += page_size

Use it like any generator:

python

client = Geopera(token="gpra_...")

for item in iter_items(client, project_id="proj_123"):
    print(item["id"], item.get("datetime"))

The stopping condition is len(rows) < page_size, not len(rows) == 0. A full final page that lands exactly on a multiple of page_size triggers one extra request that returns an empty page — which then ends the loop. Checking for a short page avoids depending on a total field that not every list operation returns. See Errors for how Problem is structured.

The low-level escape hatch

The fluent client calls the generated operation modules under geopera.api.operations, and you can call them directly when you need the async variants, the typed Response object, or per-call control that the fluent surface does not expose. Each module exposes sync, sync_detailed, asyncio, and asyncio_detailed. You construct the typed input model and pass an AuthenticatedClient.

python

from geopera import AuthenticatedClient
from geopera.api.operations import items_list
from geopera.models.items_list_input import ItemsListInput

client = AuthenticatedClient(
    base_url="https://api.geopera.com",
    token="gpra_...",
)

page = items_list.sync(
    client=client,
    body=ItemsListInput(
        project_id="proj_123",
        limit=100,
        offset=0,
    ),
)

for item in page["items"]:
    print(item["id"])

sync returns the parsed body of a 200 response (a dict for list operations), or a typed Problem / HTTPValidationError on an error status.
sync_detailed returns the typed Response, giving you .status_code, .headers, .content, and .parsed.

If you already have a Geopera instance, client.client exposes the same underlying AuthenticatedClient, so you can mix the two surfaces without constructing a second client:

python

from geopera import Geopera
from geopera.api.operations import items_list
from geopera.models.items_list_input import ItemsListInput

fluent = Geopera(token="gpra_...")

# Reuse the fluent client's connection for a low-level call.
page = items_list.sync(
    client=fluent.client,
    body=ItemsListInput(project_id="proj_123", limit=100, offset=0),
)

Paging with the low-level module

The same offset loop works against items_list.sync. The only differences are the explicit input model and passing client= by keyword.

python

from collections.abc import Iterator
from typing import Any

from geopera import AuthenticatedClient
from geopera.api.operations import items_list
from geopera.models.items_list_input import ItemsListInput
from geopera.models.problem import Problem


def iter_items_lowlevel(
    client: AuthenticatedClient,
    project_id: str,
    page_size: int = 100,
) -> Iterator[dict[str, Any]]:
    offset = 0
    while True:
        page = items_list.sync(
            client=client,
            body=ItemsListInput(
                project_id=project_id,
                limit=page_size,
                offset=offset,
            ),
        )
        if isinstance(page, Problem):
            raise RuntimeError(f"{page.title}: {page.detail}")

        rows = page["items"]
        yield from rows

        if len(rows) < page_size:
            return
        offset += page_size

Async paging

The async functions mirror the sync ones. Use asyncio (parsed body) or asyncio_detailed (typed Response) inside an async def, and advance offset exactly as above. See Async usage for the full async client lifecycle.

python

import asyncio
from collections.abc import AsyncIterator
from typing import Any

from geopera import AuthenticatedClient
from geopera.api.operations import items_list
from geopera.models.items_list_input import ItemsListInput
from geopera.models.problem import Problem


async def aiter_items(
    client: AuthenticatedClient,
    project_id: str,
    page_size: int = 100,
) -> AsyncIterator[dict[str, Any]]:
    offset = 0
    while True:
        page = await items_list.asyncio(
            client=client,
            body=ItemsListInput(
                project_id=project_id,
                limit=page_size,
                offset=offset,
            ),
        )
        if isinstance(page, Problem):
            raise RuntimeError(f"{page.title}: {page.detail}")

        rows = page["items"]
        for row in rows:
            yield row

        if len(rows) < page_size:
            return
        offset += page_size


async def main() -> None:
    async with AuthenticatedClient(
        base_url="https://api.geopera.com",
        token="gpra_...",
    ) as client:
        async for item in aiter_items(client, project_id="proj_123"):
            print(item["id"])


asyncio.run(main())

Worked example: collect every item into a list

A complete fluent script that exhausts a list operation and reports the total it gathered.

python

from geopera import Geopera
from geopera.models.problem import Problem

PAGE_SIZE = 200  # larger pages = fewer round trips, within the operation's ceiling

client = Geopera(token="gpra_...")

all_items: list[dict] = []
offset = 0

while True:
    page = client.items.list(
        {
            "project_id": "proj_123",
            "limit": PAGE_SIZE,
            "offset": offset,
        }
    )
    if isinstance(page, Problem):
        raise RuntimeError(f"{page.title}: {page.detail}")

    batch = page["items"]
    all_items.extend(batch)

    if len(batch) < PAGE_SIZE:
        break
    offset += PAGE_SIZE

print(f"Collected {len(all_items)} items")

Paging a different list operation

The loop is identical for every *.list operation — only the operation path, the required body fields, and the response key change. For example, alerts.events.list lists alert events, defaults limit to 50, and returns its rows under the events key. Read the response key from the operation’s output rather than hard-coding items.

python

from geopera import Geopera
from geopera.models.problem import Problem

client = Geopera(token="gpra_...")

offset = 0
page_size = 50  # alerts.events.list defaults to 50

while True:
    page = client.alerts.events.list(
        {
            "rule_id": "rule_abc",
            "limit": page_size,
            "offset": offset,
        }
    )
    if isinstance(page, Problem):
        raise RuntimeError(f"{page.title}: {page.detail}")

    rows = page["events"]
    for event in rows:
        print(event["id"])

    if len(rows) < page_size:
        break
    offset += page_size

Cursor paging vs. offset paging

Not every list-shaped operation uses offset. Catalog search (catalog.search) follows the STAC convention: its input has a limit and an opaque next_ token instead of an offset. You advance by passing the next token the previous response handed back, and you stop when no next token comes back. This is a separate paging style — do not mix offset into a catalog.search call.

python

from geopera import Geopera

client = Geopera(token="gpra_...")

body = {
    "host_name": "earthsearch-aws",
    "collections": ["sentinel-2-l2a"],
    "limit": 100,
}

while True:
    page = client.catalog.search(body)
    for feature in page["features"]:
        print(feature["id"])

    # Stop when the response carries no continuation token.
    next_token = page.get("next")
    if not next_token:
        break
    body = {**body, "next_": next_token}

Use offset paging for the project-scoped *.list operations and next_ cursor paging for catalog.search. See API pagination for which operations use which convention.

Gotchas

Pick a page_size within the operation’s ceiling. Some list inputs cap limit. Sending a limit above the maximum returns an RFC 9457 problem on a 422 status, surfaced as an HTTPValidationError (or, from sync_detailed, a Response whose .parsed is an HTTPValidationError). See Errors.
offset paging is not snapshot-isolated. If rows are inserted or deleted while you page, fixed offsets can skip or repeat rows. For volatile collections, prefer narrowing the query (for example by datetime or collection_id) over deep offsets, or use cursor paging where it is available.
Don’t loop on total. Not every list operation returns a total count, and result keys differ per operation (items, events, collections, orders, features, …). Reading the response key and checking for a short page is the portable stop condition.
dict body vs. typed model. The fluent client accepts both: client.items.list({...}) converts the dict to ItemsListInput for you, while the low-level modules require the model. Either way, only fields you set are serialized — unset limit/offset fall back to the server defaults.
Check for Problem before indexing. A 200 response is a dict; an error status is a Problem model. Indexing a Problem with page["items"] raises, so branch on isinstance(page, Problem) first.
Same field meaning, different defaults. limit and offset always mean the same thing, but the default page size and maximum vary per operation. The defaults shown here (limit=100, offset=0) match most list inputs; confirm against the specific input model.

API pagination — protocol-level limit/offset and cursor conventions and limits
Operations — calling operations with the fluent and low-level surfaces
Models — constructing typed input models like ItemsListInput
Async usage — the async client lifecycle and asyncio functions
Errors — how 422 validation problems and Problem are returned
Authentication — Bearer tokens (gpra_ keys or session tokens)

Pagination

How it works

A single page with the fluent client

Paging through every result (fluent)

The low-level escape hatch

Paging with the low-level module

Async paging

Worked example: collect every item into a list

Paging a different list operation

Cursor paging vs. offset paging

Gotchas

Related