Pagination

Use async paginators to iterate through all results without manual offset management.

Basic Usage

from spatialflow import paginate_geofences

async for geofence in paginate_geofences(
    lambda offset, limit: client.geofences.list(offset=offset, limit=limit)
):
    print(geofence.name)

Paginators by resource

Each list response names its items differently and reports "more" differently, and the endpoints cap limit at different values:

Resource	Items field	How it reports more	Maximum limit
Geofences	geofences	count (this page) and total_count	500, larger values are capped
Workflows	workflows	total, page, page_size	100, larger values are rejected
Webhooks	webhooks	pagination.has_more	200

paginate_geofences works as shown above. Use the generic paginate with items_field for workflows and webhooks. Webhook calls need an access token (access_token=).

from spatialflow import paginate

# Workflows
async for workflow in paginate(
    lambda o, l: client.workflows.list(offset=o, limit=l),
    items_field="workflows",
):
    print(workflow.name)

# Webhooks
async for webhook in paginate(
    lambda o, l: client.webhooks.list(offset=o, limit=l),
    items_field="webhooks",
):
    print(webhook.name)

paginate_workflows and paginate_webhooks read a count field that the workflow and webhook list responses do not have, so they raise AttributeError on the first page. Use paginate instead. paginate_users has no matching list call in the SDK.

Devices, integrations, and storage files are not paginated. These endpoints take no offset. The devices and integrations lists return every result in one response; the storage file list returns at most 1,000 files of a type and doesn't say when it stops short. Call the generated methods through client.raw, for example client.raw.devices.apps_devices_api_list_devices(). The wrappers client.devices.list(), client.integrations.list(), and client.storage.list_files() always send limit and offset, which these endpoints reject, so they fail validation before any request is made. Raw calls return the response models but raise the generated exceptions instead of SpatialFlow errors. Do not use paginate_files for them, because it keeps requesting the same list.

Generic Paginate Function

For custom pagination needs:

from spatialflow import paginate

async for item in paginate(
    lambda o, l: client.geofences.list(offset=o, limit=l),
    limit=50,  # Items per page
    items_field="geofences",
):
    print(item.name)

Parameters

Parameter	Type	Default	Description
fetch_page	Callable	required	Async function that takes (offset, limit) and returns a response
limit	int	100	Number of items per page
items_field	str	"geofences"	Key in response containing the items list (keyword-only)

The loop stops when a page has fewer than limit items, so a total that is an exact multiple of limit costs one extra request that returns an empty page.

AsyncPaginator Class

For more control, use the AsyncPaginator class directly:

from spatialflow import AsyncPaginator

paginator = AsyncPaginator(
    fetch_page=lambda o, l: client.geofences.list(offset=o, limit=l),
    extract_items=lambda r: r.geofences,
    extract_count=lambda r: r.total_count,
    extract_next=lambda r: None,
    limit=50,
)

async for geofence in paginator:
    print(geofence.name)
    if some_condition:
        break  # Early exit is supported

PaginatedResponse

Wrap a paginated response for easier access:

from spatialflow import PaginatedResponse

response = await client.geofences.list(limit=100)
paginated = PaginatedResponse(
    items=response.geofences,
    count=response.total_count,
)

print(f"Got {len(paginated)} of {paginated.count} items")

for geofence in paginated:
    print(geofence.name)

Filtering While Paginating

Combine pagination with filtering:

from spatialflow import paginate_geofences

# Filter by tag
async for geofence in paginate_geofences(
    lambda o, l: client.geofences.list(offset=o, limit=l, tags=["depot"])
):
    print(geofence.name)

Collecting All Results

To collect all results into a list:

from spatialflow import paginate_geofences

all_geofences = [
    geofence
    async for geofence in paginate_geofences(
        lambda o, l: client.geofences.list(offset=o, limit=l)
    )
]

print(f"Total geofences: {len(all_geofences)}")

Performance Tips

• Use a reasonable page size (50-100 items) to balance API calls vs memory
• Break early if you find what you need
• Consider filtering server-side when possible to reduce data transfer