Build a REST API with Flask: The No-Magic Approach
Build a production-ready REST API with Flask. Covers routes, blueprints, SQLAlchemy, authentication, error handling, testing, and deployment.
Django gives you everything. FastAPI gives you speed and type hints. Flask gives you freedom. It's a microframework, which means it comes with almost nothing built in, and that's exactly the point. You pick your ORM, your authentication library, your project structure. Nothing is decided for you.
That freedom is also why Flask tutorials are all over the place. Some are toy examples with three routes in a single file. Others jump straight to factory patterns and twelve-factor app principles. This tutorial sits in the middle: we'll build a real API with a sensible structure, a database, authentication, and tests.
Setting Up
Create a project directory and a virtual environment:
mkdir bookshelf-api && cd bookshelf-api
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
Install the dependencies we'll need:
pip install flask flask-sqlalchemy flask-migrate flask-cors python-dotenv PyJWT bcrypt
Quick rundown: Flask is the framework, SQLAlchemy is the ORM, Flask-Migrate handles database migrations with Alembic under the hood, Flask-CORS handles cross-origin requests, python-dotenv loads environment variables, PyJWT handles JSON Web Tokens, and bcrypt hashes passwords.
Project Structure
Here's what we're building toward:
bookshelf-api/
app/
__init__.py # Application factory
models.py # Database models
auth/
__init__.py
routes.py # Auth routes (register, login)
books/
__init__.py
routes.py # Book CRUD routes
errors.py # Error handlers
tests/
conftest.py
test_auth.py
test_books.py
config.py
run.py
.env
This structure scales. Each feature gets its own Blueprint (Flask's way of organizing routes into modules).
The Application Factory
Instead of creating the Flask app as a global variable, we use a factory function. This makes testing easier and prevents import headaches.
# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_cors import CORS
from config import Config
db = SQLAlchemy()
migrate = Migrate()
def create_app(config_class=Config):
app = Flask(__name__)
app.config.from_object(config_class)
db.init_app(app)
migrate.init_app(app, db)
CORS(app)
from app.auth.routes import auth_bp
from app.books.routes import books_bp
from app.errors import errors_bp
app.register_blueprint(auth_bp, url_prefix='/api/auth')
app.register_blueprint(books_bp, url_prefix='/api/books')
app.register_blueprint(errors_bp)
return app
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
class Config:
SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-secret-change-me')
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL', 'sqlite:///bookshelf.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
class TestConfig(Config):
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
# run.py
from app import create_app
app = create_app()
if __name__ == '__main__':
app.run(debug=True)
# .env
SECRET_KEY=your-super-secret-key
DATABASE_URL=sqlite:///bookshelf.db
Defining Models
# app/models.py
from datetime import datetime, timezone
from app import db
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
password_hash = db.Column(db.String(128), nullable=False)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
books = db.relationship('Book', backref='owner', lazy=True)
def to_dict(self):
return {
'id': self.id,
'username': self.username,
'email': self.email,
'created_at': self.created_at.isoformat(),
}
class Book(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
author = db.Column(db.String(200), nullable=False)
isbn = db.Column(db.String(13), unique=True)
rating = db.Column(db.Integer)
notes = db.Column(db.Text)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
def to_dict(self):
return {
'id': self.id,
'title': self.title,
'author': self.author,
'isbn': self.isbn,
'rating': self.rating,
'notes': self.notes,
'created_at': self.created_at.isoformat(),
}
Initialize the database:
flask --app run db init
flask --app run db migrate -m "Initial migration"
flask --app run db upgrade
Routes and Request Handling
Flask routes are straightforward. Decorate a function, return a response. Let's build the books Blueprint.
# app/books/__init__.py
# (empty, makes it a package)
# app/books/routes.py
from flask import Blueprint, request, jsonify
from app import db
from app.models import Book
from app.auth.routes import token_required
books_bp = Blueprint('books', __name__)
@books_bp.route('/', methods=['GET'])
@token_required
def get_books(current_user):
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 20, type=int)
pagination = Book.query.filter_by(user_id=current_user.id)\
.order_by(Book.created_at.desc())\
.paginate(page=page, per_page=per_page, error_out=False)
return jsonify({
'books': [book.to_dict() for book in pagination.items],
'total': pagination.total,
'page': pagination.page,
'pages': pagination.pages,
})
@books_bp.route('/', methods=['POST'])
@token_required
def create_book(current_user):
data = request.get_json()
if not data or not data.get('title') or not data.get('author'):
return jsonify({'error': 'Title and author are required'}), 400
book = Book(
title=data['title'],
author=data['author'],
isbn=data.get('isbn'),
rating=data.get('rating'),
notes=data.get('notes'),
user_id=current_user.id,
)
db.session.add(book)
db.session.commit()
return jsonify(book.to_dict()), 201
@books_bp.route('/<int:book_id>', methods=['GET'])
@token_required
def get_book(current_user, book_id):
book = Book.query.filter_by(id=book_id, user_id=current_user.id).first()
if not book:
return jsonify({'error': 'Book not found'}), 404
return jsonify(book.to_dict())
@books_bp.route('/<int:book_id>', methods=['PUT'])
@token_required
def update_book(current_user, book_id):
book = Book.query.filter_by(id=book_id, user_id=current_user.id).first()
if not book:
return jsonify({'error': 'Book not found'}), 404
data = request.get_json()
if data.get('title'):
book.title = data['title']
if data.get('author'):
book.author = data['author']
if 'isbn' in data:
book.isbn = data['isbn']
if 'rating' in data:
book.rating = data['rating']
if 'notes' in data:
book.notes = data['notes']
db.session.commit()
return jsonify(book.to_dict())
@books_bp.route('/<int:book_id>', methods=['DELETE'])
@token_required
def delete_book(current_user, book_id):
book = Book.query.filter_by(id=book_id, user_id=current_user.id).first()
if not book:
return jsonify({'error': 'Book not found'}), 404
db.session.delete(book)
db.session.commit()
return jsonify({'message': 'Book deleted'}), 200
Notice a few things. Every route checks user_id so users can only see their own books. Pagination is built in. The @token_required decorator (which we'll build next) handles authentication.
Authentication
# app/auth/__init__.py
# (empty)
# app/auth/routes.py
import jwt
import bcrypt
from datetime import datetime, timedelta, timezone
from functools import wraps
from flask import Blueprint, request, jsonify, current_app
from app import db
from app.models import User
auth_bp = Blueprint('auth', __name__)
def token_required(f):
@wraps(f)
def decorated(args, *kwargs):
token = None
auth_header = request.headers.get('Authorization')
if auth_header and auth_header.startswith('Bearer '):
token = auth_header.split(' ')[1]
if not token:
return jsonify({'error': 'Token is missing'}), 401
try:
payload = jwt.decode(token, current_app.config['SECRET_KEY'], algorithms=['HS256'])
current_user = User.query.get(payload['user_id'])
if not current_user:
return jsonify({'error': 'User not found'}), 401
except jwt.ExpiredSignatureError:
return jsonify({'error': 'Token has expired'}), 401
except jwt.InvalidTokenError:
return jsonify({'error': 'Invalid token'}), 401
return f(current_user, args, *kwargs)
return decorated
def generate_token(user):
payload = {
'user_id': user.id,
'exp': datetime.now(timezone.utc) + timedelta(hours=24),
'iat': datetime.now(timezone.utc),
}
return jwt.encode(payload, current_app.config['SECRET_KEY'], algorithm='HS256')
@auth_bp.route('/register', methods=['POST'])
def register():
data = request.get_json()
if not data or not data.get('username') or not data.get('email') or not data.get('password'):
return jsonify({'error': 'Username, email, and password are required'}), 400
if User.query.filter_by(email=data['email']).first():
return jsonify({'error': 'Email already registered'}), 409
if User.query.filter_by(username=data['username']).first():
return jsonify({'error': 'Username already taken'}), 409
password_hash = bcrypt.hashpw(
data['password'].encode('utf-8'),
bcrypt.gensalt()
).decode('utf-8')
user = User(
username=data['username'],
email=data['email'],
password_hash=password_hash,
)
db.session.add(user)
db.session.commit()
token = generate_token(user)
return jsonify({
'message': 'User created',
'token': token,
'user': user.to_dict(),
}), 201
@auth_bp.route('/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 = User.query.filter_by(email=data['email']).first()
if not user or not bcrypt.checkpw(
data['password'].encode('utf-8'),
user.password_hash.encode('utf-8')
):
return jsonify({'error': 'Invalid email or password'}), 401
token = generate_token(user)
return jsonify({
'token': token,
'user': user.to_dict(),
})
Error Handling
Flask lets you register global error handlers:
# app/errors.py
from flask import Blueprint, jsonify
from sqlalchemy.exc import IntegrityError
from app import db
errors_bp = Blueprint('errors', __name__)
@errors_bp.app_errorhandler(404)
def not_found(error):
return jsonify({'error': 'Resource not found'}), 404
@errors_bp.app_errorhandler(405)
def method_not_allowed(error):
return jsonify({'error': 'Method not allowed'}), 405
@errors_bp.app_errorhandler(500)
def internal_error(error):
db.session.rollback()
return jsonify({'error': 'Internal server error'}), 500
@errors_bp.app_errorhandler(IntegrityError)
def handle_integrity_error(error):
db.session.rollback()
return jsonify({'error': 'Database integrity error. Duplicate or invalid data.'}), 409
The app_errorhandler (note the app_ prefix) registers the handler for the entire application, not just the blueprint.
Input Validation
For a production API, you'd want something like Marshmallow or Pydantic for validation. Here's a lightweight approach:
def validate_book_data(data, required=True):
errors = []
if required:
if not data.get('title'):
errors.append('Title is required')
if not data.get('author'):
errors.append('Author is required')
if data.get('rating') is not None:
if not isinstance(data['rating'], int) or not 1 <= data['rating'] <= 5:
errors.append('Rating must be an integer between 1 and 5')
if data.get('isbn'):
isbn = data['isbn'].replace('-', '')
if len(isbn) not in (10, 13) or not isbn.isdigit():
errors.append('ISBN must be 10 or 13 digits')
return errors
Then use it in your routes:
@books_bp.route('/', methods=['POST'])
@token_required
def create_book(current_user):
data = request.get_json()
errors = validate_book_data(data)
if errors:
return jsonify({'errors': errors}), 400
# ... create the book
Testing
Flask's test client makes testing straightforward:
# tests/conftest.py
import pytest
from app import create_app, db
from config import TestConfig
@pytest.fixture
def app():
app = create_app(TestConfig)
with app.app_context():
db.create_all()
yield app
db.session.remove()
db.drop_all()
@pytest.fixture
def client(app):
return app.test_client()
@pytest.fixture
def auth_header(client):
# Register a user and return the auth header
response = client.post('/api/auth/register', json={
'username': 'testuser',
'email': 'test@example.com',
'password': 'securepassword123',
})
token = response.get_json()['token']
return {'Authorization': f'Bearer {token}'}
# tests/test_books.py
def test_create_book(client, auth_header):
response = client.post('/api/books/', json={
'title': 'The Pragmatic Programmer',
'author': 'David Thomas, Andrew Hunt',
'rating': 5,
}, headers=auth_header)
assert response.status_code == 201
data = response.get_json()
assert data['title'] == 'The Pragmatic Programmer'
assert data['rating'] == 5
def test_create_book_without_auth(client):
response = client.post('/api/books/', json={
'title': 'Some Book',
'author': 'Some Author',
})
assert response.status_code == 401
def test_get_books_pagination(client, auth_header):
# Create 25 books
for i in range(25):
client.post('/api/books/', json={
'title': f'Book {i}',
'author': f'Author {i}',
}, headers=auth_header)
response = client.get('/api/books/?page=1&per_page=10', headers=auth_header)
data = response.get_json()
assert len(data['books']) == 10
assert data['total'] == 25
assert data['pages'] == 3
def test_user_isolation(client):
# Register two users
r1 = client.post('/api/auth/register', json={
'username': 'user1', 'email': 'u1@test.com', 'password': 'pass123',
})
r2 = client.post('/api/auth/register', json={
'username': 'user2', 'email': 'u2@test.com', 'password': 'pass123',
})
h1 = {'Authorization': f'Bearer {r1.get_json()["token"]}'}
h2 = {'Authorization': f'Bearer {r2.get_json()["token"]}'}
# User 1 creates a book
client.post('/api/books/', json={
'title': 'Private Book', 'author': 'Author',
}, headers=h1)
# User 2 shouldn't see it
response = client.get('/api/books/', headers=h2)
assert len(response.get_json()['books']) == 0
Run tests with:
pytest -v
Deploying to Production
For production, you don't use flask run. You use a WSGI server like Gunicorn:
pip install gunicorn
gunicorn -w 4 -b 0.0.0.0:8000 "app:create_app()"
The -w 4 flag runs four worker processes. A common formula is (2 * CPU cores) + 1.
For a Dockerfile:
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8000", "app:create_app()"]
Common Mistakes
Circular imports. The most common Flask headache. That's why we use the application factory pattern and initializedb separately from create_app. If you're importing app inside a model file that app also imports, you have a circular dependency.
Not using app.app_context(). Outside of a request (like in a script or test), you need with app.app_context(): before doing anything that touches the database or config.
Returning Python dicts directly. Flask 2.2+ automatically converts dicts to JSON responses, but the status code defaults to 200. Use jsonify() and explicit status codes for clarity: return jsonify(data), 201.
Not rolling back on errors. If a database operation fails mid-transaction, you need db.session.rollback(). The error handler we set up does this automatically for unhandled exceptions.
Trusting user input. Always validate. Use parameterized queries (SQLAlchemy does this by default). Never concatenate user input into SQL strings.
What's Next
You have a working API with authentication, CRUD operations, pagination, error handling, and tests. The natural next steps are adding rate limiting with Flask-Limiter, implementing refresh tokens, adding file upload support, setting up logging with structured output, and deploying behind Nginx with SSL.
For hands-on practice building Flask APIs and other backend projects, check out CodeUp.