Pagination

The Intapp REST Client provides memory-efficient streaming pagination for large datasets.

Why Streaming Pagination?

Traditional pagination loads all pages into memory:

# ❌ Memory-intensive approach
all_data = []
page = 0
while True:
    response = client.get("/api/items", params={"skip": page * 1000, "limit": 1000})
    items = response.get("items", [])
    if not items:
        break
    all_data.extend(items)  # All in memory!
    page += 1
# With 1M records, all_data uses ~1GB+ memory

Streaming pagination processes data in batches:

# ✅ Memory-efficient approach
for batch in client.get_paginated("/api/items", page_size=1000, data_key="items"):
    process_batch(batch)  # Only one batch in memory at a time
# Memory usage stays constant regardless of total records
 
# Optional: fetch pages in parallel for faster throughput (still yielded in order)
for batch in client.get_paginated("/api/items", 1000, "items", max_concurrency=5):
    process_batch(batch)

GET Pagination

Standard pagination using query parameters:

from intapp_rest_client import RestClient
 
with RestClient(base_url="https://api.example.com", api_key="key") as client:
    for batch in client.get_paginated(
        endpoint="/api/v4/data/rows/Company",
        page_size=1000,
        data_key="rows",  # Key in response containing the array
        params={"status": "active"},  # Additional query params
        skip_param="skip",   # Default: "skip"
        limit_param="limit"  # Default: "limit"
    ):
        for row in batch:
            print(row["CompanyName"])

Parameters

Parameter	Default	Description
`endpoint`	(required)	API endpoint
`page_size`	(required)	Records per page
`data_key`	(required)	JSON key containing the array
`params`	None	Additional query parameters
`headers`	None	Additional headers
`skip_param`	"skip"	Query param name for offset
`limit_param`	"limit"	Query param name for page size
`max_concurrency`	1	Parallel page fetches; when > 1, pages are fetched in parallel but yielded in order

Example Response Structure

The pagination methods expect responses like:

{
  "rows": [
    {"EntryId": 1, "CompanyName": "Acme"},
    {"EntryId": 2, "CompanyName": "Beta"}
  ],
  "totalCount": 50000
}

Where data_key="rows" extracts the array.

POST Pagination (Body)

For APIs where pagination parameters go in the request body:

# Query with filter, pagination in body
for batch in client.post_paginated(
    endpoint="/api/v4/query",
    page_size=1000,
    data_key="results",
    json={"filter": {"status": "Active"}},  # Base request body
    skip_key="skip",   # Key in body for offset
    limit_key="limit",  # Key in body for page size
    max_concurrency=1   # Use > 1 for parallel page fetches (yielded in order)
):
    process_batch(batch)

Each request sends:

{
  "filter": {"status": "Active"},
  "skip": 0,
  "limit": 1000
}

Then:

{
  "filter": {"status": "Active"},
  "skip": 1000,
  "limit": 1000
}

POST Pagination (Query Params)

For APIs using POST method but with pagination in URL parameters:

# DealCloud view with query string pagination
for batch in client.post_paginated_params(
    endpoint="/api/v4/data/rows/view/123",
    page_size=1000,
    data_key="rows",
    json={"filter": {"status": "Active"}},  # Optional body (unchanged)
    params={"wrapIntoArrays": "true"},       # Base query params
    skip_param="skip",   # Query param for offset
    limit_param="limit",  # Query param for page size
    max_concurrency=1     # Use > 1 for parallel page fetches (yielded in order)
):
    entry_ids = [row["EntryId"] for row in batch]
    # Process or delete in batches

💡

Use post_paginated() when skip/limit go in the request body. Use post_paginated_params() when skip/limit go in the query string.

Async Pagination

All pagination methods have async versions:

import asyncio
from intapp_rest_client import RestClient
 
async def process_all_data():
    async with RestClient(base_url="https://api.example.com", api_key="key") as client:
        # Async GET pagination
        async for batch in client.aget_paginated(
            "/api/v4/data/rows/Company",
            page_size=1000,
            data_key="rows"
        ):
            await process_batch(batch)
        
        # Async POST pagination (body)
        async for batch in client.apost_paginated(
            "/api/v4/query",
            page_size=1000,
            data_key="results",
            json={"filter": {"status": "Active"}}
        ):
            await process_batch(batch)
        
        # Async POST pagination (query params)
        async for batch in client.apost_paginated_params(
            "/api/v4/data/rows/view/123",
            page_size=1000,
            data_key="rows"
        ):
            await process_batch(batch)
 
asyncio.run(process_all_data())

Common Patterns

Stream to File

Export large datasets without loading into memory:

import json
 
with open("export.jsonl", "w") as f:
    for batch in client.get_paginated("/api/v4/data/rows/Contact", 1000, "rows"):
        for row in batch:
            f.write(json.dumps(row) + "\n")

Batch Delete

Delete records in manageable chunks:

# Get all record IDs to delete
for batch in client.get_paginated("/api/v4/data/rows/OldRecords", 1000, "rows"):
    entry_ids = [row["EntryId"] for row in batch]
    client.delete("/api/v4/data/entrydata/OldRecords", json=entry_ids)
    print(f"Deleted {len(entry_ids)} records")

Transform and Load

ETL-style processing:

def transform(row):
    return {
        "name": row["FirstName"] + " " + row["LastName"],
        "email": row["Email"].lower()
    }
 
for batch in client.get_paginated("/api/source/contacts", 1000, "data"):
    transformed = [transform(row) for row in batch]
    client.post("/api/target/contacts/bulk", json=transformed)

Progress Tracking

from tqdm import tqdm
 
# First, get total count if available
response = client.get("/api/v4/data/rows/Company", params={"limit": 1})
total = response.get("totalCount", 0)
 
processed = 0
with tqdm(total=total, desc="Processing") as pbar:
    for batch in client.get_paginated("/api/v4/data/rows/Company", 1000, "rows"):
        process_batch(batch)
        processed += len(batch)
        pbar.update(len(batch))

Parallel Batch Processing

Process batches concurrently while fetching:

import asyncio
from collections import deque
 
async def process_with_prefetch(client, max_concurrent=3):
    """Process batches with concurrent execution."""
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def process_one(batch):
        async with semaphore:
            await heavy_processing(batch)
    
    tasks = []
    async for batch in client.aget_paginated("/api/items", 1000, "items"):
        task = asyncio.create_task(process_one(batch))
        tasks.append(task)
    
    await asyncio.gather(*tasks)

Stopping Early

Break out of pagination when you've found what you need:

target_id = 12345
found = None
 
for batch in client.get_paginated("/api/v4/data/rows/Company", 1000, "rows"):
    for row in batch:
        if row["EntryId"] == target_id:
            found = row
            break
    if found:
        break
 
if found:
    print(f"Found: {found['CompanyName']}")

Error Handling

from intapp_rest_client import HttpError
 
try:
    for batch in client.get_paginated("/api/data", 1000, "rows"):
        process_batch(batch)
except HttpError as e:
    print(f"Pagination failed at some point: {e}")
    # The client logs which skip/limit failed

Best Practices

Choose appropriate page_size:
- Too small (100): Many requests, slower
- Too large (10000): Memory pressure, timeout risk
- Sweet spot: 500-2000 for most APIs
Use streaming for large datasets: Don't collect all pages into a list
Handle errors gracefully: Pagination can fail mid-stream
Consider rate limits: Add delays if hitting 429 errors frequently

Next Steps

Async Operations - Async pagination patterns
Error Handling - Handle pagination errors
Retry Configuration - Retry failed pages

Async Operations Streaming File Download