Pagination

Nexios provides a powerful, flexible pagination system that supports multiple strategies, custom data handlers, and seamless integration with your existing APIs. Whether you're building simple list pagination or complex cursor-based navigation, Nexios has you covered.

Quick Start

Here's a quick example of how to paginate a list of items using response.paginate():

from nexios import NexiosApp

app = NexiosApp()

@app.get("/get-items")
async def get_items(request, response):
    sample_data = [{"id": i, "content": f"Item {i}"} for i in range(1, 101)]
    return response.paginate(sample_data)

For async endpoints, use response.apaginate():

from nexios import NexiosApp

app = NexiosApp()

@app.get("/get-items")
async def get_items(request, response):
    sample_data = [{"id": i, "content": f"Item {i}"} for i in range(1, 101)]
    return response.apaginate(sample_data)

Response Format

The pagination response includes both the data and comprehensive metadata:

{
  "items": [
    {"id": 1, "content": "Item 1"},
    {"id": 2, "content": "Item 2"},
    "...more items..."
  ],
  "pagination": {
    "total_items": 100,
    "total_pages": 5,
    "page": 1,
    "page_size": 20,
    "links": {
      "next": "http://127.0.0.1:8000/items?page=2&page_size=20",
      "first": "http://127.0.0.1:8000/items?page=1&page_size=20",
      "last": "http://127.0.0.1:8000/items?page=5&page_size=20"
    }
  }
}

Navigation Links

The links section provides ready-to-use URLs for navigating between pages. The prev link appears only when not on the first page.

Pagination Strategies

Nexios supports three main pagination strategies, each suited for different use cases:

1. Page Number Pagination (Default)

The most common pagination style, using page numbers and page sizes. Perfect for traditional web applications.

from nexios import NexiosApp

app = NexiosApp()

@app.get("/get-items")
async def get_items(request, response):
    sample_data = [{"id": i, "content": f"Item {i}"} for i in range(1, 101)]
    return response.paginate(sample_data, strategy="page_number")

Parameters:

• page_param: Query parameter name for page number (default: "page")
• page_size_param: Query parameter name for page size (default: "page_size")
• default_page: Default page number (default: 1)
• default_page_size: Default page size (default: 20)
• max_page_size: Maximum allowed page size (default: 100)

Example URLs:

• /items?page=2&page_size=10 - Page 2 with 10 items per page
• /items?page_size=50 - First page with 50 items per page
• /items - Uses defaults (page 1, 20 items per page)

2. Limit-Offset Pagination

Traditional SQL-style pagination using limit and offset. Ideal for database queries and APIs that follow REST conventions.

from nexios import NexiosApp

app = NexiosApp()

@app.get("/get-items")
async def get_items(request, response):
    sample_data = [{"id": i, "content": f"Item {i}"} for i in range(1, 101)]
    return response.paginate(sample_data, strategy="limit_offset")

Parameters:

• limit_param: Query parameter name for limit (default: "limit")
• offset_param: Query parameter name for offset (default: "offset")
• default_limit: Default limit value (default: 20)
• max_limit: Maximum allowed limit value (default: 100)

Example URLs:

• /items?limit=10&offset=20 - Items 21-30
• /items?limit=50 - First 50 items
• /items?offset=100 - Items starting from 101 (uses default limit)

3. Cursor Pagination

Cursor-based pagination for consistent pagination with changing datasets. Perfect for real-time feeds, infinite scroll, and large datasets.

from nexios import NexiosApp

app = NexiosApp()

@app.get("/get-items")
async def get_items(request, response):
    sample_data = [{"id": i, "content": f"Item {i}"} for i in range(1, 101)]
    return response.paginate(sample_data, strategy="cursor")

Parameters:

• cursor_param: Query parameter name for cursor (default: "cursor")
• page_size_param: Query parameter name for page size (default: "page_size")
• default_page_size: Default page size (default: 20)
• max_page_size: Maximum allowed page size (default: 100)
• sort_field: Field to use for cursor sorting (default: "id")

Example URLs:

• /items?cursor=eyJpZCI6IDEwfQ%3D%3D&page_size=10 - Items after ID 10
• /items?page_size=50 - First 50 items
• /items - Uses defaults (first page, 20 items per page)

Cursor Encoding

Cursors are base64-encoded JSON objects containing the sort field value. This makes them URL-safe and opaque to clients.

Advanced Configuration

Custom Strategy Parameters

You can customize pagination strategies by passing strategy instances instead of strings:

from nexios import NexiosApp
from nexios.pagination import PageNumberPagination, CursorPagination

app = NexiosApp()

