How to Handle API Pagination

TL;DR

Fetch all data from an API that returns results in paginated pages, without loading everything into memory at once.

Prerequisites

  • A REST API source already configured (see How to Add a REST API Source)

  • Understanding of your API’s pagination method (offset/limit, cursor-based, or page numbers)

  • API documentation showing pagination parameters

Steps

Step 1: Understand Your API’s Pagination

Check your API documentation to identify which pagination pattern it uses:

Offset/Limit (most common):

GET /users?offset=0&limit=100
GET /users?offset=100&limit=100
GET /users?offset=200&limit=100

Cursor-Based:

GET /users?limit=100&cursor=abc123
GET /users?limit=100&cursor=def456

Page Numbers:

GET /users?page=1&per_page=100
GET /users?page=2&per_page=100

Step 2: Implement Pagination in Your Connection Module

Update modules/my_api_client.py to handle pagination:

import requests
import logging

logger = logging.getLogger(__name__)

class MyAPIClient:
    def __init__(self, base_url=None, api_key=None, **kwargs):
        self.base_url = base_url
        self.api_key = api_key
        self.session = requests.Session()
        
        if self.api_key:
            self.session.headers.update({
                'Authorization': f'Bearer {self.api_key}'
            })
    
    def fetch(self, resource, page_size=100, **params):
        """
        Fetch all records from paginated API endpoint.
        
        Args:
            resource: API endpoint path
            page_size: Records per page (default: 100)
            **params: Additional query parameters
        
        Yields:
            Individual records from all pages
        """
        offset = 0
        total_fetched = 0
        
        while True:
            # Add pagination parameters
            pagination_params = {
                **params,
                'offset': offset,
                'limit': page_size
            }
            
            url = f"{self.base_url}{resource}"
            
            try:
                logger.debug(f"Fetching page: offset={offset}, limit={page_size}")
                response = self.session.get(url, params=pagination_params, timeout=30)
                response.raise_for_status()
                
                data = response.json()
                items = data.get('items', data.get('results', data.get('data', [])))
                
                if not items:
                    logger.info(f"Reached end of results at offset {offset}")
                    break
                
                # Yield each item individually (memory efficient)
                for item in items:
                    total_fetched += 1
                    yield item
                
                # Move to next page
                offset += page_size
                
            except requests.RequestException as e:
                logger.error(f"API request failed: {e}")
                raise


def fetch(connection=None, source=None, **kwargs):
    """
    Fetch the data. This is the function that is called by oaa-runner.
    
    Args:
        connection: Connection configuration from config.yaml
        source: Source configuration from config.yaml
        **kwargs: Additional keyword arguments
        
    Returns:
        PETL table with fetched records (using lazy_dict_table for memory efficiency)
    """
    from oaa.modules.lazy_dicts import lazy_dict_table
    
    api = MyAPIClient(**connection.params)
    resource = source.params.get('resource', '')
    page_size = source.params.get('page_size', 100)
    records = api.fetch(resource=resource, page_size=page_size)
    return lazy_dict_table(records)

Step 3: Configure Page Size in config.yaml

Set the pagination page size in your source configuration:

sources:
  users:
    destination: LocalUser
    source_type: custom
    connection: my_api
    params:
      resource: /users
      page_size: 100  # Adjust based on API limits
    field_mapping:
      user_id:
        destination_field: unique_id
      email:
        destination_field: email

Step 4: Handle Different Pagination Styles

Different APIs use different pagination patterns. Here are implementations for each:

Cursor-Based Pagination

def fetch(self, resource, page_size=100, **params):
    """Cursor-based pagination (e.g., Slack, GitHub)"""
    cursor = None
    total_fetched = 0
    
    while True:
        pagination_params = {**params, 'limit': page_size}
        
        if cursor:
            pagination_params['cursor'] = cursor
        
        response = self.session.get(
            f"{self.base_url}{resource}",
            params=pagination_params,
            timeout=30
        )
        response.raise_for_status()
        
        data = response.json()
        items = data.get('items', [])
        
        if not items:
            break
        
        for item in items:
            total_fetched += 1
            yield item
        
        # Get next cursor from response
        cursor = data.get('pagination', {}).get('next_cursor')
        if not cursor:
            break
        
        logger.info(f"Fetched {total_fetched} records so far...")

Page Number Pagination

def fetch(self, resource, page_size=100, **params):
    """Page-based pagination (e.g., many REST APIs)"""
    page = 1
    total_fetched = 0
    
    while True:
        pagination_params = {
            **params,
            'page': page,
            'per_page': page_size
        }
        
        response = self.session.get(
            f"{self.base_url}{resource}",
            params=pagination_params,
            timeout=30
        )
        response.raise_for_status()
        
        items = response.json()
        
        if not items:
            break
        
        for item in items:
            total_fetched += 1
            yield item
        
        page += 1
        logger.info(f"Fetched {total_fetched} records so far...")

Examples

Real Integration: ThousandEyes

ThousandEyes uses offset/limit pagination:

From netapp/thousandeyes/modules/thousandeyes.py:

class ThousandEyesAPI:
    def fetch(self, resource, **params):
        """Fetch all paginated results"""
        offset = 0
        page_size = 300  # ThousandEyes allows up to 300
        
        while True:
            params_with_pagination = {
                **params,
                'offset': offset,
                'limit': page_size
            }
            
            response = self.session.get(
                f"{self.base_uri}/{resource}",
                params=params_with_pagination
            )
            response.raise_for_status()
            
            data = response.json()
            results = data.get('results', [])
            
            if not results:
                break
            
            yield from results
            offset += page_size

Real Integration: Abnormal Security

Abnormal uses limit and offset pagination:

From gxo/abnormal/modules/abnormal.py:

class AbnormalAPI:
    def fetch(self, resource, **params):
        """Fetch users with pagination"""
        offset = 0
        page_size = 100
        
        while True:
            url = f"{self.base_url}{resource}"
            
            pagination_params = {
                **params,
                'limit': page_size,
                'offset': offset
            }
            
            response = self.session.get(url, params=pagination_params)
            response.raise_for_status()
            
            data = response.json()
            users = data.get('users', [])
            
            if not users:
                break
            
            for user in users:
                yield user
            
            offset += page_size

Performance Tips

✅ DO:

  • Use reasonable page sizes (100-300 is common)

  • Log progress for long-running fetches

  • Use yield to avoid loading all data into memory

  • Handle rate limiting with exponential backoff

  • Set appropriate timeouts (30 seconds typical)

❌ DON’T:

  • Request page_size > API limit (check docs)

  • Fetch all pages into a list before processing

  • Ignore rate limit headers

  • Use very small page sizes (causes many requests)

  • Fetch without handling network errors

Monitoring Pagination

Add logging to track progress:

def fetch(self, resource, page_size=100, **params):
    offset = 0
    total_fetched = 0
    start_time = time.time()
    
    while True:
        pagination_params = {**params, 'offset': offset, 'limit': page_size}
        response = self.session.get(f"{self.base_url}{resource}", 
                                    params=pagination_params, timeout=30)
        response.raise_for_status()
        
        items = response.json().get('items', [])
        if not items:
            elapsed = time.time() - start_time
            logger.info(f"Fetched {total_fetched} records in {elapsed:.1f}s")
            break
        
        for item in items:
            total_fetched += 1
            yield item
        
        if total_fetched % 1000 == 0:
            logger.info(f"Progress: {total_fetched} records fetched...")
        
        offset += page_size

Troubleshooting

Error: offset parameter not recognized

  • Check API documentation for correct pagination parameter names

  • Your API might use different names (page, start, skip, etc.)

Error: Getting duplicate records across pages

  • Verify offset is incrementing correctly

  • Check page_size matches API expectations

  • Some APIs have race conditions if data changes during pagination

Integration running very slowly

  • Try increasing page_size (check API limits)

  • Check network latency to API

  • Add logging to identify slow pages

Next Steps