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.

{
  "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)

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)

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:

{
  "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:

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:

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

Complete Python Implementation

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:

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:

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:

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

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):

{
  "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

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

# 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:

# 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:

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:

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

Are cursor strings safe to store?

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.

Do cursors expire?

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:

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

What is the difference in credit cost between standard and managed pagination?

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.

Can I use managed pagination in n8n?

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.

What happens if `maxItems` exceeds the endpoint's limit?

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.

Can I combine `cursor` and `maxItems`?

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

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.