# Custom page number pagination
@app.get("/items-page")
async def get_items_page(request, response):
    data = [{"id": i, "name": f"Item {i}"} for i in range(1, 101)]
    strategy = PageNumberPagination(
        page_param="p",
        page_size_param="size",
        default_page=1,
        default_page_size=10,
        max_page_size=50
    )
    return response.paginate(data, strategy=strategy)

# Custom cursor pagination with timestamp sorting
@app.get("/items-cursor")
async def get_items_cursor(request, response):
    data = [
        {"id": i, "name": f"Item {i}", "created_at": f"2023-01-{i:02d}T00:00:00Z"}
        for i in range(1, 31)
    ]
    strategy = CursorPagination(
        sort_field="created_at",
        default_page_size=5
    )
    return response.paginate(data, strategy=strategy)

Error Handling

Nexios provides built-in error handling for invalid pagination parameters:

from nexios import NexiosApp
from nexios.pagination import InvalidPageError, InvalidPageSizeError, InvalidCursorError
from nexios.http import HTTPException

app = NexiosApp()

@app.get("/items")
async def get_items(request, response):
    try:
        data = [{"id": i, "name": f"Item {i}"} for i in range(1, 101)]
        return response.paginate(data)
    except InvalidPageError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except InvalidPageSizeError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except InvalidCursorError as e:
        raise HTTPException(status_code=400, detail=str(e))

Common Errors:

• InvalidPageError: Page number < 1 or offset < 0
• InvalidPageSizeError: Page size < 1 or limit < 0
• InvalidCursorError: Malformed cursor encoding

Filtering and Sorting

Pagination works seamlessly with filtering and sorting:

from nexios import NexiosApp

app = NexiosApp()

@app.get("/products")
async def get_products(request, response):
    # Sample data with categories
    all_products = [
        {"id": i, "name": f"Product {i}", "category": "electronics" if i % 2 == 0 else "books", "price": i * 10}
        for i in range(1, 101)
    ]
    
    # Apply filters
    category = request.query_params.get("category")
    min_price = request.query_params.get("min_price")
    
    filtered_products = all_products
    if category:
        filtered_products = [p for p in filtered_products if p["category"] == category]
    if min_price:
        filtered_products = [p for p in filtered_products if p["price"] >= float(min_price)]
    
    # Sort results
    sort_by = request.query_params.get("sort", "id")
    reverse = request.query_params.get("order") == "desc"
    filtered_products.sort(key=lambda x: x.get(sort_by, 0), reverse=reverse)
    
    return response.paginate(filtered_products)

Example URLs:

• /products?category=electronics&sort=price&order=desc
• /products?min_price=50&page=2&page_size=20
• /products?category=books&sort=name

Preserving Query Parameters

Pagination links automatically preserve non-pagination query parameters, so filters and sorting persist across page navigation.

Data Handlers

Data handlers abstract the data source, allowing pagination to work with any type of data storage. Nexios provides built-in handlers for in-memory lists and async operations.

Built-in Data Handlers

SyncDataHandler

Base class for synchronous data handlers with two required methods:

• get_total_items() -> int: Returns total item count
• get_items(offset: int, limit: int) -> List[Any]: Returns paginated items

Built-in Implementation:

• SyncListDataHandler: Handles in-memory lists

AsyncDataHandler

Base class for asynchronous data handlers with two required methods:

• async get_total_items() -> int: Returns total item count
• async get_items(offset: int, limit: int) -> List[Any]: Returns paginated items

Built-in Implementation:

• AsyncListDataHandler: Handles in-memory lists asynchronously

Automatic Selection

By default, .paginate() uses AsyncListDataHandler for async functions and SyncListDataHandler for sync functions.

Custom Data Handlers

Create custom data handlers to integrate with databases, external APIs, or any data source:

Database Integration Examples

Tortoise ORM Example

from nexios import NexiosApp
from nexios.pagination import AsyncDataHandler
from tortoise.models import Model
from .models import Item

app = NexiosApp()

class TortoiseDataHandler(AsyncDataHandler):
    def __init__(self, query):
        self.query = query
    
    async def get_total_items(self) -> int:
        return await self.query.count()
    
    async def get_items(self, offset: int, limit: int) -> List[Any]:
        return await self.query.offset(offset).limit(limit).all()

@app.get("/items")
async def get_items(request, response):
    # Apply filters from query parameters
    query = Item.all()
    return response.paginate(query, data_handler=TortoiseDataHandler(query))

Custom Pagination Strategies

Create custom pagination strategies by subclassing BasePaginationStrategy:

Custom Strategy Example

