Pagination

Modified on: Tue, 14 Jul, 2026 at 11:37 AM

When fetching data from external APIs, results are often returned across multiple pages. The maltego-transforms library provides built-in paginators that handle the complexity of fetching, rate-limiting, and aggregating paginated results.

All pagination examples shown here are available in the template project. Run ``maltego-transforms start my_project`` and see ``transforms/pagination_example.py`` to experiment with these patterns directly using **real free APIs**.

Overview

Three pagination strategies are supported:

Paginator	Use Case	Example API Pattern
`OffsetLimitPaginator`	SQL-style pagination	`?offset=0&limit=50` or `?skip=0&limit=50`
`PageBasedPaginator`	Page number pagination	`?page=1&per_page=50`
`CursorBasedPaginator`	Token/cursor pagination	`?cursor=abc123`

All paginators:

Integrate with IntegrationClient for rate limiting and throttling
Support parallel fetching when total count is known
Use the Maltego slider value (user's result limit) to control pagination
Handle errors gracefully (partial results on failure)

**Slider Limit**: Paginators decide whether to fetch the *next page* based on slider, but return full pages. Your transform should truncate results to exactly match the slider:

.. container::

   ::

      items = await paginator.fetch_all_items(slider=slider, ...)
      return [Phrase(item["name"]) for item in items[:slider]]  # Truncate!

Response Parser Functions

Before using any paginator, you need helper functions to parse API responses:

from typing import Any, Dict, List, Optional
from httpx import Response


def parse_items(response: Response, **kwargs: Any) -> Optional[List[Dict[str, Any]]]:
    """
    Extract items from an API response.
    Return None to stop pagination (e.g., on error or empty response).
    """
    try:
        data = response.json()
        return data.get("products", [])  # Adapt to your API
    except Exception:
        return None



def parse_total(response: Response) -> int:
    """Extract total count for parallel fetching optimization."""
    try:
        data = response.json()
        return data.get("total", 0)  # Adapt to your API
    except Exception:
        return 0

OffsetLimitPaginator

Use for APIs that accept offset/skip and limit query parameters.

How It Works

Request 1: GET /api/items?skip=0&limit=10   → items 1-10
Request 2: GET /api/items?skip=10&limit=10  → items 11-20
Request 3: GET /api/items?skip=20&limit=10  → items 21-30

Example: DummyJSON Products

This example uses the free DummyJSON API - no API key required!

from typing import Any, Dict, List, Optional
from httpx import Response
from maltego.entities import Phrase
from maltego.model.context import MaltegoContext
from maltego.pagination import OffsetLimitPaginator
from maltego.server import IntegrationClient, register_transform

client = IntegrationClient(
    max_concurrent=50,
    max_concurrent_per_key=6,
    max_calls_per_period=25,
    period_length_seconds=1.0,
    timeout=30,
)


def parse_dummyjson_products(
    response: Response, **kwargs: Any
) -> Optional[List[Dict[str, Any]]]:
    """Parse products from DummyJSON API."""
    try:
        data = response.json()
        return data.get("products", [])
    except Exception:
        return None


def parse_dummyjson_total(response: Response) -> int:
    """Get total product count from DummyJSON API."""
    try:
        data = response.json()
        return data.get("total", 0)
    except Exception:
        return 0


@register_transform(
    display_name="Fetch Products (Offset Pagination)",
    description="Fetch products from DummyJSON API",
    transform_set="New Maltego Integration",
)
async def fetch_dummyjson_products(
    input_entity: Phrase,
    slider: int,
    context: MaltegoContext,
) -> List[Phrase]:
    """
    Fetch products from DummyJSON using offset/limit pagination.

    API: https://dummyjson.com/products?skip=0&limit=10

    DummyJSON uses 'skip' instead of 'offset', demonstrating how to
    customize parameter names for different APIs.
    """
    paginator: OffsetLimitPaginator[Dict[str, Any]] = OffsetLimitPaginator(
        client=client,
        response_to_items=parse_dummyjson_products,
        page_size=10,
        page_size_param_name="limit",
        offset_param_name="skip",  # DummyJSON uses 'skip' not 'offset'
        response_to_total_cnt=parse_dummyjson_total,
        max_pages=10,
    )

    context.log.inform(f"Fetching products from DummyJSON (max {slider} items)...")

    items = await paginator.fetch_all_items(
        slider=slider,
        context=context,
        url="https://dummyjson.com/products",
    )

    context.log.inform(f"Fetched {len(items)} products")

    # Truncate to slider limit (paginator may return slightly more)
    return [
        Phrase(f"{item.get('title', 'Unknown')} - ${item.get('price', 0)}")
        for item in items[:slider]
    ]

Configuration Options

paginator = OffsetLimitPaginator(
    client=client,
    response_to_items=parse_items,

    # Required
    page_size=50,

    # Optional - customize parameter names
    page_size_param_name="limit",    # Default: "limit"
    offset_param_name="offset",      # Default: "offset" (use "skip" for DummyJSON)

    # Optional - optimization
    response_to_total_cnt=parse_total,  # Enable parallel fetching

    # Optional - safety limits
    max_pages=10,                    # Stop after N pages
    min_page_fill=0.8,               # Stop if page <80% full
    request_extra_items_pct=0.1,     # Request 10% extra items
)

PageBasedPaginator

Use for APIs that accept page and per_page (or similar) query parameters.

How It Works

Request 1: GET /api/items?page=1&limit=12  → items 1-12
Request 2: GET /api/items?page=2&limit=12  → items 13-24
Request 3: GET /api/items?page=3&limit=12  → items 25-36

Example: Art Institute of Chicago Artworks

This example uses the free Art Institute of Chicago API and demonstrates:

Page-based pagination
Image entities with dynamic images loaded from URL
HTML display fields with embedded base64 thumbnails

from typing import Any, Dict, List, Optional
from httpx import Response
from maltego.entities import Image, Phrase
from maltego.model.context import MaltegoContext
from maltego.pagination import PageBasedPaginator
from maltego.server import IntegrationClient, register_transform


def parse_artic_artworks(
    response: Response, **kwargs: Any
) -> Optional[List[Dict[str, Any]]]:
    """Parse artworks from Art Institute of Chicago API."""
    try:
        data = response.json()
        return data.get("data", [])
    except Exception:
        return None


def parse_artic_total(response: Response) -> int:
    """Get total artwork count."""
    try:
        data = response.json()
        return data.get("pagination", {}).get("total", 0)
    except Exception:
        return 0


@register_transform(
    display_name="Fetch Artworks (Page Pagination)",
    description="Fetch artworks from Art Institute of Chicago API",
    transform_set="New Maltego Integration",
)
async def fetch_artic_artworks(
    input_entity: Phrase,
    slider: int,
    context: MaltegoContext,
) -> List[Image]:
    """
    Fetch artworks using page-based pagination.
    Returns Image entities with artwork thumbnails!
    """
    paginator: PageBasedPaginator[Dict[str, Any]] = PageBasedPaginator(
        client=client,
        response_to_items=parse_artic_artworks,
        page_size=12,
        page_size_param_name="limit",
        page_param_name="page",
        page_start_num=1,
        response_to_total_cnt=parse_artic_total,
        max_pages=10,
    )

    items = await paginator.fetch_all_items(
        slider=slider,
        context=context,
        url="https://api.artic.edu/api/v1/artworks",
    )

    # Convert to Image entities - truncate to slider limit
    results = []
    for item in items[:slider]:
        image_id = item.get("image_id")
        if not image_id:
            continue

        title = item.get("title", "Untitled")
        full_url = f"https://www.artic.edu/iiif/2/{image_id}/full/843,/0/default.jpg"

        img = Image(title)
        img.url = full_url  # Maltego loads the image from this URL
        img.set_property("artist", item.get("artist_display", "Unknown"), display_name="Artist")
        img.set_property("date", item.get("date_display", ""), display_name="Date")
        results.append(img)

    return results

CursorBasedPaginator

Use for APIs that return a cursor/token for the next page (common in modern APIs like Slack, Twitter, etc.).

Cursor lifetime is owned by the upstream API and caller. The SDK stores and reuses the next-page URL returned by ``response_to_cursor``, but it does not apply a generic TTL or expiry check to cursor values. If an upstream API expires a cursor, handle that API error in your transform or restart pagination from the first page according to that API's rules.

How It Works

Request 1: GET /api/items              → items + next_cursor="abc123"
Request 2: GET /api/items?cursor=abc123 → items + next_cursor="def456"
Request 3: GET /api/items?cursor=def456 → items + next_cursor=null (done)

Example

from maltego.pagination import CursorBasedPaginator

def get_next_url(response: Response) -> str:
    """Extract next page URL from response."""
    data = response.json()
    next_cursor = data.get("next_cursor")
    if next_cursor:
        return f"https://api.example.com/items?cursor={next_cursor}"
    return ""  # Empty string stops pagination

paginator = CursorBasedPaginator(
    client=client,
    response_to_items=parse_items,
    response_to_cursor=get_next_url,
    response_to_total_cnt=None,  # Usually not available with cursor pagination
    max_pages=10,
)

Streaming Results

For long-running transforms or real-time feedback, use stream_all_items() to yield entities as each page is fetched.

When streaming, you must manually track the count to respect the slider limit, since the paginator yields full pages.

@register_transform(
    display_name="Stream Products (Pagination)",
    description="Stream products page-by-page for real-time feedback",
    transform_set="New Maltego Integration",
)
async def stream_dummyjson_products(
    input_entity: Phrase,
    slider: int,
    context: MaltegoContext,
):
    """
    Stream products page-by-page using async generator.

    Note: The paginator yields full pages, so we track yielded count
    to respect the slider limit exactly.
    """
    paginator: OffsetLimitPaginator[Dict[str, Any]] = OffsetLimitPaginator(
        client=client,
        response_to_items=parse_dummyjson_products,
        page_size=10,
        page_size_param_name="limit",
        offset_param_name="skip",
        response_to_total_cnt=None,  # Sequential fetching for streaming
        max_pages=5,
    )

    page_num = 0
    yielded_count = 0  # Track how many entities we've yielded

    async for page_items in paginator.stream_all_items(
        slider=slider,
        context=context,
        url="https://dummyjson.com/products",
    ):
        page_num += 1
        context.log.inform(f"Page {page_num}: {len(page_items)} products")

        for item in page_items:
            if yielded_count >= slider:
                return  # Stop once we've hit the slider limit
            yield Phrase(f"{item.get('title', 'Unknown')} - ${item.get('price', 0)}")
            yielded_count += 1

Benefits of streaming:

Real-time feedback: Users see results as they arrive
Memory efficient: No need to hold all items in memory
Early termination: Can stop early if needed

Parallel Fetching

When response_to_total_cnt is provided, paginators can fetch pages in parallel after determining the total count:

paginator = OffsetLimitPaginator(
    client=client,
    response_to_items=parse_items,
    response_to_total_cnt=parse_total,  # Enables parallel fetching
    page_size=100,
)

How it works:

First request determines total count
Remaining pages are fetched in parallel (respecting rate limits)
Results are aggregated and returned

Error Handling

Paginators handle errors gracefully:

Safe mode (default): Failed requests return partial results

items = await paginator.fetch_all_items(...)  # Returns items from successful pages

Unsafe mode: Failed requests raise exceptions

items = await paginator.fetch_all_items_unsafe(...)  # Raises on any failure

Try It Yourself

To experiment with pagination using real APIs:

# Create a new project from the template
maltego-transforms start my_project
cd my_project


# The pagination examples are in:
# transforms/pagination_example.py


# Start the server
python project.py

The template includes working examples that fetch:

Products from DummyJSON (offset/limit pagination)
Artworks from Art Institute of Chicago (page-based pagination with images!)
Streaming example with real-time feedback

Overview

Response Parser Functions

OffsetLimitPaginator

How It Works

Example: DummyJSON Products

Configuration Options

PageBasedPaginator

How It Works

Example: Art Institute of Chicago Artworks

CursorBasedPaginator

How It Works

Example

Streaming Results

Parallel Fetching

Error Handling

Try It Yourself

Search all Maltego Guides: