A REST API (Representational State Transfer Application Programming Interface) is a software architectural style that defines a set of constraints for building web services. REST APIs communicate over HTTP, use standard methods (GET, POST, PUT, DELETE) to perform operations on resources, and exchange data in a structured format — almost always JSON in modern applications.
The core RESTful principles are:
/api/books/42 identifies a book, not an action.Flask is an excellent choice for building REST APIs because it gives you control without overhead. There is no built-in ORM you are forced to use, no admin panel you have to disable, and no opinionated project layout. You choose your serialization library, your database layer, and your authentication strategy. For microservices and lightweight APIs, Flask’s minimal footprint and fast startup time are significant advantages over heavier frameworks.
This tutorial walks through building production-quality REST APIs with Flask, from your first endpoint to a fully authenticated, database-backed, tested API. Every code example is written as you would actually write it in a professional codebase.
A well-organized Flask API project separates concerns from the start. Here is the structure we will build toward throughout this tutorial:
flask-rest-api/ ├── app/ │ ├── __init__.py # Application factory │ ├── config.py # Configuration classes │ ├── models.py # SQLAlchemy models │ ├── schemas.py # Marshmallow schemas │ ├── routes/ │ │ ├── __init__.py │ │ ├── books.py # Book endpoints │ │ └── auth.py # Authentication endpoints │ ├── errors.py # Custom error handlers │ └── extensions.py # Extension instances (db, ma, jwt) ├── tests/ │ ├── conftest.py # Test fixtures │ ├── test_books.py # Book endpoint tests │ └── test_auth.py # Auth endpoint tests ├── requirements.txt ├── run.py # Entry point └── .env # Environment variables (not committed)
Create a virtual environment and install the packages we will use throughout this tutorial:
python3 -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
pip install flask flask-sqlalchemy flask-marshmallow marshmallow-sqlalchemy \
flask-jwt-extended flask-cors python-dotenv pytest
Pin your dependencies immediately:
pip freeze > requirements.txt
Each of these packages serves a specific role: flask-sqlalchemy for database integration, flask-marshmallow and marshmallow-sqlalchemy for serialization, flask-jwt-extended for JWT authentication, flask-cors for cross-origin requests, and pytest for testing.
Let us start with the simplest possible Flask API — a single endpoint that returns JSON:
from flask import Flask, jsonify
app = Flask(__name__)
@app.route('/api/health', methods=['GET'])
def health_check():
return jsonify({
'status': 'healthy',
'service': 'flask-rest-api',
'version': '1.0.0'
}), 200
if __name__ == '__main__':
app.run(debug=True)
Key points here:
jsonify() converts a Python dictionary to a JSON response with the correct Content-Type: application/json header. Never return raw strings from an API endpoint.200 is the default, but being explicit makes your intent clear.methods=['GET'] argument restricts this route to GET requests only. Without it, Flask defaults to GET.Run it and test:
python run.py # In another terminal: curl http://localhost:5000/api/health
{
"status": "healthy",
"service": "flask-rest-api",
"version": "1.0.0"
}
REST APIs communicate intent through status codes. Here are the ones you will use constantly:
200 OK — Successful GET, PUT, or DELETE201 Created — Successful POST that created a resource204 No Content — Successful DELETE with no response body400 Bad Request — Client sent invalid data401 Unauthorized — Authentication required or failed403 Forbidden — Authenticated but not authorized404 Not Found — Resource does not exist409 Conflict — Resource already exists (e.g., duplicate email)422 Unprocessable Entity — Request is well-formed but semantically invalid500 Internal Server Error — Something broke on the serverCRUD (Create, Read, Update, Delete) maps directly to HTTP methods. Here is a complete in-memory example before we add a database:
from flask import Flask, jsonify, request, abort
app = Flask(__name__)
# In-memory data store (replaced by a database later)
books = [
{'id': 1, 'title': 'Clean Code', 'author': 'Robert C. Martin', 'year': 2008},
{'id': 2, 'title': 'The Pragmatic Programmer', 'author': 'David Thomas', 'year': 1999},
]
next_id = 3
@app.route('/api/books', methods=['GET'])
def get_books():
"""List all books."""
return jsonify({
'books': books,
'total': len(books)
}), 200
@app.route('/api/books/<int:book_id>', methods=['GET'])
def get_book(book_id):
"""Get a single book by ID."""
book = next((b for b in books if b['id'] == book_id), None)
if book is None:
return jsonify({'error': 'Book not found'}), 404
return jsonify(book), 200
@app.route('/api/books', methods=['POST'])
def create_book():
"""Create a new book."""
global next_id
if not request.is_json:
return jsonify({'error': 'Content-Type must be application/json'}), 415
data = request.get_json()
# Validate required fields
required_fields = ['title', 'author']
missing = [f for f in required_fields if f not in data]
if missing:
return jsonify({
'error': 'Missing required fields',
'missing_fields': missing
}), 400
book = {
'id': next_id,
'title': data['title'],
'author': data['author'],
'year': data.get('year') # Optional field
}
next_id += 1
books.append(book)
return jsonify(book), 201
@app.route('/api/books/<int:book_id>', methods=['PUT'])
def update_book(book_id):
"""Update an existing book."""
book = next((b for b in books if b['id'] == book_id), None)
if book is None:
return jsonify({'error': 'Book not found'}), 404
if not request.is_json:
return jsonify({'error': 'Content-Type must be application/json'}), 415
data = request.get_json()
book['title'] = data.get('title', book['title'])
book['author'] = data.get('author', book['author'])
book['year'] = data.get('year', book['year'])
return jsonify(book), 200
@app.route('/api/books/<int:book_id>', methods=['DELETE'])
def delete_book(book_id):
"""Delete a book."""
global books
original_length = len(books)
books = [b for b in books if b['id'] != book_id]
if len(books) == original_length:
return jsonify({'error': 'Book not found'}), 404
return '', 204
if __name__ == '__main__':
app.run(debug=True)
Test each operation with curl:
# List all books
curl http://localhost:5000/api/books
# Get a single book
curl http://localhost:5000/api/books/1
# Create a new book
curl -X POST http://localhost:5000/api/books \
-H "Content-Type: application/json" \
-d '{"title": "Design Patterns", "author": "Gang of Four", "year": 1994}'
# Update a book
curl -X PUT http://localhost:5000/api/books/1 \
-H "Content-Type: application/json" \
-d '{"title": "Clean Code (Revised)"}'
# Delete a book
curl -X DELETE http://localhost:5000/api/books/2
Notice the patterns: POST returns 201 Created with the new resource in the body, DELETE returns 204 No Content with an empty body, and every error returns a JSON object with an error key.
Flask provides several ways to access incoming data. Understanding when to use each one prevents subtle bugs.
request.get_json()@app.route('/api/books', methods=['POST'])
def create_book():
# request.get_json() returns None if Content-Type is not application/json
data = request.get_json()
if data is None:
return jsonify({'error': 'Request body must be JSON'}), 400
# request.get_json(silent=True) suppresses parsing errors
# request.get_json(force=True) ignores Content-Type header
title = data.get('title') # Safe — returns None if missing
# title = data['title'] # Dangerous — raises KeyError if missing
request.args@app.route('/api/books', methods=['GET'])
def get_books():
# GET /api/books?page=2&per_page=10&sort=title
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 20, type=int)
sort_by = request.args.get('sort', 'id')
# The type parameter handles conversion and returns the default
# if conversion fails — no try/except needed
# GET /api/books?page=abc -> page = 1 (default)
# Flask converters: int, float, string (default), path, uuid
@app.route('/api/books/<int:book_id>')
def get_book(book_id):
# book_id is already an int — Flask returns 404 if conversion fails
pass
@app.route('/api/users/<uuid:user_id>')
def get_user(user_id):
# user_id is a UUID object
pass
def validate_book_data(data, required=True):
"""Validate book input data. Returns (cleaned_data, errors)."""
errors = {}
if required and 'title' not in data:
errors['title'] = 'Title is required'
elif 'title' in data:
if not isinstance(data['title'], str):
errors['title'] = 'Title must be a string'
elif len(data['title'].strip()) == 0:
errors['title'] = 'Title cannot be empty'
elif len(data['title']) > 200:
errors['title'] = 'Title must be 200 characters or fewer'
if 'year' in data:
if not isinstance(data['year'], int):
errors['year'] = 'Year must be an integer'
elif data['year'] < 0 or data['year'] > 2030:
errors['year'] = 'Year must be between 0 and 2030'
if errors:
return None, errors
cleaned = {
'title': data.get('title', '').strip(),
'author': data.get('author', '').strip(),
'year': data.get('year'),
}
return cleaned, None
@app.route('/api/books', methods=['POST'])
def create_book():
data = request.get_json()
if data is None:
return jsonify({'error': 'Request body must be JSON'}), 400
cleaned, errors = validate_book_data(data, required=True)
if errors:
return jsonify({
'error': 'Validation failed',
'details': errors
}), 422
# Proceed with cleaned data
# ...
The validation function returns a tuple of cleaned data and errors. This pattern separates validation from endpoint logic and makes it easy to reuse across POST and PUT handlers. We will replace this manual validation with marshmallow schemas later, but understanding the manual approach first is important.
Consistent response formatting is one of the hallmarks of a well-designed API. Your clients should be able to predict the shape of every response without reading documentation for each endpoint.
from flask import Flask, jsonify
from functools import wraps
app = Flask(__name__)
def api_response(data=None, message=None, status_code=200, **kwargs):
"""Build a consistent API response."""
response = {
'success': 200 <= status_code < 300,
}
if message:
response['message'] = message
if data is not None:
response['data'] = data
# Include any extra kwargs (e.g., pagination metadata)
response.update(kwargs)
return jsonify(response), status_code
# Usage in endpoints
@app.route('/api/books', methods=['GET'])
def get_books():
books = get_all_books()
return api_response(data=books, message='Books retrieved successfully')
@app.route('/api/books', methods=['POST'])
def create_book():
# ... validation ...
book = save_book(data)
return api_response(data=book, message='Book created', status_code=201)
@app.route('/api/books/<int:book_id>', methods=['DELETE'])
def delete_book(book_id):
# ... deletion logic ...
return api_response(message='Book deleted', status_code=200)
def error_response(message, status_code, details=None):
"""Build a consistent error response."""
response = {
'success': False,
'error': {
'message': message,
'code': status_code,
}
}
if details:
response['error']['details'] = details
return jsonify(response), status_code
# Usage
@app.route('/api/books/<int:book_id>', methods=['GET'])
def get_book(book_id):
book = find_book(book_id)
if not book:
return error_response('Book not found', 404)
return api_response(data=book)
@app.route('/api/books', methods=['GET'])
def get_books():
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 20, type=int)
# Clamp per_page to prevent abuse
per_page = min(per_page, 100)
# With SQLAlchemy (shown later):
pagination = Book.query.paginate(page=page, per_page=per_page, error_out=False)
return api_response(
data=[book.to_dict() for book in pagination.items],
pagination={
'page': pagination.page,
'per_page': pagination.per_page,
'total_pages': pagination.pages,
'total_items': pagination.total,
'has_next': pagination.has_next,
'has_prev': pagination.has_prev,
}
)
The response for a paginated request looks like this:
{
"success": true,
"data": [
{"id": 1, "title": "Clean Code", "author": "Robert C. Martin"},
{"id": 2, "title": "The Pragmatic Programmer", "author": "David Thomas"}
],
"pagination": {
"page": 1,
"per_page": 20,
"total_pages": 5,
"total_items": 93,
"has_next": true,
"has_prev": false
}
}
When your endpoints handle multiple HTTP methods, function-based views can get cluttered. Flask's MethodView organizes CRUD operations into a single class where each HTTP method gets its own method:
from flask import Flask, jsonify, request
from flask.views import MethodView
app = Flask(__name__)
class BookAPI(MethodView):
"""Handles /api/books and /api/books/<id>"""
def get(self, book_id=None):
if book_id is None:
# GET /api/books — list all
books = get_all_books()
return jsonify({'books': books}), 200
else:
# GET /api/books/42 — get one
book = find_book(book_id)
if not book:
return jsonify({'error': 'Book not found'}), 404
return jsonify(book), 200
def post(self):
"""POST /api/books — create"""
data = request.get_json()
if not data or 'title' not in data:
return jsonify({'error': 'Title is required'}), 400
book = create_book(data)
return jsonify(book), 201
def put(self, book_id):
"""PUT /api/books/42 — update"""
book = find_book(book_id)
if not book:
return jsonify({'error': 'Book not found'}), 404
data = request.get_json()
updated = update_book(book_id, data)
return jsonify(updated), 200
def delete(self, book_id):
"""DELETE /api/books/42 — delete"""
book = find_book(book_id)
if not book:
return jsonify({'error': 'Book not found'}), 404
delete_book(book_id)
return '', 204
# Register the view with URL rules
book_view = BookAPI.as_view('book_api')
app.add_url_rule('/api/books', defaults={'book_id': None},
view_func=book_view, methods=['GET'])
app.add_url_rule('/api/books', view_func=book_view, methods=['POST'])
app.add_url_rule('/api/books/<int:book_id>', view_func=book_view,
methods=['GET', 'PUT', 'DELETE'])
The advantage of MethodView is organization. All operations for a resource live in one class. When you add decorators like authentication, you apply them once instead of to five separate functions. The registration at the bottom is more verbose, but it is done once and is explicit about which methods are allowed on which URL patterns.
By default, Flask returns HTML error pages. For an API, every response — including errors — must be JSON. Register custom error handlers to enforce this:
from flask import Flask, jsonify
from werkzeug.exceptions import HTTPException
app = Flask(__name__)
# Custom exception class for API errors
class APIError(Exception):
"""Custom exception for API-specific errors."""
def __init__(self, message, status_code=400, details=None):
super().__init__()
self.message = message
self.status_code = status_code
self.details = details
def to_dict(self):
response = {
'success': False,
'error': {
'message': self.message,
'code': self.status_code,
}
}
if self.details:
response['error']['details'] = self.details
return response
@app.errorhandler(APIError)
def handle_api_error(error):
"""Handle custom API errors."""
return jsonify(error.to_dict()), error.status_code
@app.errorhandler(HTTPException)
def handle_http_error(error):
"""Convert all HTTP exceptions to JSON."""
return jsonify({
'success': False,
'error': {
'message': error.description,
'code': error.code,
}
}), error.code
@app.errorhandler(Exception)
def handle_unexpected_error(error):
"""Catch-all for unhandled exceptions."""
app.logger.error(f'Unhandled exception: {error}', exc_info=True)
return jsonify({
'success': False,
'error': {
'message': 'An unexpected error occurred',
'code': 500,
}
}), 500
# Usage in endpoints
@app.route('/api/books/<int:book_id>', methods=['GET'])
def get_book(book_id):
book = find_book(book_id)
if not book:
raise APIError('Book not found', status_code=404)
return jsonify(book), 200
@app.route('/api/books', methods=['POST'])
def create_book():
data = request.get_json()
errors = validate(data)
if errors:
raise APIError(
'Validation failed',
status_code=422,
details=errors
)
The three-layer approach is important: APIError handles your intentional errors, HTTPException catches Flask's built-in errors (like 404 for unknown routes or 405 for wrong methods), and the generic Exception handler prevents stack traces from leaking to clients in production. Always log the actual error server-side before returning a generic message.
APIs need to verify who is making requests. We will cover two approaches: API keys for simple service-to-service authentication, and JWT tokens for user authentication.
from functools import wraps
from flask import request, jsonify
API_KEYS = {
'sk-abc123': {'name': 'Mobile App', 'permissions': ['read', 'write']},
'sk-def456': {'name': 'Analytics Service', 'permissions': ['read']},
}
def require_api_key(f):
@wraps(f)
def decorated(*args, **kwargs):
api_key = request.headers.get('X-API-Key')
if not api_key:
return jsonify({'error': 'API key is required'}), 401
if api_key not in API_KEYS:
return jsonify({'error': 'Invalid API key'}), 401
# Attach client info to the request context
request.api_client = API_KEYS[api_key]
return f(*args, **kwargs)
return decorated
def require_permission(permission):
def decorator(f):
@wraps(f)
def decorated(*args, **kwargs):
if permission not in request.api_client.get('permissions', []):
return jsonify({'error': 'Insufficient permissions'}), 403
return f(*args, **kwargs)
return decorated
return decorator
@app.route('/api/books', methods=['GET'])
@require_api_key
@require_permission('read')
def get_books():
return jsonify({'books': books})
JWT (JSON Web Token) is the standard for stateless user authentication in REST APIs. Flask-JWT-Extended makes this straightforward:
from flask import Flask, jsonify, request
from flask_jwt_extended import (
JWTManager, create_access_token, create_refresh_token,
jwt_required, get_jwt_identity, get_jwt
)
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import timedelta
app = Flask(__name__)
# JWT Configuration
app.config['JWT_SECRET_KEY'] = 'your-secret-key-change-in-production'
app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(hours=1)
app.config['JWT_REFRESH_TOKEN_EXPIRES'] = timedelta(days=30)
jwt = JWTManager(app)
# In-memory user store (use a database in production)
users = {}
@app.route('/api/auth/register', methods=['POST'])
def register():
data = request.get_json()
if not data or not data.get('email') or not data.get('password'):
return jsonify({'error': 'Email and password are required'}), 400
if data['email'] in users:
return jsonify({'error': 'User already exists'}), 409
users[data['email']] = {
'email': data['email'],
'password': generate_password_hash(data['password']),
'role': 'user',
}
return jsonify({'message': 'User registered successfully'}), 201
@app.route('/api/auth/login', methods=['POST'])
def login():
data = request.get_json()
if not data or not data.get('email') or not data.get('password'):
return jsonify({'error': 'Email and password are required'}), 400
user = users.get(data['email'])
if not user or not check_password_hash(user['password'], data['password']):
return jsonify({'error': 'Invalid email or password'}), 401
# Create tokens with user identity and additional claims
access_token = create_access_token(
identity=data['email'],
additional_claims={'role': user['role']}
)
refresh_token = create_refresh_token(identity=data['email'])
return jsonify({
'access_token': access_token,
'refresh_token': refresh_token,
'token_type': 'Bearer',
}), 200
@app.route('/api/auth/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
"""Get a new access token using a refresh token."""
identity = get_jwt_identity()
access_token = create_access_token(identity=identity)
return jsonify({'access_token': access_token}), 200
@app.route('/api/books', methods=['POST'])
@jwt_required()
def create_book():
"""Protected endpoint — requires a valid JWT."""
current_user = get_jwt_identity()
claims = get_jwt()
# Access role from token claims
if claims.get('role') != 'admin':
return jsonify({'error': 'Admin access required'}), 403
data = request.get_json()
# ... create book ...
return jsonify({'message': 'Book created', 'created_by': current_user}), 201
# Custom JWT error handlers
@jwt.expired_token_loader
def expired_token_callback(jwt_header, jwt_payload):
return jsonify({
'error': 'Token has expired',
'code': 'token_expired'
}), 401
@jwt.invalid_token_loader
def invalid_token_callback(error):
return jsonify({
'error': 'Invalid token',
'code': 'token_invalid'
}), 401
@jwt.unauthorized_loader
def missing_token_callback(error):
return jsonify({
'error': 'Authorization token is required',
'code': 'token_missing'
}), 401
Test the JWT flow:
# Register
curl -X POST http://localhost:5000/api/auth/register \
-H "Content-Type: application/json" \
-d '{"email": "admin@example.com", "password": "securepassword"}'
# Login — get tokens
curl -X POST http://localhost:5000/api/auth/login \
-H "Content-Type: application/json" \
-d '{"email": "admin@example.com", "password": "securepassword"}'
# Use the access token
curl -X POST http://localhost:5000/api/books \
-H "Content-Type: application/json" \
-H "Authorization: Bearer eyJhbGciOiJIUzI1NiIs..." \
-d '{"title": "New Book", "author": "Author Name"}'
# Refresh the token when it expires
curl -X POST http://localhost:5000/api/auth/refresh \
-H "Authorization: Bearer <refresh_token>"
An in-memory list is fine for learning, but real APIs need a database. Flask-SQLAlchemy provides an elegant ORM layer:
app/extensions.py)from flask_sqlalchemy import SQLAlchemy from flask_marshmallow import Marshmallow from flask_jwt_extended import JWTManager from flask_cors import CORS db = SQLAlchemy() ma = Marshmallow() jwt = JWTManager() cors = CORS()
app/config.py)import os
class Config:
SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-secret-key')
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY', 'jwt-dev-secret')
SQLALCHEMY_TRACK_MODIFICATIONS = False
class DevelopmentConfig(Config):
DEBUG = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///dev.db'
class TestingConfig(Config):
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
class ProductionConfig(Config):
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')
config = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,
'default': DevelopmentConfig,
}
app/__init__.py)from flask import Flask
from app.config import config
from app.extensions import db, ma, jwt, cors
def create_app(config_name='default'):
app = Flask(__name__)
app.config.from_object(config[config_name])
# Initialize extensions
db.init_app(app)
ma.init_app(app)
jwt.init_app(app)
cors.init_app(app)
# Register blueprints
from app.routes.books import books_bp
from app.routes.auth import auth_bp
app.register_blueprint(books_bp, url_prefix='/api')
app.register_blueprint(auth_bp, url_prefix='/api/auth')
# Register error handlers
from app.errors import register_error_handlers
register_error_handlers(app)
# Create tables
with app.app_context():
db.create_all()
return app
app/models.py)from datetime import datetime
from app.extensions import db
from werkzeug.security import generate_password_hash, check_password_hash
class Book(db.Model):
__tablename__ = 'books'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
author = db.Column(db.String(100), nullable=False)
isbn = db.Column(db.String(13), unique=True, nullable=True)
year = db.Column(db.Integer, nullable=True)
description = db.Column(db.Text, nullable=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
def __repr__(self):
return f'<Book {self.title}>'
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(120), unique=True, nullable=False, index=True)
password_hash = db.Column(db.String(256), nullable=False)
role = db.Column(db.String(20), default='user')
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
def __repr__(self):
return f'<User {self.email}>'
from flask import Blueprint, jsonify, request
from app.extensions import db
from app.models import Book
books_bp = Blueprint('books', __name__)
@books_bp.route('/books', methods=['GET'])
def get_books():
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 100)
# Filter by author if provided
query = Book.query
author = request.args.get('author')
if author:
query = query.filter(Book.author.ilike(f'%{author}%'))
# Sort
sort = request.args.get('sort', 'created_at')
order = request.args.get('order', 'desc')
if hasattr(Book, sort):
sort_column = getattr(Book, sort)
query = query.order_by(sort_column.desc() if order == 'desc' else sort_column.asc())
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
return jsonify({
'success': True,
'data': [book_to_dict(b) for b in pagination.items],
'pagination': {
'page': pagination.page,
'per_page': pagination.per_page,
'total_pages': pagination.pages,
'total_items': pagination.total,
}
}), 200
@books_bp.route('/books/<int:book_id>', methods=['GET'])
def get_book(book_id):
book = db.session.get(Book, book_id)
if not book:
return jsonify({'error': 'Book not found'}), 404
return jsonify({'success': True, 'data': book_to_dict(book)}), 200
@books_bp.route('/books', methods=['POST'])
def create_book():
data = request.get_json()
if not data:
return jsonify({'error': 'Request body must be JSON'}), 400
if not data.get('title') or not data.get('author'):
return jsonify({'error': 'Title and author are required'}), 400
# Check for duplicate ISBN
if data.get('isbn'):
existing = Book.query.filter_by(isbn=data['isbn']).first()
if existing:
return jsonify({'error': 'A book with this ISBN already exists'}), 409
book = Book(
title=data['title'],
author=data['author'],
isbn=data.get('isbn'),
year=data.get('year'),
description=data.get('description'),
)
db.session.add(book)
db.session.commit()
return jsonify({'success': True, 'data': book_to_dict(book)}), 201
@books_bp.route('/books/<int:book_id>', methods=['PUT'])
def update_book(book_id):
book = db.session.get(Book, book_id)
if not book:
return jsonify({'error': 'Book not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'Request body must be JSON'}), 400
book.title = data.get('title', book.title)
book.author = data.get('author', book.author)
book.isbn = data.get('isbn', book.isbn)
book.year = data.get('year', book.year)
book.description = data.get('description', book.description)
db.session.commit()
return jsonify({'success': True, 'data': book_to_dict(book)}), 200
@books_bp.route('/books/<int:book_id>', methods=['DELETE'])
def delete_book(book_id):
book = db.session.get(Book, book_id)
if not book:
return jsonify({'error': 'Book not found'}), 404
db.session.delete(book)
db.session.commit()
return jsonify({'success': True, 'message': 'Book deleted'}), 200
def book_to_dict(book):
return {
'id': book.id,
'title': book.title,
'author': book.author,
'isbn': book.isbn,
'year': book.year,
'description': book.description,
'created_at': book.created_at.isoformat(),
'updated_at': book.updated_at.isoformat(),
}
Manual to_dict() methods work, but they do not validate input. Marshmallow handles both serialization (object to JSON) and deserialization (JSON to validated data) in a single schema definition:
from marshmallow import fields, validate, validates, ValidationError
from app.extensions import ma
from app.models import Book
class BookSchema(ma.SQLAlchemyAutoSchema):
class Meta:
model = Book
load_instance = True # Deserialize to model instances
include_fk = True
# Override auto-generated fields with validation rules
title = fields.String(
required=True,
validate=validate.Length(min=1, max=200, error='Title must be 1-200 characters')
)
author = fields.String(
required=True,
validate=validate.Length(min=1, max=100)
)
isbn = fields.String(
validate=validate.Length(equal=13, error='ISBN must be exactly 13 characters')
)
year = fields.Integer(
validate=validate.Range(min=0, max=2030)
)
description = fields.String(validate=validate.Length(max=2000))
# Read-only fields — included in output but ignored on input
created_at = fields.DateTime(dump_only=True)
updated_at = fields.DateTime(dump_only=True)
@validates('isbn')
def validate_isbn(self, value):
if value and not value.isdigit():
raise ValidationError('ISBN must contain only digits')
# Schema instances
book_schema = BookSchema()
books_schema = BookSchema(many=True)
# Usage in endpoints
@books_bp.route('/books', methods=['POST'])
def create_book():
data = request.get_json()
if not data:
return jsonify({'error': 'Request body must be JSON'}), 400
try:
book = book_schema.load(data, session=db.session)
except ValidationError as err:
return jsonify({
'error': 'Validation failed',
'details': err.messages
}), 422
db.session.add(book)
db.session.commit()
return jsonify({
'success': True,
'data': book_schema.dump(book)
}), 201
@books_bp.route('/books', methods=['GET'])
def get_books():
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 100)
pagination = Book.query.paginate(page=page, per_page=per_page, error_out=False)
return jsonify({
'success': True,
'data': books_schema.dump(pagination.items),
'pagination': {
'page': pagination.page,
'per_page': pagination.per_page,
'total_pages': pagination.pages,
'total_items': pagination.total,
}
}), 200
@books_bp.route('/books/<int:book_id>', methods=['PUT'])
def update_book(book_id):
book = db.session.get(Book, book_id)
if not book:
return jsonify({'error': 'Book not found'}), 404
data = request.get_json()
try:
# partial=True allows updating only some fields
updated_book = book_schema.load(data, instance=book, partial=True, session=db.session)
except ValidationError as err:
return jsonify({
'error': 'Validation failed',
'details': err.messages
}), 422
db.session.commit()
return jsonify({
'success': True,
'data': book_schema.dump(updated_book)
}), 200
Marshmallow gives you three critical capabilities in one place: it validates incoming data and returns structured error messages, it serializes model instances to dictionaries (and then JSON), and it deserializes JSON input directly into SQLAlchemy model instances. The partial=True parameter on PUT/PATCH is particularly useful — it makes all fields optional so clients only need to send the fields they want to update.
If your API is consumed by a frontend on a different domain, browsers will block requests unless you configure CORS headers. Flask-CORS makes this trivial:
from flask import Flask
from flask_cors import CORS
app = Flask(__name__)
# Option 1: Allow all origins (development only)
CORS(app)
# Option 2: Restrict to specific origins (production)
CORS(app, resources={
r"/api/*": {
"origins": [
"https://yourfrontend.com",
"https://admin.yourfrontend.com",
],
"methods": ["GET", "POST", "PUT", "DELETE", "OPTIONS"],
"allow_headers": ["Content-Type", "Authorization"],
"expose_headers": ["X-Total-Count"],
"max_age": 600, # Cache preflight for 10 minutes
}
})
# Option 3: Per-blueprint CORS
from flask_cors import cross_origin
@app.route('/api/public/data', methods=['GET'])
@cross_origin(origins='*')
def public_data():
return jsonify({'data': 'Anyone can access this'})
The key settings to understand:
origins — Which domains can make requests. Never use '*' in production if your API requires authentication.methods — Which HTTP methods are allowed.allow_headers — Which request headers the client can send. You must include Authorization if you use JWT tokens.expose_headers — Which response headers the client's JavaScript can read. Custom headers like pagination totals need to be explicitly exposed.max_age — How long the browser caches the preflight (OPTIONS) response. Higher values reduce preflight requests.Let us assemble everything into a complete, runnable application. This brings together the application factory, database models, marshmallow schemas, JWT authentication, error handling, CORS, and pagination.
run.py)from app import create_app
app = create_app('development')
if __name__ == '__main__':
app.run(debug=True, port=5000)
app/extensions.py)from flask_sqlalchemy import SQLAlchemy from flask_marshmallow import Marshmallow from flask_jwt_extended import JWTManager from flask_cors import CORS db = SQLAlchemy() ma = Marshmallow() jwt = JWTManager() cors = CORS()
app/__init__.py)from flask import Flask
from app.config import config
from app.extensions import db, ma, jwt, cors
def create_app(config_name='default'):
app = Flask(__name__)
app.config.from_object(config[config_name])
# Initialize extensions
db.init_app(app)
ma.init_app(app)
jwt.init_app(app)
cors.init_app(app, resources={r"/api/*": {"origins": "*"}})
# Register blueprints
from app.routes.books import books_bp
from app.routes.auth import auth_bp
app.register_blueprint(books_bp, url_prefix='/api')
app.register_blueprint(auth_bp, url_prefix='/api/auth')
# Register error handlers
from app.errors import register_error_handlers
register_error_handlers(app)
# Create database tables
with app.app_context():
db.create_all()
return app
app/models.py)from datetime import datetime
from app.extensions import db
from werkzeug.security import generate_password_hash, check_password_hash
class Book(db.Model):
__tablename__ = 'books'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
author = db.Column(db.String(100), nullable=False)
isbn = db.Column(db.String(13), unique=True, nullable=True)
year = db.Column(db.Integer, nullable=True)
description = db.Column(db.Text, nullable=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
def __repr__(self):
return f'<Book {self.title}>'
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(120), unique=True, nullable=False, index=True)
password_hash = db.Column(db.String(256), nullable=False)
role = db.Column(db.String(20), default='user')
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
def __repr__(self):
return f'<User {self.email}>'
app/schemas.py)from marshmallow import fields, validate, validates, ValidationError
from app.extensions import ma
from app.models import Book, User
class BookSchema(ma.SQLAlchemyAutoSchema):
class Meta:
model = Book
load_instance = True
include_fk = True
title = fields.String(required=True, validate=validate.Length(min=1, max=200))
author = fields.String(required=True, validate=validate.Length(min=1, max=100))
isbn = fields.String(validate=validate.Length(equal=13))
year = fields.Integer(validate=validate.Range(min=0, max=2030))
description = fields.String(validate=validate.Length(max=2000))
created_at = fields.DateTime(dump_only=True)
updated_at = fields.DateTime(dump_only=True)
@validates('isbn')
def validate_isbn(self, value):
if value and not value.isdigit():
raise ValidationError('ISBN must contain only digits')
class UserSchema(ma.SQLAlchemyAutoSchema):
class Meta:
model = User
load_instance = True
exclude = ('password_hash',)
email = fields.Email(required=True)
role = fields.String(dump_only=True)
created_at = fields.DateTime(dump_only=True)
class LoginSchema(ma.Schema):
email = fields.Email(required=True)
password = fields.String(required=True, validate=validate.Length(min=6))
class RegisterSchema(ma.Schema):
email = fields.Email(required=True)
password = fields.String(required=True, validate=validate.Length(min=8))
book_schema = BookSchema()
books_schema = BookSchema(many=True)
user_schema = UserSchema()
login_schema = LoginSchema()
register_schema = RegisterSchema()
app/errors.py)from flask import jsonify
from werkzeug.exceptions import HTTPException
class APIError(Exception):
def __init__(self, message, status_code=400, details=None):
super().__init__()
self.message = message
self.status_code = status_code
self.details = details
def to_dict(self):
response = {
'success': False,
'error': {'message': self.message, 'code': self.status_code}
}
if self.details:
response['error']['details'] = self.details
return response
def register_error_handlers(app):
@app.errorhandler(APIError)
def handle_api_error(error):
return jsonify(error.to_dict()), error.status_code
@app.errorhandler(HTTPException)
def handle_http_error(error):
return jsonify({
'success': False,
'error': {'message': error.description, 'code': error.code}
}), error.code
@app.errorhandler(Exception)
def handle_unexpected_error(error):
app.logger.error(f'Unhandled exception: {error}', exc_info=True)
return jsonify({
'success': False,
'error': {'message': 'An unexpected error occurred', 'code': 500}
}), 500
app/routes/auth.py)from flask import Blueprint, jsonify, request
from flask_jwt_extended import (
create_access_token, create_refresh_token,
jwt_required, get_jwt_identity
)
from marshmallow import ValidationError
from app.extensions import db
from app.models import User
from app.schemas import login_schema, register_schema, user_schema
auth_bp = Blueprint('auth', __name__)
@auth_bp.route('/register', methods=['POST'])
def register():
data = request.get_json()
if not data:
return jsonify({'error': 'Request body must be JSON'}), 400
try:
validated = register_schema.load(data)
except ValidationError as err:
return jsonify({'error': 'Validation failed', 'details': err.messages}), 422
if User.query.filter_by(email=validated['email']).first():
return jsonify({'error': 'Email already registered'}), 409
user = User(email=validated['email'])
user.set_password(validated['password'])
db.session.add(user)
db.session.commit()
return jsonify({
'success': True,
'message': 'User registered successfully',
'data': user_schema.dump(user)
}), 201
@auth_bp.route('/login', methods=['POST'])
def login():
data = request.get_json()
if not data:
return jsonify({'error': 'Request body must be JSON'}), 400
try:
validated = login_schema.load(data)
except ValidationError as err:
return jsonify({'error': 'Validation failed', 'details': err.messages}), 422
user = User.query.filter_by(email=validated['email']).first()
if not user or not user.check_password(validated['password']):
return jsonify({'error': 'Invalid email or password'}), 401
access_token = create_access_token(
identity=str(user.id),
additional_claims={'role': user.role, 'email': user.email}
)
refresh_token = create_refresh_token(identity=str(user.id))
return jsonify({
'success': True,
'data': {
'access_token': access_token,
'refresh_token': refresh_token,
'token_type': 'Bearer',
}
}), 200
@auth_bp.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
identity = get_jwt_identity()
access_token = create_access_token(identity=identity)
return jsonify({
'success': True,
'data': {'access_token': access_token}
}), 200
@auth_bp.route('/me', methods=['GET'])
@jwt_required()
def get_current_user():
user_id = get_jwt_identity()
user = db.session.get(User, int(user_id))
if not user:
return jsonify({'error': 'User not found'}), 404
return jsonify({
'success': True,
'data': user_schema.dump(user)
}), 200
app/routes/books.py)from flask import Blueprint, jsonify, request
from flask_jwt_extended import jwt_required, get_jwt
from marshmallow import ValidationError as MarshmallowValidationError
from app.extensions import db
from app.models import Book
from app.schemas import book_schema, books_schema
from app.errors import APIError
books_bp = Blueprint('books', __name__)
def admin_required(fn):
"""Decorator that requires admin role."""
from functools import wraps
@wraps(fn)
@jwt_required()
def wrapper(*args, **kwargs):
claims = get_jwt()
if claims.get('role') != 'admin':
raise APIError('Admin access required', 403)
return fn(*args, **kwargs)
return wrapper
@books_bp.route('/books', methods=['GET'])
def list_books():
"""List books with filtering, sorting, and pagination."""
page = request.args.get('page', 1, type=int)
per_page = min(request.args.get('per_page', 20, type=int), 100)
query = Book.query
# Filter by author
author = request.args.get('author')
if author:
query = query.filter(Book.author.ilike(f'%{author}%'))
# Filter by year
year = request.args.get('year', type=int)
if year:
query = query.filter_by(year=year)
# Search by title
search = request.args.get('search')
if search:
query = query.filter(Book.title.ilike(f'%{search}%'))
# Sort
sort = request.args.get('sort', 'created_at')
order = request.args.get('order', 'desc')
allowed_sorts = ['title', 'author', 'year', 'created_at']
if sort in allowed_sorts:
sort_col = getattr(Book, sort)
query = query.order_by(sort_col.desc() if order == 'desc' else sort_col.asc())
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
return jsonify({
'success': True,
'data': books_schema.dump(pagination.items),
'pagination': {
'page': pagination.page,
'per_page': pagination.per_page,
'total_pages': pagination.pages,
'total_items': pagination.total,
'has_next': pagination.has_next,
'has_prev': pagination.has_prev,
}
}), 200
@books_bp.route('/books/<int:book_id>', methods=['GET'])
def get_book(book_id):
"""Get a single book by ID."""
book = db.session.get(Book, book_id)
if not book:
raise APIError('Book not found', 404)
return jsonify({'success': True, 'data': book_schema.dump(book)}), 200
@books_bp.route('/books', methods=['POST'])
@jwt_required()
def create_book():
"""Create a new book. Requires authentication."""
data = request.get_json()
if not data:
raise APIError('Request body must be JSON', 400)
try:
book = book_schema.load(data, session=db.session)
except MarshmallowValidationError as err:
raise APIError('Validation failed', 422, details=err.messages)
# Check for duplicate ISBN
if book.isbn:
existing = Book.query.filter_by(isbn=book.isbn).first()
if existing:
raise APIError('A book with this ISBN already exists', 409)
db.session.add(book)
db.session.commit()
return jsonify({'success': True, 'data': book_schema.dump(book)}), 201
@books_bp.route('/books/<int:book_id>', methods=['PUT'])
@jwt_required()
def update_book(book_id):
"""Update a book. Requires authentication."""
book = db.session.get(Book, book_id)
if not book:
raise APIError('Book not found', 404)
data = request.get_json()
if not data:
raise APIError('Request body must be JSON', 400)
try:
updated_book = book_schema.load(
data, instance=book, partial=True, session=db.session
)
except MarshmallowValidationError as err:
raise APIError('Validation failed', 422, details=err.messages)
db.session.commit()
return jsonify({'success': True, 'data': book_schema.dump(updated_book)}), 200
@books_bp.route('/books/<int:book_id>', methods=['DELETE'])
@admin_required
def delete_book(book_id):
"""Delete a book. Requires admin role."""
book = db.session.get(Book, book_id)
if not book:
raise APIError('Book not found', 404)
db.session.delete(book)
db.session.commit()
return jsonify({'success': True, 'message': 'Book deleted'}), 200
Testing is non-negotiable for API development. Flask provides a test client that simulates HTTP requests without starting a server. Combined with pytest, you get fast, isolated tests.
tests/conftest.py)import pytest
from app import create_app
from app.extensions import db as _db
from app.models import User
@pytest.fixture(scope='session')
def app():
"""Create application for testing."""
app = create_app('testing')
return app
@pytest.fixture(scope='function')
def db(app):
"""Create a fresh database for each test."""
with app.app_context():
_db.create_all()
yield _db
_db.session.rollback()
_db.drop_all()
@pytest.fixture
def client(app, db):
"""Create a test client."""
return app.test_client()
@pytest.fixture
def auth_headers(client, db):
"""Register and login a user, return auth headers."""
# Register
client.post('/api/auth/register', json={
'email': 'test@example.com',
'password': 'testpassword123',
})
# Login
response = client.post('/api/auth/login', json={
'email': 'test@example.com',
'password': 'testpassword123',
})
token = response.get_json()['data']['access_token']
return {'Authorization': f'Bearer {token}'}
@pytest.fixture
def admin_headers(client, db, app):
"""Create an admin user and return auth headers."""
with app.app_context():
admin = User(email='admin@example.com', role='admin')
admin.set_password('adminpassword123')
_db.session.add(admin)
_db.session.commit()
response = client.post('/api/auth/login', json={
'email': 'admin@example.com',
'password': 'adminpassword123',
})
token = response.get_json()['data']['access_token']
return {'Authorization': f'Bearer {token}'}
tests/test_books.py)import pytest
class TestListBooks:
def test_list_books_empty(self, client):
response = client.get('/api/books')
assert response.status_code == 200
data = response.get_json()
assert data['success'] is True
assert data['data'] == []
assert data['pagination']['total_items'] == 0
def test_list_books_with_pagination(self, client, auth_headers):
# Create 25 books
for i in range(25):
client.post('/api/books', json={
'title': f'Book {i}',
'author': f'Author {i}',
}, headers=auth_headers)
# First page
response = client.get('/api/books?page=1&per_page=10')
data = response.get_json()
assert len(data['data']) == 10
assert data['pagination']['total_items'] == 25
assert data['pagination']['total_pages'] == 3
assert data['pagination']['has_next'] is True
def test_list_books_with_search(self, client, auth_headers):
client.post('/api/books', json={
'title': 'Python Crash Course',
'author': 'Eric Matthes',
}, headers=auth_headers)
client.post('/api/books', json={
'title': 'Clean Code',
'author': 'Robert Martin',
}, headers=auth_headers)
response = client.get('/api/books?search=python')
data = response.get_json()
assert len(data['data']) == 1
assert data['data'][0]['title'] == 'Python Crash Course'
class TestGetBook:
def test_get_existing_book(self, client, auth_headers):
# Create a book
create_resp = client.post('/api/books', json={
'title': 'Test Book',
'author': 'Test Author',
}, headers=auth_headers)
book_id = create_resp.get_json()['data']['id']
# Retrieve it
response = client.get(f'/api/books/{book_id}')
assert response.status_code == 200
assert response.get_json()['data']['title'] == 'Test Book'
def test_get_nonexistent_book(self, client):
response = client.get('/api/books/9999')
assert response.status_code == 404
class TestCreateBook:
def test_create_book_success(self, client, auth_headers):
response = client.post('/api/books', json={
'title': 'New Book',
'author': 'New Author',
'year': 2024,
'isbn': '9781234567890',
}, headers=auth_headers)
assert response.status_code == 201
data = response.get_json()
assert data['success'] is True
assert data['data']['title'] == 'New Book'
assert data['data']['isbn'] == '9781234567890'
def test_create_book_missing_title(self, client, auth_headers):
response = client.post('/api/books', json={
'author': 'Some Author',
}, headers=auth_headers)
assert response.status_code == 422
def test_create_book_without_auth(self, client):
response = client.post('/api/books', json={
'title': 'Unauthorized Book',
'author': 'Author',
})
assert response.status_code == 401
def test_create_book_duplicate_isbn(self, client, auth_headers):
# Create first book
client.post('/api/books', json={
'title': 'First Book',
'author': 'Author',
'isbn': '9781234567890',
}, headers=auth_headers)
# Try to create with same ISBN
response = client.post('/api/books', json={
'title': 'Second Book',
'author': 'Author',
'isbn': '9781234567890',
}, headers=auth_headers)
assert response.status_code == 409
class TestUpdateBook:
def test_update_book_partial(self, client, auth_headers):
# Create
create_resp = client.post('/api/books', json={
'title': 'Original Title',
'author': 'Original Author',
}, headers=auth_headers)
book_id = create_resp.get_json()['data']['id']
# Update only the title
response = client.put(f'/api/books/{book_id}', json={
'title': 'Updated Title',
}, headers=auth_headers)
assert response.status_code == 200
data = response.get_json()['data']
assert data['title'] == 'Updated Title'
assert data['author'] == 'Original Author' # Unchanged
class TestDeleteBook:
def test_delete_book_as_admin(self, client, auth_headers, admin_headers):
# Create as regular user
create_resp = client.post('/api/books', json={
'title': 'Delete Me',
'author': 'Author',
}, headers=auth_headers)
book_id = create_resp.get_json()['data']['id']
# Delete as admin
response = client.delete(f'/api/books/{book_id}', headers=admin_headers)
assert response.status_code == 200
# Verify deletion
get_resp = client.get(f'/api/books/{book_id}')
assert get_resp.status_code == 404
def test_delete_book_as_regular_user(self, client, auth_headers):
create_resp = client.post('/api/books', json={
'title': 'Cannot Delete',
'author': 'Author',
}, headers=auth_headers)
book_id = create_resp.get_json()['data']['id']
response = client.delete(f'/api/books/{book_id}', headers=auth_headers)
assert response.status_code == 403
Run the tests:
# Run all tests pytest tests/ -v # Run with coverage pytest tests/ -v --cov=app --cov-report=term-missing # Run a specific test class pytest tests/test_books.py::TestCreateBook -v
An API without documentation is an API nobody will use correctly. Two popular tools for Flask are flasgger (Swagger UI) and flask-smorest (OpenAPI 3.0). Here is a quick setup with flasgger:
pip install flasgger
from flask import Flask
from flasgger import Swagger
app = Flask(__name__)
# Swagger configuration
app.config['SWAGGER'] = {
'title': 'Book API',
'uiversion': 3,
'openapi': '3.0.0',
'description': 'A complete Book management REST API',
}
swagger = Swagger(app)
@app.route('/api/books', methods=['GET'])
def list_books():
"""List all books with pagination.
---
parameters:
- name: page
in: query
type: integer
default: 1
description: Page number
- name: per_page
in: query
type: integer
default: 20
description: Items per page (max 100)
- name: search
in: query
type: string
description: Search books by title
responses:
200:
description: A list of books
content:
application/json:
schema:
type: object
properties:
success:
type: boolean
data:
type: array
items:
$ref: '#/components/schemas/Book'
pagination:
$ref: '#/components/schemas/Pagination'
"""
# ... implementation ...
@app.route('/api/books', methods=['POST'])
def create_book():
"""Create a new book.
---
security:
- BearerAuth: []
requestBody:
required: true
content:
application/json:
schema:
type: object
required:
- title
- author
properties:
title:
type: string
example: "Clean Code"
author:
type: string
example: "Robert C. Martin"
isbn:
type: string
example: "9780132350884"
year:
type: integer
example: 2008
responses:
201:
description: Book created successfully
401:
description: Authentication required
422:
description: Validation error
"""
# ... implementation ...
After configuring flasgger, visit http://localhost:5000/apidocs to see the interactive Swagger UI where you can test endpoints directly from the browser. For larger projects, consider flask-smorest, which integrates more tightly with marshmallow schemas and generates OpenAPI specs automatically from your existing schema definitions.
These are the mistakes I see most often in Flask API codebases, and they tend to surface in production rather than development.
Flask returns HTML for 404, 405, and 500 errors by default. If your client expects JSON and gets HTML, parsing fails silently or throws confusing errors.
# BAD: Flask returns HTML for unknown routes
# GET /api/nonexistent -> <h1>404 Not Found</h1>
# GOOD: Register custom error handlers (as shown in section 7)
@app.errorhandler(404)
def not_found(error):
return jsonify({'error': 'Resource not found'}), 404
@app.errorhandler(405)
def method_not_allowed(error):
return jsonify({'error': 'Method not allowed'}), 405
Your API works perfectly in Postman but fails from a browser. The error message in the browser console says "blocked by CORS policy." This catches people every time.
# This is all it takes to fix it during development: from flask_cors import CORS CORS(app)
If a Book has a relationship to an Author, and you load 50 books, SQLAlchemy will execute 1 query for books + 50 individual queries for authors unless you eager-load:
# BAD: N+1 queries
books = Book.query.all()
for book in books:
print(book.author.name) # Each access triggers a query
# GOOD: Eager loading
from sqlalchemy.orm import joinedload
books = Book.query.options(joinedload(Book.author)).all()
# Or set it at the model level:
class Book(db.Model):
author_id = db.Column(db.Integer, db.ForeignKey('authors.id'))
author = db.relationship('Author', lazy='joined')
Trusting that request.get_json() will always return valid data leads to KeyError and TypeError exceptions that become 500 errors in production. Always validate, either manually or with marshmallow.
# BAD: Leaks internal details
@app.errorhandler(Exception)
def handle_error(error):
return jsonify({'error': str(error)}), 500
# Client sees: "UNIQUE constraint failed: users.email"
# GOOD: Log internally, return generic message
@app.errorhandler(Exception)
def handle_error(error):
app.logger.error(f'Internal error: {error}', exc_info=True)
return jsonify({'error': 'An unexpected error occurred'}), 500
Version your API from day one. When you need to make breaking changes, clients on the old version continue to work:
# URL-based versioning (most common)
app.register_blueprint(books_v1_bp, url_prefix='/api/v1')
app.register_blueprint(books_v2_bp, url_prefix='/api/v2')
# Header-based versioning
@app.before_request
def check_api_version():
version = request.headers.get('API-Version', '1')
request.api_version = int(version)
Every response from your API should follow the same shape. Clients should never have to guess whether the response has a data key or a results key or a books key:
// Success
{
"success": true,
"data": { ... },
"pagination": { ... }
}
// Error
{
"success": false,
"error": {
"message": "Validation failed",
"code": 422,
"details": { ... }
}
}
Validate everything. Check types, lengths, ranges, formats, and business rules. Return all validation errors at once — do not make the client fix them one at a time:
// BAD: One error at a time
{"error": "Title is required"}
// (client fixes, resubmits)
{"error": "Year must be a number"}
// GOOD: All errors at once
{
"error": "Validation failed",
"details": {
"title": ["Title is required"],
"year": ["Year must be a number"],
"isbn": ["ISBN must be exactly 13 characters"]
}
}
Protect your API from abuse with Flask-Limiter:
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
limiter = Limiter(
app=app,
key_func=get_remote_address,
default_limits=["200 per day", "50 per hour"]
)
@app.route('/api/books', methods=['GET'])
@limiter.limit("30 per minute")
def get_books():
pass
@app.route('/api/auth/login', methods=['POST'])
@limiter.limit("5 per minute") # Stricter for auth endpoints
def login():
pass
Log every request in production. Log the method, path, status code, response time, and user identity if authenticated:
import time
from flask import g, request
@app.before_request
def start_timer():
g.start_time = time.time()
@app.after_request
def log_request(response):
duration = time.time() - g.start_time
app.logger.info(
f'{request.method} {request.path} '
f'{response.status_code} {duration:.3f}s'
)
return response
models.py, schemas in schemas.py, routes in routes/, error handlers in errors.py. This is not over-engineering — it is how you stay sane past 10 endpoints./api/v1/ is the simplest approach and the one your clients will thank you for.joinedload or subqueryload for relationships. Monitor your SQL queries in development.Flask gives you the freedom to architect your API exactly the way you want. That freedom is its greatest strength and its greatest risk — without discipline, a Flask API can become a tangled mess. Follow these patterns consistently and your API will be maintainable, testable, and production-ready.