from nexios import NexiosApp
from nexios.pagination import BasePaginationStrategy
from typing import Any, Dict, Tuple

app = NexiosApp()

class SeekPagination(BasePaginationStrategy):
    """Custom seek-based pagination similar to Facebook's API"""
    def __init__(self, seek_param: str = "seek", page_size_param: str = "limit", default_page_size: int = 20):
        self.seek_param = seek_param
        self.page_size_param = page_size_param
        self.default_page_size = default_page_size
    
    def parse_parameters(self, request_params: Dict[str, Any]) -> Tuple[Optional[str], int]:
        seek_value = request_params.get(self.seek_param)
        page_size = int(request_params.get(self.page_size_param, self.default_page_size))
        return seek_value, page_size
    
    def calculate_offset_limit(self, seek_value: Optional[str], page_size: int) -> Tuple[int, int]:
        # For seek pagination, we'd typically use this in the database query
        # Here we simulate with offset calculation
        if seek_value:
            # In real implementation, you'd find the position of seek_value
            return int(seek_value), page_size
        return 0, page_size
    
    def generate_metadata(self, total_items: int, items: List[Any], 
                        base_url: str, request_params: Dict[str, Any]) -> Dict[str, Any]:
        seek_value, page_size = self.parse_parameters(request_params)
        
        # Generate next seek value
        next_seek = str(int(seek_value or 0) + len(items)) if items else None
        
        links = {}
        if next_seek and len(items) == page_size:
            links["next"] = f"{base_url}?{self.seek_param}={next_seek}&{self.page_size_param}={page_size}"
        
        return {
            "total_items": total_items,
            "page_size": page_size,
            "seek_value": seek_value,
            "has_more": len(items) == page_size,
            "links": links
        }

@app.get("/items-seek")
async def get_items_seek(request, response):
    data = [{"id": i, "name": f"Item {i}"} for i in range(1, 101)]
    return response.paginate(data, strategy=SeekPagination())

Overridable Methods

When creating custom strategies, you can override these methods:

• parse_parameters(request_params: Dict[str, Any]) -> Any: Extract pagination parameters from request
• calculate_offset_limit(*args) -> Tuple[int, int]: Convert parameters to offset/limit
• generate_metadata(total_items, items, base_url, request_params) -> Dict[str, Any]: Create pagination metadata

Advanced Custom Strategy

from nexios.pagination import BasePaginationStrategy, LinkBuilder
from typing import Any, Dict, Tuple
import math

class HybridPagination(BasePaginationStrategy):
    """Hybrid strategy that supports both page and cursor-based pagination"""
    def __init__(self, page_param: str = "page", cursor_param: str = "cursor", 
                 page_size_param: str = "page_size", default_page_size: int = 20):
        self.page_param = page_param
        self.cursor_param = cursor_param
        self.page_size_param = page_size_param
        self.default_page_size = default_page_size
    
    def parse_parameters(self, request_params: Dict[str, Any]) -> Dict[str, Any]:
        page = request_params.get(self.page_param)
        cursor = request_params.get(self.cursor_param)
        page_size = int(request_params.get(self.page_size_param, self.default_page_size))
        
        return {
            "page": int(page) if page else None,
            "cursor": cursor,
            "page_size": page_size,
            "is_cursor_based": cursor is not None
        }
    
    def calculate_offset_limit(self, params: Dict[str, Any]) -> Tuple[int, int]:
        if params["is_cursor_based"]:
            # Cursor-based logic (simplified)
            return int(params["cursor"] or 0), params["page_size"]
        else:
            # Page-based logic
            page = params["page"] or 1
            return (page - 1) * params["page_size"], params["page_size"]
    
    def generate_metadata(self, total_items: int, items: List[Any], 
                        base_url: str, request_params: Dict[str, Any]) -> Dict[str, Any]:
        params = self.parse_parameters(request_params)
        
        metadata = {
            "total_items": total_items,
            "page_size": params["page_size"],
            "strategy": "cursor" if params["is_cursor_based"] else "page"
        }
        
        if params["is_cursor_based"]:
            # Cursor metadata
            next_cursor = str(int(params["cursor"] or 0) + len(items)) if items else None
            metadata.update({
                "cursor": params["cursor"],
                "next_cursor": next_cursor,
                "has_more": len(items) == params["page_size"]
            })
        else:
            # Page metadata
            page = params["page"] or 1
            total_pages = math.ceil(total_items / params["page_size"]) if params["page_size"] else 1
            metadata.update({
                "page": page,
                "total_pages": total_pages,
                "has_next": page < total_pages,
                "has_prev": page > 1
            })
        
        return metadata