Full Stack Learning Hub

Comprehensive guides, cheat sheets, and code examples for full stack development.

View on GitHub

Flask Advanced Features Guide

Table of Contents

  1. Rate Limiting with Flask-Limiter
  2. Caching with Flask-Caching
  3. Pagination Patterns
  4. Advanced Blueprint Architecture
  5. Request Validation
  6. Error Handling
  7. Application Factory Pattern
  8. JWT Authentication
  9. CORS Configuration
  10. 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:


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
        }
    })
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

  1. Rate Limiting: Protect your API from abuse
  2. Caching: Improve performance by caching expensive operations
  3. Pagination: Handle large datasets efficiently
  4. Blueprints: Organize code into modular components
  5. Validation: Use schemas for request validation
  6. Error Handling: Provide consistent error responses
  7. Factory Pattern: Use for flexible application configuration
  8. JWT Auth: Secure routes with token-based authentication

Best Practices Checklist

Back to Main