Flask Advanced Features Guide
Table of Contents
- Rate Limiting with Flask-Limiter
- Caching with Flask-Caching
- Pagination Patterns
- Advanced Blueprint Architecture
- Request Validation
- Error Handling
- Application Factory Pattern
- JWT Authentication
- CORS Configuration
- Production Best Practices
Rate Limiting with Flask-Limiter
Rate limiting prevents abuse by restricting the number of requests a client can make.
Installation
pip install Flask-Limiter
Basic Setup
from flask import Flask
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
app = Flask(__name__)
# Initialize limiter
limiter = Limiter(
app=app,
key_func=get_remote_address, # Use IP address as key
default_limits=["200 per day", "50 per hour"], # Global limits
storage_uri="redis://localhost:6379" # For distributed systems
)
Applying Rate Limits
Global Limits:
# Applied to all routes via default_limits in config
limiter = Limiter(
app=app,
default_limits=["200 per day", "50 per hour"]
)
Per-Route Limits:
@app.route('/api/search')
@limiter.limit("5 per minute") # Override global limit
def search():
return {"results": []}
@app.route('/api/expensive-operation')
@limiter.limit("1 per hour") # Very restrictive
def expensive_operation():
return {"status": "processing"}
Multiple Limits:
@app.route('/api/posts')
@limiter.limit("10 per minute")
@limiter.limit("100 per hour")
@limiter.limit("1000 per day")
def get_posts():
"""Enforces all three limits."""
return {"posts": []}
Dynamic Rate Limits
def get_user_tier():
"""Determine rate limit based on user tier."""
user = get_current_user()
if user and user.tier == 'premium':
return "1000 per hour"
elif user and user.tier == 'standard':
return "100 per hour"
return "10 per hour" # Free tier
@app.route('/api/data')
@limiter.limit(get_user_tier)
def get_data():
return {"data": []}
Exempt Routes
@app.route('/health')
@limiter.exempt # No rate limiting
def health_check():
return {"status": "healthy"}
Custom Key Functions
from flask import request
def get_user_id():
"""Use user ID instead of IP for authenticated requests."""
token = request.headers.get('Authorization')
if token:
user = decode_token(token)
return f"user:{user.id}"
return get_remote_address()
limiter = Limiter(
app=app,
key_func=get_user_id
)
Rate Limit Response
@app.errorhandler(429)
def ratelimit_handler(e):
"""Custom response for rate limit exceeded."""
return {
"error": "Rate limit exceeded",
"message": str(e.description),
"retry_after": e.description.split()[-1]
}, 429
Complete Example
from flask import Flask, jsonify
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
app = Flask(__name__)
# Configure limiter
limiter = Limiter(
app=app,
key_func=get_remote_address,
default_limits=["200 per day", "50 per hour"],
storage_uri="memory://" # Use Redis in production
)
@app.route('/api/books')
@limiter.limit("10 per minute")
def get_books():
"""Get books with rate limiting."""
books = [
{"id": 1, "title": "1984"},
{"id": 2, "title": "Brave New World"}
]
return jsonify(books)
@app.route('/api/books/<int:book_id>')
@limiter.limit("30 per minute") # More generous for single item
def get_book(book_id):
"""Get single book with higher rate limit."""
book = {"id": book_id, "title": "Example Book"}
return jsonify(book)
@app.errorhandler(429)
def ratelimit_error(e):
return jsonify(error="Rate limit exceeded", message=str(e.description)), 429
if __name__ == '__main__':
app.run(debug=True)
Caching with Flask-Caching
Caching stores frequently accessed data to reduce database queries and improve performance.
Installation
pip install Flask-Caching
Basic Setup
from flask import Flask
from flask_caching import Cache
app = Flask(__name__)
# Configure cache
app.config['CACHE_TYPE'] = 'SimpleCache' # Memory cache
app.config['CACHE_DEFAULT_TIMEOUT'] = 300 # 5 minutes
# Or use Redis for production
app.config['CACHE_TYPE'] = 'RedisCache'
app.config['CACHE_REDIS_HOST'] = 'localhost'
app.config['CACHE_REDIS_PORT'] = 6379
app.config['CACHE_REDIS_DB'] = 0
# Initialize cache
cache = Cache(app)
Caching Routes
Basic Route Caching:
@app.route('/api/popular-books')
@cache.cached(timeout=600) # Cache for 10 minutes
def get_popular_books():
"""Expensive query - cache the result."""
books = Book.query.order_by(Book.views.desc()).limit(10).all()
return jsonify([book.to_dict() for book in books])
Conditional Caching:
@app.route('/api/books')
@cache.cached(timeout=300, unless=lambda: request.args.get('nocache'))
def get_books():
"""Cache unless nocache parameter is present."""
books = Book.query.all()
return jsonify([book.to_dict() for book in books])
Cache with Query Parameters:
@app.route('/api/books')
@cache.cached(timeout=300, query_string=True)
def get_books():
"""Different cache for different query parameters."""
page = request.args.get('page', 1, type=int)
books = Book.query.paginate(page=page, per_page=20)
return jsonify([book.to_dict() for book in books.items])
Memoization (Function Caching)
@cache.memoize(timeout=600)
def get_user_books(user_id):
"""Cache result based on user_id argument."""
books = Book.query.filter_by(user_id=user_id).all()
return [book.to_dict() for book in books]
@app.route('/api/users/<int:user_id>/books')
def user_books(user_id):
"""Use memoized function."""
books = get_user_books(user_id)
return jsonify(books)
Cache Invalidation
Delete Specific Cache:
@app.route('/api/books', methods=['POST'])
def create_book():
"""Create book and invalidate cache."""
# Create book
book = Book(**request.json)
db.session.add(book)
db.session.commit()
# Invalidate cached book lists
cache.delete('view//api/books')
cache.delete('view//api/popular-books')
return jsonify(book.to_dict()), 201
Delete Memoized Cache:
@app.route('/api/books/<int:book_id>', methods=['PUT'])
def update_book(book_id):
"""Update book and clear its cache."""
book = Book.query.get_or_404(book_id)
book.title = request.json.get('title', book.title)
db.session.commit()
# Clear memoized cache for this book
cache.delete_memoized(get_user_books, book.user_id)
return jsonify(book.to_dict())
Clear All Cache:
@app.route('/admin/clear-cache', methods=['POST'])
@admin_required
def clear_cache():
"""Clear entire cache."""
cache.clear()
return jsonify({"message": "Cache cleared"}), 200
Pagination with Caching
@app.route('/api/books')
@cache.cached(timeout=300, query_string=True)
def get_books_paginated():
"""Cache paginated results."""
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 20, type=int)
# Be careful with caching paginated routes
# Consider: cache.cached() may not be ideal if data changes frequently
pagination = Book.query.paginate(
page=page,
per_page=per_page,
error_out=False
)
return jsonify({
'books': [book.to_dict() for book in pagination.items],
'page': page,
'pages': pagination.pages,
'total': pagination.total
})
Note on Caching Pagination: Caching paginated routes can be tricky because:
- Data changes affect all pages
- Need to invalidate all page caches when data changes
- Consider shorter timeout for frequently updated data
Pagination Patterns
Basic Pagination
from flask import request, jsonify
@app.route('/api/books')
def get_books():
"""Paginated book list."""
# Get pagination parameters
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 20, type=int)
# Validate parameters
if per_page > 100:
per_page = 100 # Max items per page
# Query with pagination
pagination = Book.query.paginate(
page=page,
per_page=per_page,
error_out=False # Don't raise 404 for invalid page
)
# Build response
return jsonify({
'items': [book.to_dict() for book in pagination.items],
'pagination': {
'page': page,
'per_page': per_page,
'total_pages': pagination.pages,
'total_items': pagination.total,
'has_next': pagination.has_next,
'has_prev': pagination.has_prev,
'next_page': page + 1 if pagination.has_next else None,
'prev_page': page - 1 if pagination.has_prev else None
}
})
Pagination with Links
from urllib.parse import urlencode
def build_pagination_links(pagination, endpoint):
"""Build next/prev/first/last links."""
def build_url(page):
args = request.args.copy()
args['page'] = page
return f"{request.base_url}?{urlencode(args)}"
links = {
'self': build_url(pagination.page),
'first': build_url(1),
'last': build_url(pagination.pages)
}
if pagination.has_next:
links['next'] = build_url(pagination.page + 1)
if pagination.has_prev:
links['prev'] = build_url(pagination.page - 1)
return links
@app.route('/api/books')
def get_books():
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 20, type=int)
pagination = Book.query.paginate(page=page, per_page=per_page)
return jsonify({
'items': [book.to_dict() for book in pagination.items],
'links': build_pagination_links(pagination, 'get_books'),
'meta': {
'page': page,
'per_page': per_page,
'total': pagination.total
}
})
Cursor-Based Pagination
For better performance with large datasets:
@app.route('/api/books')
def get_books_cursor():
"""Cursor-based pagination (more efficient for large datasets)."""
cursor = request.args.get('cursor', type=int)
limit = request.args.get('limit', 20, type=int)
# Query items after cursor
query = Book.query.order_by(Book.id)
if cursor:
query = query.filter(Book.id > cursor)
books = query.limit(limit + 1).all()
# Check if there are more items
has_more = len(books) > limit
if has_more:
books = books[:limit]
# Build response
response = {
'items': [book.to_dict() for book in books],
'has_more': has_more
}
if has_more:
response['next_cursor'] = books[-1].id
return jsonify(response)
Advanced Blueprint Architecture
Modular Blueprint Structure
app/
├── __init__.py # Application factory
├── models.py # Database models
├── config.py # Configuration
├── auth/
│ ├── __init__.py
│ ├── routes.py # Auth routes
│ └── utils.py # Auth utilities
├── books/
│ ├── __init__.py
│ ├── routes.py # Book routes
│ └── schemas.py # Marshmallow schemas
└── users/
├── __init__.py
├── routes.py # User routes
└── schemas.py # Marshmallow schemas
Blueprint Definition
auth/__init__.py:
from flask import Blueprint
auth_bp = Blueprint('auth', __name__, url_prefix='/auth')
from . import routes # Import routes to register them
auth/routes.py:
from flask import request, jsonify
from . import auth_bp
from .utils import generate_token, verify_password
from app.models import User
@auth_bp.route('/login', methods=['POST'])
def login():
"""User login endpoint."""
data = request.get_json()
user = User.query.filter_by(email=data['email']).first()
if not user or not verify_password(data['password'], user.password_hash):
return jsonify({"error": "Invalid credentials"}), 401
token = generate_token(user)
return jsonify({"token": token}), 200
@auth_bp.route('/register', methods=['POST'])
def register():
"""User registration endpoint."""
data = request.get_json()
# Registration logic
user = User(**data)
db.session.add(user)
db.session.commit()
return jsonify(user.to_dict()), 201
Registering Blueprints
app/__init__.py:
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_cors import CORS
db = SQLAlchemy()
def create_app(config_name='development'):
"""Application factory pattern."""
app = Flask(__name__)
# Load configuration
app.config.from_object(f'app.config.{config_name.capitalize()}Config')
# Initialize extensions
db.init_app(app)
CORS(app)
# Register blueprints
from app.auth import auth_bp
from app.books import books_bp
from app.users import users_bp
app.register_blueprint(auth_bp)
app.register_blueprint(books_bp)
app.register_blueprint(users_bp)
return app
Blueprint with Decorators
from functools import wraps
from flask import request, jsonify
from . import books_bp
def token_required(f):
"""Decorator to require authentication."""
@wraps(f)
def decorated(*args, **kwargs):
token = request.headers.get('Authorization')
if not token:
return jsonify({"error": "Token required"}), 401
try:
# Verify token logic
user = verify_token(token)
return f(current_user=user, *args, **kwargs)
except Exception as e:
return jsonify({"error": "Invalid token"}), 401
return decorated
@books_bp.route('/books', methods=['POST'])
@token_required
def create_book(current_user):
"""Create book - requires authentication."""
data = request.get_json()
book = Book(user_id=current_user.id, **data)
db.session.add(book)
db.session.commit()
return jsonify(book.to_dict()), 201
Request Validation
Using Marshmallow for Validation
from marshmallow import Schema, fields, validate, ValidationError
class BookSchema(Schema):
"""Schema for book validation."""
id = fields.Int(dump_only=True)
title = fields.Str(required=True, validate=validate.Length(min=1, max=200))
author = fields.Str(required=True, validate=validate.Length(min=1, max=100))
isbn = fields.Str(validate=validate.Regexp(r'^\d{10}(\d{3})?$'))
published_year = fields.Int(validate=validate.Range(min=1000, max=2100))
genre = fields.Str(validate=validate.OneOf(['Fiction', 'Non-Fiction', 'Science', 'History']))
price = fields.Float(validate=validate.Range(min=0))
class Meta:
ordered = True
book_schema = BookSchema()
books_schema = BookSchema(many=True)
Using Schemas in Routes
@app.route('/api/books', methods=['POST'])
def create_book():
"""Create book with validation."""
try:
# Validate and deserialize input
data = book_schema.load(request.get_json())
# Create book
book = Book(**data)
db.session.add(book)
db.session.commit()
# Serialize and return
return jsonify(book_schema.dump(book)), 201
except ValidationError as err:
return jsonify({"errors": err.messages}), 400
Custom Validation
from marshmallow import validates, ValidationError
class UserSchema(Schema):
email = fields.Email(required=True)
username = fields.Str(required=True, validate=validate.Length(min=3, max=50))
password = fields.Str(required=True, load_only=True)
@validates('username')
def validate_username(self, value):
"""Custom validation for username."""
if User.query.filter_by(username=value).first():
raise ValidationError('Username already exists')
@validates('email')
def validate_email(self, value):
"""Custom validation for email."""
if User.query.filter_by(email=value).first():
raise ValidationError('Email already registered')
Error Handling
Global Error Handlers
from werkzeug.exceptions import HTTPException
@app.errorhandler(404)
def not_found_error(error):
"""Handle 404 errors."""
return jsonify({
"error": "Not Found",
"message": "The requested resource was not found"
}), 404
@app.errorhandler(500)
def internal_error(error):
"""Handle 500 errors."""
db.session.rollback() # Rollback any failed transactions
return jsonify({
"error": "Internal Server Error",
"message": "An unexpected error occurred"
}), 500
@app.errorhandler(ValidationError)
def validation_error(error):
"""Handle Marshmallow validation errors."""
return jsonify({
"error": "Validation Error",
"messages": error.messages
}), 400
@app.errorhandler(HTTPException)
def handle_http_exception(e):
"""Handle all HTTP exceptions."""
return jsonify({
"error": e.name,
"message": e.description
}), e.code
### Custom 404 Behavior (SPA Pattern)
In Single Page Applications (SPAs) like React or Vue, you often want the backend to ignore unknown routes and let the frontend handle them.
```python
from flask import redirect, url_for
@app.errorhandler(404)
def handle_404(e):
"""Redirect unknown routes to the home page (SPA entry point)."""
# For APIs, you usually return JSON.
# For SPAs, you might want to redirect to the frontend route.
return redirect(url_for('home'))
### Custom Exceptions
```python
class APIError(Exception):
"""Base class for API exceptions."""
status_code = 400
def __init__(self, message, status_code=None, payload=None):
super().__init__()
self.message = message
if status_code is not None:
self.status_code = status_code
self.payload = payload
def to_dict(self):
rv = dict(self.payload or ())
rv['message'] = self.message
return rv
class ResourceNotFoundError(APIError):
"""Raised when resource is not found."""
status_code = 404
class UnauthorizedError(APIError):
"""Raised when user is not authorized."""
status_code = 401
@app.errorhandler(APIError)
def handle_api_error(error):
"""Handle custom API errors."""
response = jsonify(error.to_dict())
response.status_code = error.status_code
return response
# Usage in routes
@app.route('/api/books/<int:book_id>')
def get_book(book_id):
book = Book.query.get(book_id)
if not book:
raise ResourceNotFoundError(f"Book with id {book_id} not found")
return jsonify(book.to_dict())
Application Factory Pattern
Complete Application Factory
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_cors import CORS
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flask_caching import Cache
# Initialize extensions
db = SQLAlchemy()
migrate = Migrate()
cors = CORS()
limiter = Limiter(key_func=get_remote_address)
cache = Cache()
def create_app(config_name='development'):
"""
Application factory.
Args:
config_name: Configuration to use (development, production, testing)
Returns:
Configured Flask application
"""
app = Flask(__name__)
# Load configuration
app.config.from_object(f'app.config.{config_name.capitalize()}Config')
# Initialize extensions with app
db.init_app(app)
migrate.init_app(app, db)
cors.init_app(app, resources={r"/api/*": {"origins": "*"}})
limiter.init_app(app)
cache.init_app(app)
# Register blueprints
from app.auth import auth_bp
from app.books import books_bp
from app.users import users_bp
app.register_blueprint(auth_bp, url_prefix='/api/auth')
app.register_blueprint(books_bp, url_prefix='/api/books')
app.register_blueprint(users_bp, url_prefix='/api/users')
# Register error handlers
register_error_handlers(app)
# Register CLI commands
register_commands(app)
return app
def register_error_handlers(app):
"""Register error handlers."""
@app.errorhandler(404)
def not_found(error):
return jsonify({"error": "Not found"}), 404
@app.errorhandler(500)
def internal_error(error):
db.session.rollback()
return jsonify({"error": "Internal server error"}), 500
def register_commands(app):
"""Register CLI commands."""
@app.cli.command()
def init_db():
"""Initialize the database."""
db.create_all()
print("Database initialized")
@app.cli.command()
def seed_db():
"""Seed the database with sample data."""
# Seeding logic
print("Database seeded")
Configuration
config.py:
import os
class Config:
"""Base configuration."""
SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key')
SQLALCHEMY_TRACK_MODIFICATIONS = False
# Rate limiting
RATELIMIT_STORAGE_URL = os.getenv('REDIS_URL', 'memory://')
# Caching
CACHE_TYPE = 'SimpleCache'
CACHE_DEFAULT_TIMEOUT = 300
class DevelopmentConfig(Config):
"""Development configuration."""
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.getenv(
'DEV_DATABASE_URL',
'sqlite:///dev.db'
)
SQLALCHEMY_ECHO = True
class ProductionConfig(Config):
"""Production configuration."""
DEBUG = False
SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL')
# Redis for caching and rate limiting
CACHE_TYPE = 'RedisCache'
CACHE_REDIS_URL = os.getenv('REDIS_URL')
RATELIMIT_STORAGE_URL = os.getenv('REDIS_URL')
class TestingConfig(Config):
"""Testing configuration."""
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
WTF_CSRF_ENABLED = False
JWT Authentication
Setup
pip install python-jose[cryptography]
Token Generation and Verification
from jose import jwt, JWTError
from datetime import datetime, timedelta
from flask import current_app
def generate_token(user):
"""Generate JWT token for user."""
payload = {
'user_id': user.id,
'username': user.username,
'exp': datetime.utcnow() + timedelta(hours=24),
'iat': datetime.utcnow()
}
token = jwt.encode(
payload,
current_app.config['SECRET_KEY'],
algorithm='HS256'
)
return token
def verify_token(token):
"""Verify JWT token."""
try:
# Remove "Bearer " prefix if present
if token.startswith('Bearer '):
token = token[7:]
payload = jwt.decode(
token,
current_app.config['SECRET_KEY'],
algorithms=['HS256']
)
return payload['user_id']
except JWTError:
return None
Authentication Decorator
from functools import wraps
from flask import request, jsonify
def token_required(f):
"""Decorator to require valid JWT token."""
@wraps(f)
def decorated_function(*args, **kwargs):
token = request.headers.get('Authorization')
if not token:
return jsonify({"error": "Token is missing"}), 401
user_id = verify_token(token)
if not user_id:
return jsonify({"error": "Invalid or expired token"}), 401
# Load user
user = User.query.get(user_id)
if not user:
return jsonify({"error": "User not found"}), 401
# Pass user to route function
return f(current_user=user, *args, **kwargs)
return decorated_function
# Usage
@app.route('/api/profile')
@token_required
def get_profile(current_user):
return jsonify(current_user.to_dict())
CORS Configuration
Basic CORS Setup
from flask_cors import CORS
app = Flask(__name__)
# Allow all origins (development only)
CORS(app)
# Restrict to specific origins (production)
CORS(app, origins=['https://example.com', 'https://app.example.com'])
# Configure per resource
CORS(app, resources={
r"/api/*": {
"origins": ["https://example.com"],
"methods": ["GET", "POST", "PUT", "DELETE"],
"allow_headers": ["Content-Type", "Authorization"]
}
})
Blueprint-Specific CORS
from flask_cors import cross_origin
@books_bp.route('/books', methods=['GET'])
@cross_origin(origins=['https://example.com'])
def get_books():
return jsonify([])
Production Best Practices
Environment Variables
.env:
FLASK_APP=run.py
FLASK_ENV=production
SECRET_KEY=your-secret-key-here
DATABASE_URL=postgresql://user:password@localhost/dbname
REDIS_URL=redis://localhost:6379/0
Logging Configuration
import logging
from logging.handlers import RotatingFileHandler
import os
def configure_logging(app):
"""Configure application logging."""
if not app.debug:
# Create logs directory
if not os.path.exists('logs'):
os.mkdir('logs')
# File handler
file_handler = RotatingFileHandler(
'logs/app.log',
maxBytes=10240000, # 10MB
backupCount=10
)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s %(levelname)s: %(message)s '
'[in %(pathname)s:%(lineno)d]'
))
file_handler.setLevel(logging.INFO)
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
app.logger.info('Application startup')
# In create_app()
configure_logging(app)
Health Check Endpoint
@app.route('/health')
@limiter.exempt
def health_check():
"""Health check endpoint for load balancers."""
try:
# Check database
db.session.execute('SELECT 1')
return jsonify({
"status": "healthy",
"database": "connected",
"timestamp": datetime.utcnow().isoformat()
}), 200
except Exception as e:
return jsonify({
"status": "unhealthy",
"error": str(e)
}), 500
Summary
Key Takeaways
- Rate Limiting: Protect your API from abuse
- Caching: Improve performance by caching expensive operations
- Pagination: Handle large datasets efficiently
- Blueprints: Organize code into modular components
- Validation: Use schemas for request validation
- Error Handling: Provide consistent error responses
- Factory Pattern: Use for flexible application configuration
- JWT Auth: Secure routes with token-based authentication
Best Practices Checklist
- Implement rate limiting on public endpoints
- Cache expensive database queries
- Use pagination for list endpoints
- Organize routes into blueprints
- Validate all request data
- Handle errors consistently
- Use application factory pattern
- Implement JWT authentication
- Configure CORS appropriately
- Set up proper logging
- Add health check endpoint
- Use environment variables for config
Related Resources
- Flask REST API Development - Basic Flask API patterns
- API Authentication - Authentication strategies
- OAuth2 and Token Management - OAuth2 implementation
- SQLAlchemy Advanced Patterns - Database optimization
- Error Handling - Exception handling