Difficulty: Advanced
Pagination and filtering are essential for APIs that serve large datasets. Without pagination, an endpoint returning thousands of records would be slow, consume excessive bandwidth, and overwhelm clients. Filtering lets clients request only the data they need.
Offset-based pagination is the most common approach. Clients pass page and per_page (or limit and offset) query parameters. The API returns a slice of results along with metadata: total count, current page, total pages, and links to next/previous pages. Flask-SQLAlchemy's paginate() method handles this pattern well.
Cursor-based pagination is more efficient for large datasets. Instead of page numbers, the client passes a cursor (usually the ID or timestamp of the last item). The API returns items after that cursor. This avoids the performance issues of OFFSET on large tables and handles concurrent inserts gracefully.
Filtering allows clients to narrow down results. Common patterns include exact match (status=active), partial match (name__contains=john), range (price_min=10&price_max=50), and boolean filters (in_stock=true). Parse these from query parameters and build dynamic queries.
Sorting lets clients control the order of results. The sort parameter typically uses a field name with an optional prefix for direction: sort=name (ascending) or sort=-name (descending). Support multiple sort fields: sort=category,-price.
HATEOAS (Hypermedia as the Engine of Application State) is a REST principle where API responses include links to related resources and actions. For pagination, this means including links to the next page, previous page, first page, and last page. Clients follow these links instead of constructing URLs themselves.
Performance considerations: always limit the maximum per_page value to prevent clients from requesting millions of records. Add database indexes on columns used for filtering and sorting. Use select_related or joinedload for related data. Consider caching paginated responses for frequently accessed data.
from flask import Flask, request, jsonify, url_for
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
db = SQLAlchemy(app)
class Product(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100))
price = db.Column(db.Float)
category = db.Column(db.String(50))
@app.get('/api/products')
def list_products():
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 100)
pagination = Product.query.order_by(Product.id)\
.paginate(page=page, per_page=per_page, error_out=False)
return jsonify({
'items': [{'id': p.id, 'name': p.name, 'price': p.price}
for p in pagination.items],
'meta': {
'page': page,
'per_page': per_page,
'total': pagination.total,
'pages': pagination.pages,
'has_next': pagination.has_next,
'has_prev': pagination.has_prev,
},
'links': {
'next': url_for('list_products', page=page+1, per_page=per_page,
_external=True) if pagination.has_next else None,
'prev': url_for('list_products', page=page-1, per_page=per_page,
_external=True) if pagination.has_prev else None,
}
})Offset-based pagination uses page and per_page parameters. The response includes items, metadata (total, pages), and links to next/previous pages. Always cap per_page with min() to prevent abuse.
from flask import request, jsonify
@app.get('/api/products')
def list_products():
query = Product.query
# Filter by category
category = request.args.get('category')
if category:
query = query.filter(Product.category == category)
# Filter by price range
min_price = request.args.get('min_price', type=float)
max_price = request.args.get('max_price', type=float)
if min_price is not None:
query = query.filter(Product.price >= min_price)
if max_price is not None:
query = query.filter(Product.price <= max_price)
# Search by name
search = request.args.get('q')
if search:
query = query.filter(Product.name.ilike(f'%{search}%'))
# Sorting: sort=price or sort=-price
sort = request.args.get('sort', 'id')
if sort.startswith('-'):
query = query.order_by(getattr(Product, sort[1:]).desc())
else:
query = query.order_by(getattr(Product, sort).asc())
# Paginate
page = request.args.get('page', 1, type=int)
pagination = query.paginate(page=page, per_page=20, error_out=False)
return jsonify({
'items': [p.to_dict() for p in pagination.items],
'total': pagination.total
})Build the query dynamically based on query parameters. Filter by exact match, range, and search. Sort by field name with '-' prefix for descending. Apply pagination last. Each filter is optional.
from flask import request, jsonify
from sqlalchemy import and_
@app.get('/api/posts')
def list_posts():
limit = min(request.args.get('limit', 20, type=int), 100)
cursor = request.args.get('cursor', type=int) # Last seen ID
query = Post.query.order_by(Post.id.asc())
if cursor:
query = query.filter(Post.id > cursor)
posts = query.limit(limit + 1).all() # Fetch one extra to check has_next
has_next = len(posts) > limit
if has_next:
posts = posts[:limit] # Remove the extra item
next_cursor = posts[-1].id if posts and has_next else None
return jsonify({
'items': [p.to_dict() for p in posts],
'next_cursor': next_cursor,
'has_next': has_next,
})
# Client usage:
# GET /api/posts?limit=20 -> first page
# GET /api/posts?limit=20&cursor=42 -> items after id 42Cursor pagination uses the last item's ID as a cursor. Query for items with ID > cursor. Fetch limit+1 items to determine if there are more. Return the last item's ID as next_cursor for the client to use.
from flask import request, url_for, jsonify
def paginate_query(query, schema_fn, endpoint, kwargs):
"""Generic pagination helper."""
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 100)
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
return jsonify({
'items': [schema_fn(item) for item in pagination.items],
'meta': {
'page': page,
'per_page': per_page,
'total': pagination.total,
'pages': pagination.pages,
},
'links': {
'self': url_for(endpoint, page=page, per_page=per_page,
_external=True, kwargs),
'next': url_for(endpoint, page=page+1, per_page=per_page,
_external=True, kwargs) if pagination.has_next else None,
'prev': url_for(endpoint, page=page-1, per_page=per_page,
_external=True, kwargs) if pagination.has_prev else None,
}
})
# Usage:
@app.get('/api/users')
def list_users():
query = User.query.order_by(User.created_at.desc())
return paginate_query(query, lambda u: u.to_dict(), 'list_users')
@app.get('/api/posts')
def list_posts():
query = Post.query.order_by(Post.created_at.desc())
return paginate_query(query, lambda p: p.to_dict(), 'list_posts')A reusable helper function handles pagination for any query. Pass the query, a serialization function, and the endpoint name. The helper handles page/per_page parsing, metadata, and HATEOAS links.
Pagination, Filtering, Sorting, Query Parameters, HATEOAS, Cursor Pagination