> For the complete documentation index, see [llms.txt](https://docs.pullbay.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.pullbay.com/documentation/concepts/pagination.md).

# Pagination

## Overview

Pagination is essential for efficiently retrieving large datasets from APIs. Pullbay offers two pagination models to give you complete flexibility: **Standard pagination** for granular control and **Managed pagination** for simplicity. This guide explains both approaches, when to use each, and best practices for production workloads.

## Why Pagination Matters

Large datasets present three critical challenges:

1. **Performance**: Retrieving millions of records in a single request strains both client and server resources, causing timeouts and memory exhaustion.
2. **Network Efficiency**: Breaking data into smaller pages reduces bandwidth usage and allows incremental processing.
3. **Credit Control**: Pullbay charges per request. Pagination lets you fetch only what you need, controlling costs and avoiding wasted credits on unused data.

Without pagination, a single request for 100,000 items would consume massive credits and likely fail. Pagination lets you process data incrementally, caching and stopping when you have enough results.

## Standard Pagination (You Control the Flow)

Standard pagination gives you complete control: you request pages one at a time using the `cursor` parameter, process each, and decide whether to continue. This model is ideal when you need only a subset of data or want to implement custom logic.

### How It Works

1. Make an initial request to the endpoint without a cursor
2. Receive the first page of results plus a `cursor` token in the `pagination` block
3. Request the next page by passing `cursor=<token>`
4. Repeat until `hasNextPage` is `false`

### Standard Pagination Response Format

Every paginated response includes a `pagination` block. Fields that are not relevant to the current endpoint are returned as `null`.

```json
{
  "requestId": "3b8dcb68-1f8c-4a7b-b9e7-63a7b9986f24",
  "status": 200,
  "message": "OK",
  "success": true,
  "data": [
    { "id": "rev_001", "text": "Great service!", "rating": 5 },
    { "id": "rev_002", "text": "Good experience", "rating": 4 }
  ],
  "pagination": {
    "page": null,
    "hasNextPage": true,
    "cursor": "eyJpZCI6InJldl8wMDIifQ==",
    "offset": null
  },
  "pricing": { "creditsCharged": 52 }
}
```

When `hasNextPage` is `false`, no more pages exist — stop paginating.

### Example: First Request (No Cursor)

```bash
curl -X GET "https://api.pullbay.com/appstore/reviews" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -G \
  --data-urlencode "appId=389801252" \
  --data-urlencode "country=us"
```

### Example: Second Request (With Cursor)

```bash
curl -X GET "https://api.pullbay.com/appstore/reviews" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -G \
  --data-urlencode "appId=389801252" \
  --data-urlencode "country=us" \
  --data-urlencode "cursor=eyJpZCI6InJldl8wMDIifQ=="
```

When the final page is returned, `hasNextPage` is `false` and `cursor` is `null`:

```json
{
  "requestId": "...",
  "status": 200,
  "message": "OK",
  "success": true,
  "data": [{ "id": "rev_951", "text": "Excellent work", "rating": 5 }],
  "pagination": {
    "page": null,
    "hasNextPage": false,
    "cursor": null,
    "offset": null
  },
  "pricing": { "creditsCharged": 2 }
}
```

### Page-Based Pagination

Some endpoints support `page` instead of `cursor`. Pass `page=<number>` (1-indexed) and the response will mirror it in `pagination.page`:

```bash
curl -X GET "https://api.pullbay.com/google-search/search" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -G \
  --data-urlencode "query=pullbay api" \
  --data-urlencode "page=2"
```

Response:

```json
{
  "requestId": "...",
  "status": 200,
  "success": true,
  "data": [...],
  "pagination": {
    "page": 2,
    "hasNextPage": true,
    "cursor": null,
    "offset": null
  },
  "pricing": { "creditsCharged": 25 }
}
```

### Complete Python Implementation

```python
import requests
import time

class PullbayPaginator:
    def __init__(self, api_key, base_url="https://api.pullbay.com"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
        })

    def paginate(self, endpoint, params=None, max_results=None):
        """
        Generator that yields individual items across all cursor-paginated pages.

        Args:
            endpoint: API endpoint e.g. 'appstore/reviews'
            params: Query parameters dict
            max_results: Stop after this many results (None = all)

        Yields:
            Individual items from all pages
        """
        if params is None:
            params = {}

        cursor = None
        items_fetched = 0

        while True:
            req_params = dict(params)
            if cursor:
                req_params["cursor"] = cursor

            try:
                response = self.session.get(
                    f"{self.base_url}/{endpoint}",
                    params=req_params,
                    timeout=30,
                )
                response.raise_for_status()
            except requests.exceptions.RequestException as e:
                print(f"Request failed: {e}")
                raise

            body = response.json()
            items = body.get("data", [])
            pagination = body.get("pagination", {})
            request_id = body.get("requestId")
            credits = body.get("pricing", {}).get("creditsCharged", 0)
            print(f"Page fetched — requestId: {request_id}, creditsCharged: {credits}")

            for item in items:
                yield item
                items_fetched += 1
                if max_results and items_fetched >= max_results:
                    return

            if not pagination.get("hasNextPage"):
                break

            cursor = pagination.get("cursor")
            if not cursor:
                break

            time.sleep(0.1)  # Respect rate limits

# Usage
paginator = PullbayPaginator(api_key="your_api_key")

all_reviews = []
for review in paginator.paginate(
    endpoint="appstore/reviews",
    params={"appId": "389801252", "country": "us"},
    max_results=100,
):
    all_reviews.append(review)

print(f"Total reviews fetched: {len(all_reviews)}")
```

### Tips for Large Datasets with Standard Pagination

**Stream results** — don't wait for all data before processing. Use generators (as shown above) to handle results incrementally:

```python
for review in paginator.paginate("appstore/reviews", params={"appId": "389801252"}):
    process_review(review)  # Handle immediately
```

**Checkpoint and resume** — for long-running jobs, save the last cursor. If interrupted, resume from it instead of restarting:

```python
import json, os

checkpoint_file = "pagination_checkpoint.json"

last_cursor = None
if os.path.exists(checkpoint_file):
    with open(checkpoint_file) as f:
        last_cursor = json.load(f).get("cursor")

if last_cursor:
    params = {"appId": "389801252", "cursor": last_cursor}
else:
    params = {"appId": "389801252"}

for i, review in enumerate(paginator.paginate("appstore/reviews", params=params)):
    process_review(review)
    if i % 100 == 0:
        # Checkpointing cursor requires reading it from response — see full implementation
        pass
```

**Parallel fetching** — if you have multiple IDs, fetch them concurrently:

```python
from concurrent.futures import ThreadPoolExecutor

app_ids = ["389801252", "544007664", "310633997"]

def fetch_reviews(app_id):
    return list(paginator.paginate(
        "appstore/reviews",
        params={"appId": app_id},
    ))

with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(fetch_reviews, app_ids))
```

### When to Use Standard Pagination

* **Fine-grained control**: You want to process data as it arrives
* **Subset of data**: You need only the first N results, not everything
* **Cost efficiency**: You want to stop early and avoid fetching unneeded data
* **Real-time processing**: You're streaming results to another system

## Managed Pagination (Pullbay Controls the Flow)

Managed pagination simplifies everything: pass `maxItems` in a single request, and Pullbay fetches all the pages internally and returns them in one response. Perfect for automation tools and batch jobs.

### How It Works

1. Make a single request with `maxItems=<limit>` set to the total number of records you want
2. Pullbay fetches data across as many internal pages as needed
3. Receive a single response containing all matching items up to `maxItems`
4. No cursor handling or loop logic needed on your end

`maxItems` and `cursor` are **mutually exclusive** — you cannot combine them. The maximum value of `maxItems` varies by endpoint (see the endpoint reference for each service).

### Example: Managed Pagination with cURL

```bash
curl -X GET "https://api.pullbay.com/appstore/reviews" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -G \
  --data-urlencode "appId=389801252" \
  --data-urlencode "country=us" \
  --data-urlencode "maxItems=500"
```

**Response (all up to 500 items in one response):**

```json
{
  "requestId": "3b8dcb68-1f8c-4a7b-b9e7-63a7b9986f24",
  "status": 200,
  "message": "OK",
  "success": true,
  "data": [
    { "id": "rev_001", "text": "Great service!", "rating": 5 },
    { "id": "rev_002", "text": "Good experience", "rating": 4 }
  ],
  "pagination": {
    "page": null,
    "hasNextPage": false,
    "cursor": null,
    "offset": null
  },
  "pricing": { "creditsCharged": 503 }
}
```

`hasNextPage` will be `false` because the response already contains everything up to `maxItems`.

### Complete Python Implementation

```python
import requests
from typing import List, Dict, Any

class PullbayManagedPaginator:
    def __init__(self, api_key, base_url="https://api.pullbay.com"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
        })

    def fetch_all(self, endpoint, params=None, max_items=200, timeout=120) -> List[Dict[str, Any]]:
        """
        Fetch up to max_items results in a single request using managed pagination.

        Args:
            endpoint: API endpoint e.g. 'appstore/reviews'
            params: Query parameters dict
            max_items: Maximum items to fetch (bounded by endpoint limit)
            timeout: HTTP timeout in seconds

        Returns:
            List of all returned items
        """
        if params is None:
            params = {}

        params = dict(params)
        params["maxItems"] = max_items

        try:
            response = self.session.get(
                f"{self.base_url}/{endpoint}",
                params=params,
                timeout=timeout,
            )
            response.raise_for_status()
        except requests.exceptions.RequestException as e:
            print(f"Failed to fetch all results: {e}")
            raise

        body = response.json()
        print(f"requestId: {body['requestId']}, creditsCharged: {body['pricing']['creditsCharged']}")
        return body.get("data", [])

# Usage
managed = PullbayManagedPaginator(api_key="your_api_key")

all_reviews = managed.fetch_all(
    endpoint="appstore/reviews",
    params={"appId": "389801252", "country": "us"},
    max_items=500,
)

print(f"Total reviews: {len(all_reviews)}")
for review in all_reviews:
    print(f"Review {review['id']}: rating {review.get('rating')}")
```

### When to Use Managed Pagination

* **Automation tools**: Using n8n, Make.com, Zapier, or similar platforms
* **Complete datasets**: You need all matching records up to the endpoint's limit
* **Simplicity**: You prefer one request over pagination loops
* **Batch jobs**: Running scheduled exports or data synchronization

### Important: Timeout Configuration

Managed pagination fetches all pages internally, which takes longer than a single page. Set your HTTP client timeout accordingly:

* **Standard pagination** (one page): 30 seconds is sufficient
* **Managed pagination** (many internal pages): 60–120 seconds recommended

```python
# Correct: generous timeout for managed
reviews = managed.fetch_all(
    "appstore/reviews",
    params={"appId": "389801252"},
    max_items=500,
    timeout=120,
)

# Risky: default 30s may not be enough for large maxItems
reviews = managed.fetch_all(
    "appstore/reviews",
    params={"appId": "389801252"},
    max_items=500,
    timeout=30,  # Too short for large datasets
)
```

## Choosing the Right Pagination Model

| Scenario                          | Best Choice | Reason                                  |
| --------------------------------- | ----------- | --------------------------------------- |
| Building a custom application     | Standard    | Fine-grained control over data flow     |
| Using n8n, Make.com, Zapier       | Managed     | Platforms prefer single requests        |
| Need first 100 results only       | Standard    | Avoid fetching unnecessary data         |
| Exporting all data monthly        | Managed     | Simplicity; no pagination loop needed   |
| Real-time data processing         | Standard    | Process results as they arrive          |
| Reducing credit usage             | Standard    | Fetch only what you need, stop early    |
| Large dataset (near endpoint max) | Standard    | Managed may timeout for very large sets |
| Simplest possible implementation  | Managed     | Single API call vs. pagination loop     |
| Scheduled batch job               | Managed     | Fire-and-forget simplicity              |

## Performance Comparison

| Metric                  | Standard Pagination             | Managed Pagination                      |
| ----------------------- | ------------------------------- | --------------------------------------- |
| Time to first result    | 1–2 seconds                     | Longer (Pullbay fetches all internally) |
| Total time for all data | Gradual (per page)              | All at once                             |
| Memory usage            | Low (process incrementally)     | Higher (full result in one response)    |
| Code complexity         | Higher (cursor loop)            | Low (single request)                    |
| Partial fetching        | Yes (stop when you have enough) | No (fetches up to `maxItems`)           |
| Best for                | Custom apps, cost optimization  | Automation tools, exports               |
| Failure recovery        | Easy (resume from last cursor)  | Retry entire request                    |

## Credit Costs

**The credit cost is identical for both models when fetching the same number of items.**

Credits are always charged as: `base_cost + (items_returned × 1)`.

If you fetch 500 App Store reviews for one app:

* **Standard**: multiple requests totalling 500 items → same total credits
* **Managed** with `maxItems=500`: one request, 500 items → same total credits

The difference is **control**, not cost:

* **Standard**: You decide how many items to fetch and can stop early
* **Managed**: Pullbay fetches up to `maxItems`; you pay for everything returned

To minimize costs with Managed pagination, use query filters to reduce the dataset before requesting:

```python
# Expensive: fetch all reviews regardless of sort
all_reviews = managed.fetch_all("appstore/reviews", {"appId": "389801252"}, max_items=500)

# Cheaper: fetch only what you need with targeted filters
recent_reviews = managed.fetch_all(
    "appstore/reviews",
    {"appId": "389801252", "sort": "recent"},
    max_items=50,
)
```

## Caching Strategies

Combine pagination with caching to maximize efficiency and reduce credit usage.

### Standard Pagination Caching

Cache results by cursor key to avoid re-fetching pages:

```python
import hashlib, json
from datetime import datetime, timedelta

class CachedPaginator:
    def __init__(self, api_key, cache_ttl=3600):
        self.paginator = PullbayPaginator(api_key)
        self.cache = {}
        self.cache_ttl = cache_ttl

    def paginate_cached(self, endpoint, params):
        cursor = None

        while True:
            cache_key = self._make_key(endpoint, params, cursor)

            if cache_key in self.cache:
                entry = self.cache[cache_key]
                if datetime.now() < entry["expires"]:
                    items, next_cursor = entry["items"], entry["next_cursor"]
                else:
                    del self.cache[cache_key]
                    items, next_cursor = self._fetch_page(endpoint, params, cursor)
                    self._cache(cache_key, items, next_cursor)
            else:
                items, next_cursor = self._fetch_page(endpoint, params, cursor)
                self._cache(cache_key, items, next_cursor)

            yield from items

            if not next_cursor:
                break
            cursor = next_cursor

    def _fetch_page(self, endpoint, params, cursor):
        req_params = dict(params)
        if cursor:
            req_params["cursor"] = cursor
        resp = self.paginator.session.get(
            f"{self.paginator.base_url}/{endpoint}", params=req_params
        )
        resp.raise_for_status()
        body = resp.json()
        pagination = body.get("pagination", {})
        return body.get("data", []), pagination.get("cursor") if pagination.get("hasNextPage") else None

    def _cache(self, key, items, next_cursor):
        self.cache[key] = {
            "items": items,
            "next_cursor": next_cursor,
            "expires": datetime.now() + timedelta(seconds=self.cache_ttl),
        }

    def _make_key(self, endpoint, params, cursor):
        return hashlib.md5(
            json.dumps({"endpoint": endpoint, "params": params, "cursor": cursor}, sort_keys=True).encode()
        ).hexdigest()
```

### Managed Pagination Caching

Cache the entire result as a complete dataset:

```python
class CachedManagedPaginator:
    def __init__(self, api_key, cache_ttl=3600):
        self.managed = PullbayManagedPaginator(api_key)
        self.cache = {}
        self.cache_ttl = cache_ttl

    def fetch_all_cached(self, endpoint, params, max_items=200):
        key = hashlib.md5(
            json.dumps({"endpoint": endpoint, "params": params, "max_items": max_items}, sort_keys=True).encode()
        ).hexdigest()

        if key in self.cache:
            entry = self.cache[key]
            if datetime.now() < entry["expires"]:
                return entry["data"]
            del self.cache[key]

        data = self.managed.fetch_all(endpoint, params, max_items=max_items)
        self.cache[key] = {"data": data, "expires": datetime.now() + timedelta(seconds=self.cache_ttl)}
        return data
```

### Cache TTL Guidelines

| Data Type           | Freshness | Recommended TTL |
| ------------------- | --------- | --------------- |
| Real-time metrics   | Minutes   | 5 minutes       |
| Daily reports       | Hours     | 6 hours         |
| Historical reviews  | Days      | 24 hours        |
| Static app metadata | Weeks     | 7 days          |

## Frequently Asked Questions

<details>

<summary>Are cursor strings safe to store?</summary>

Yes, but treat them as temporary tokens. Cursors are opaque (you cannot modify them), may expire after some time, and are tied to the specific query that generated them. You can safely store them in a database for resuming interrupted jobs.

</details>

<details>

<summary>Do cursors expire?</summary>

Yes. If you resume from an expired cursor, the API will return an error. Handle this by catching the error and restarting from the beginning:

```python
try:
    results = list(paginator.paginate(endpoint, {**params, "cursor": saved_cursor}))
except Exception as e:
    if "cursor" in str(e).lower():
        print("Cursor expired — restarting from beginning")
        results = list(paginator.paginate(endpoint, params))
    else:
        raise
```

</details>

<details>

<summary>What is the difference in credit cost between standard and managed pagination?</summary>

There is no difference for the same number of items fetched. Both charge `base_cost + (items × 1)`. The practical difference is that with Standard you can stop early (and pay less), while Managed always fetches up to `maxItems`.

</details>

<details>

<summary>Can I use managed pagination in n8n?</summary>

Yes. Set `maxItems` as a query parameter in your HTTP Request node. n8n will execute the single request, receive all results in one response, and process them — no pagination loop needed in your workflow.

</details>

<details>

<summary>What happens if `maxItems` exceeds the endpoint's limit?</summary>

The API caps the result at the endpoint's maximum (e.g. 500 for App Store reviews). Requesting `maxItems=9999` on that endpoint returns at most 500 items. Check the endpoint reference for each service's maximum.

</details>

<details>

<summary>Can I combine `cursor` and `maxItems`?</summary>

No. They are mutually exclusive. Providing both will result in a `400 VALIDATION_ERROR`.

</details>

## Summary

**Choose Standard pagination for:**

* Custom control over data flow
* Fetching subsets (first N results)
* Cost optimization (stop early)
* Processing results incrementally

**Choose Managed pagination for:**

* Automation tools (n8n, Make.com, Zapier)
* Complete dataset exports up to the endpoint limit
* Simplicity (one request)
* Batch jobs with sufficient timeout

Both models charge the same per item retrieved. The difference is flexibility and control. Combine either with caching for maximum efficiency and cost savings.