Flask REST API: The Complete Guide for 2026

Flask remains one of the most popular Python frameworks for building REST APIs. Its minimal core, extensive ecosystem, and straightforward approach make it the go-to choice for developers who want control over their stack without the overhead of a full framework. This guide walks you through building a production-ready Flask REST API from scratch, covering everything from initial setup to deployment.

1. Project Setup and Structure

Start by creating a virtual environment and installing Flask. Use Python 3.12 or later for the best experience in 2026.

# Create project and virtual environment
mkdir flask-api && cd flask-api
python -m venv venv
source venv/bin/activate  # Linux/macOS

# Install Flask and common extensions
pip install flask flask-sqlalchemy flask-migrate flask-jwt-extended
pip install flask-cors marshmallow python-dotenv gunicorn

Recommended Project Structure

Organize your project using the application factory pattern. This makes testing easier and allows multiple configurations.

flask-api/
    app/
        __init__.py          # Application factory
        config.py            # Configuration classes
        models/
            __init__.py
            user.py
            item.py
        routes/
            __init__.py
            auth.py
            items.py
        schemas/
            __init__.py
            item.py
        utils/
            errors.py
    tests/
        conftest.py
        test_items.py
    .env
    requirements.txt
    wsgi.py

Application Factory

# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager
from flask_cors import CORS

db = SQLAlchemy()
migrate = Migrate()
jwt = JWTManager()

def create_app(config_name='development'):
    app = Flask(__name__)

    # Load configuration
    if config_name == 'testing':
        app.config.from_object('app.config.TestingConfig')
    elif config_name == 'production':
        app.config.from_object('app.config.ProductionConfig')
    else:
        app.config.from_object('app.config.DevelopmentConfig')

    # Initialize extensions
    db.init_app(app)
    migrate.init_app(app, db)
    jwt.init_app(app)
    CORS(app)

    # Register blueprints
    from app.routes.auth import auth_bp
    from app.routes.items import items_bp
    app.register_blueprint(auth_bp, url_prefix='/api/auth')
    app.register_blueprint(items_bp, url_prefix='/api/items')

    # Register error handlers
    from app.utils.errors import register_error_handlers
    register_error_handlers(app)

    return app

Configuration Classes

# app/config.py
import os
from datetime import timedelta

class BaseConfig:
    SECRET_KEY = os.environ.get('SECRET_KEY', 'change-me-in-production')
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
    JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)

class DevelopmentConfig(BaseConfig):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL', 'sqlite:///dev.db')

class TestingConfig(BaseConfig):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'

class ProductionConfig(BaseConfig):
    SQLALCHEMY_DATABASE_URI = os.environ['DATABASE_URL']
    JWT_ACCESS_TOKEN_EXPIRES = timedelta(minutes=15)

2. Creating Routes and Endpoints

Flask uses blueprints to organize routes into logical groups. Each resource gets its own blueprint with endpoints for standard CRUD operations.

# app/routes/items.py
from flask import Blueprint, request, jsonify
from app.models.item import Item
from app import db

items_bp = Blueprint('items', __name__)

@items_bp.route('/', methods=['GET'])
def get_items():
    items = Item.query.all()
    return jsonify([item.to_dict() for item in items]), 200

@items_bp.route('/<int:item_id>', methods=['GET'])
def get_item(item_id):
    item = db.session.get(Item, item_id)
    if not item:
        return jsonify({'error': 'Item not found'}), 404
    return jsonify(item.to_dict()), 200

@items_bp.route('/', methods=['POST'])
def create_item():
    data = request.get_json()
    if not data or 'name' not in data:
        return jsonify({'error': 'Name is required'}), 400
    item = Item(name=data['name'], description=data.get('description', ''))
    db.session.add(item)
    db.session.commit()
    return jsonify(item.to_dict()), 201

@items_bp.route('/<int:item_id>', methods=['PUT'])
def update_item(item_id):
    item = db.session.get(Item, item_id)
    if not item:
        return jsonify({'error': 'Item not found'}), 404
    data = request.get_json()
    item.name = data.get('name', item.name)
    item.description = data.get('description', item.description)
    db.session.commit()
    return jsonify(item.to_dict()), 200

@items_bp.route('/<int:item_id>', methods=['DELETE'])
def delete_item(item_id):
    item = db.session.get(Item, item_id)
    if not item:
        return jsonify({'error': 'Item not found'}), 404
    db.session.delete(item)
    db.session.commit()
    return jsonify({'message': 'Item deleted'}), 200

3. Request Handling

JSON Request Bodies

Flask parses JSON bodies automatically with request.get_json(). Always handle the case where the body is missing or malformed.

@items_bp.route('/', methods=['POST'])
def create_item():
    # Returns None if Content-Type is not application/json or body is invalid
    data = request.get_json(silent=True)
    if data is None:
        return jsonify({'error': 'Request body must be valid JSON'}), 400

    name = data.get('name')
    price = data.get('price', 0.0)
    tags = data.get('tags', [])
    return jsonify({'received': data}), 201

Query Parameters and Pagination

@items_bp.route('/', methods=['GET'])
def get_items():
    # /api/items?page=2&per_page=20&sort=name&category=tools
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 10, type=int)
    sort_by = request.args.get('sort', 'created_at')
    category = request.args.get('category')

    query = Item.query
    if category:
        query = query.filter_by(category=category)

    pagination = query.order_by(sort_by).paginate(
        page=page, per_page=min(per_page, 100), error_out=False
    )
    return jsonify({
        'items': [item.to_dict() for item in pagination.items],
        'total': pagination.total,
        'page': pagination.page,
        'has_next': pagination.has_next
    }), 200

URL Parameter Types

@app.route('/items/<int:item_id>')      # Integer
@app.route('/users/<username>')          # String (default)
@app.route('/orders/<uuid:order_id>')    # UUID
@app.route('/files/<path:filepath>')     # Path (includes slashes)

4. Response Formatting

Always return consistent JSON structures. Create a helper function that every endpoint uses.

def api_response(data=None, message=None, status=200):
    body = {}
    if data is not None:
        body['data'] = data
    if message:
        body['message'] = message
    return jsonify(body), status

# Response with custom headers
@items_bp.route('/', methods=['POST'])
def create_item():
    item = Item(name='New Item')
    db.session.add(item)
    db.session.commit()
    response = jsonify(item.to_dict())
    response.status_code = 201
    response.headers['Location'] = f'/api/items/{item.id}'
    return response

Common HTTP Status Codes for APIs

5. Error Handling

Register global error handlers so every error returns a consistent JSON response instead of HTML.

# app/utils/errors.py
from flask import jsonify
from werkzeug.exceptions import HTTPException

class APIError(Exception):
    """Custom API exception with status code and payload."""
    def __init__(self, message, status_code=400, payload=None):
        super().__init__()
        self.message = message
        self.status_code = status_code
        self.payload = payload

    def to_dict(self):
        rv = {'error': self.message}
        if self.payload:
            rv['details'] = self.payload
        return rv

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({'error': error.description, 'status': error.code}), error.code

    @app.errorhandler(500)
    def internal_error(error):
        return jsonify({'error': 'Internal server error'}), 500

Use the custom exception in your route handlers:

from app.utils.errors import APIError

@items_bp.route('/', methods=['POST'])
def create_item():
    data = request.get_json(silent=True)
    if data is None:
        raise APIError('Request body must be valid JSON', 400)
    if 'name' not in data:
        raise APIError('Validation failed', 422, payload={
            'fields': {'name': 'This field is required'}
        })
    if Item.query.filter_by(name=data['name']).first():
        raise APIError('An item with this name already exists', 409)

6. SQLAlchemy Database Integration

Flask-SQLAlchemy provides a clean interface for defining models and querying a database. Define models with relationships, serialization, and constraints.

# app/models/item.py
from datetime import datetime, timezone
from app import db

class Item(db.Model):
    __tablename__ = 'items'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False, unique=True)
    description = db.Column(db.Text, default='')
    price = db.Column(db.Float, default=0.0)
    category = db.Column(db.String(50), index=True)
    is_active = db.Column(db.Boolean, default=True)
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
    updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc),
                           onupdate=lambda: datetime.now(timezone.utc))
    owner_id = db.Column(db.Integer, db.ForeignKey('users.id'))
    owner = db.relationship('User', back_populates='items')

    def to_dict(self):
        return {
            'id': self.id, 'name': self.name, 'description': self.description,
            'price': self.price, 'category': self.category,
            'is_active': self.is_active,
            'created_at': self.created_at.isoformat(),
            'updated_at': self.updated_at.isoformat(),
            'owner_id': self.owner_id
        }

# app/models/user.py
from werkzeug.security import generate_password_hash, check_password_hash
from app import db

class User(db.Model):
    __tablename__ = 'users'

    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(256), nullable=False)
    is_admin = db.Column(db.Boolean, default=False)
    items = db.relationship('Item', back_populates='owner', lazy='dynamic')

    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 to_dict(self):
        return {'id': self.id, 'username': self.username,
                'email': self.email, 'is_admin': self.is_admin}

Database Migrations with Flask-Migrate

flask db init                          # Initialize (first time only)
flask db migrate -m "Add items table"  # Create migration
flask db upgrade                       # Apply migrations
flask db downgrade                     # Roll back one migration

7. Request Validation

Option A: Marshmallow (Recommended)

Marshmallow provides serialization, deserialization, and validation in one library. It integrates naturally with Flask and SQLAlchemy.

# app/schemas/item.py
from marshmallow import Schema, fields, validate, validates, ValidationError

class ItemCreateSchema(Schema):
    name = fields.String(required=True, validate=validate.Length(min=1, max=100))
    description = fields.String(load_default='', validate=validate.Length(max=500))
    price = fields.Float(load_default=0.0, validate=validate.Range(min=0))
    category = fields.String(validate=validate.OneOf(['tools', 'books', 'electronics']))

    @validates('name')
    def validate_name(self, value):
        if value.strip() != value:
            raise ValidationError('Name cannot have leading/trailing spaces')

class ItemUpdateSchema(Schema):
    name = fields.String(validate=validate.Length(min=1, max=100))
    description = fields.String(validate=validate.Length(max=500))
    price = fields.Float(validate=validate.Range(min=0))
    category = fields.String(validate=validate.OneOf(['tools', 'books', 'electronics']))

class ItemQuerySchema(Schema):
    page = fields.Integer(load_default=1, validate=validate.Range(min=1))
    per_page = fields.Integer(load_default=10, validate=validate.Range(min=1, max=100))
    category = fields.String()
    sort = fields.String(load_default='created_at',
                         validate=validate.OneOf(['name', 'price', 'created_at']))

Use schemas in your routes:

from marshmallow import ValidationError as MarshmallowError
from app.schemas.item import ItemCreateSchema, ItemQuerySchema

item_schema = ItemCreateSchema()

@items_bp.route('/', methods=['POST'])
def create_item():
    try:
        data = item_schema.load(request.get_json())
    except MarshmallowError as err:
        return jsonify({'error': 'Validation failed', 'details': err.messages}), 422
    item = Item(**data)
    db.session.add(item)
    db.session.commit()
    return jsonify(item.to_dict()), 201

Option B: Pydantic

from pydantic import BaseModel, Field, field_validator, ValidationError

class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    description: str = Field(default='', max_length=500)
    price: float = Field(default=0.0, ge=0)
    category: str | None = None

    @field_validator('category')
    @classmethod
    def validate_category(cls, v):
        if v and v not in ('tools', 'books', 'electronics'):
            raise ValueError('Invalid category')
        return v

@items_bp.route('/', methods=['POST'])
def create_item():
    try:
        data = ItemCreate.model_validate(request.get_json())
    except ValidationError as err:
        return jsonify({'error': 'Validation failed', 'details': err.errors()}), 422
    item = Item(**data.model_dump(exclude_none=True))
    db.session.add(item)
    db.session.commit()
    return jsonify(item.to_dict()), 201

8. Authentication

JWT Authentication with flask-jwt-extended

# app/routes/auth.py
from flask import Blueprint, request, jsonify
from flask_jwt_extended import (
    create_access_token, create_refresh_token,
    jwt_required, get_jwt_identity, get_jwt
)
from app.models.user import User
from app import db

auth_bp = Blueprint('auth', __name__)

@auth_bp.route('/register', methods=['POST'])
def register():
    data = request.get_json()
    if User.query.filter_by(username=data['username']).first():
        return jsonify({'error': 'Username already taken'}), 409
    user = User(username=data['username'], email=data['email'])
    user.set_password(data['password'])
    db.session.add(user)
    db.session.commit()
    return jsonify({'message': 'User created', 'user': user.to_dict()}), 201

@auth_bp.route('/login', methods=['POST'])
def login():
    data = request.get_json()
    user = User.query.filter_by(username=data.get('username')).first()
    if not user or not user.check_password(data.get('password', '')):
        return jsonify({'error': 'Invalid credentials'}), 401
    access_token = create_access_token(
        identity=str(user.id), additional_claims={'is_admin': user.is_admin}
    )
    refresh_token = create_refresh_token(identity=str(user.id))
    return jsonify({'access_token': access_token,
                    'refresh_token': refresh_token, 'user': user.to_dict()}), 200

@auth_bp.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
    new_token = create_access_token(identity=get_jwt_identity())
    return jsonify({'access_token': new_token}), 200

Protecting Routes and Admin Access

from functools import wraps

# Basic protection - require valid JWT
@items_bp.route('/', methods=['POST'])
@jwt_required()
def create_item():
    current_user_id = get_jwt_identity()
    data = request.get_json()
    item = Item(name=data['name'], owner_id=int(current_user_id))
    db.session.add(item)
    db.session.commit()
    return jsonify(item.to_dict()), 201

# Custom decorator for admin-only routes
def admin_required(fn):
    @wraps(fn)
    @jwt_required()
    def wrapper(*args, **kwargs):
        claims = get_jwt()
        if not claims.get('is_admin', False):
            return jsonify({'error': 'Admin access required'}), 403
        return fn(*args, **kwargs)
    return wrapper

@items_bp.route('/<int:item_id>', methods=['DELETE'])
@admin_required
def delete_item(item_id):
    item = db.session.get(Item, item_id)
    if not item:
        return jsonify({'error': 'Not found'}), 404
    db.session.delete(item)
    db.session.commit()
    return jsonify({'message': 'Deleted'}), 200

API Key Authentication

For server-to-server communication, API keys are simpler than JWT.

def require_api_key(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        api_key = request.headers.get('X-API-Key')
        if not api_key:
            return jsonify({'error': 'API key required'}), 401
        key_record = APIKey.query.filter_by(key=api_key, is_active=True).first()
        if not key_record:
            return jsonify({'error': 'Invalid API key'}), 401
        from flask import g
        g.api_client = key_record.client_name
        return fn(*args, **kwargs)
    return wrapper

9. CORS Configuration

Cross-Origin Resource Sharing (CORS) is required when your frontend runs on a different domain than your API.

from flask_cors import CORS

# Development: allow all origins
CORS(app)

# Production: restrict origins
CORS(app, resources={
    r"/api/*": {
        "origins": ["https://yourdomain.com", "https://app.yourdomain.com"],
        "methods": ["GET", "POST", "PUT", "DELETE", "OPTIONS"],
        "allow_headers": ["Content-Type", "Authorization"],
        "max_age": 3600
    }
})

10. Testing with pytest

Flask's test client lets you simulate HTTP requests without starting a server. Combine it with pytest fixtures for clean, repeatable tests.

# tests/conftest.py
import pytest
from app import create_app, db

@pytest.fixture
def app():
    app = create_app('testing')
    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_headers(client):
    """Create a test user and return auth headers."""
    client.post('/api/auth/register', json={
        'username': 'testuser', 'email': 'test@example.com',
        'password': 'securepass123'
    })
    response = client.post('/api/auth/login', json={
        'username': 'testuser', 'password': 'securepass123'
    })
    token = response.get_json()['access_token']
    return {'Authorization': f'Bearer {token}'}

# tests/test_items.py
def test_create_item(client, auth_headers):
    resp = client.post('/api/items/', json={
        'name': 'Test Item', 'price': 9.99
    }, headers=auth_headers)
    assert resp.status_code == 201
    assert resp.get_json()['name'] == 'Test Item'

def test_create_item_validation(client, auth_headers):
    resp = client.post('/api/items/', json={'description': 'No name'}, headers=auth_headers)
    assert resp.status_code == 422

def test_create_item_unauthorized(client):
    resp = client.post('/api/items/', json={'name': 'Test'})
    assert resp.status_code == 401

def test_get_item_not_found(client):
    assert client.get('/api/items/999').status_code == 404

def test_update_and_delete(client, auth_headers):
    resp = client.post('/api/items/', json={'name': 'Original'}, headers=auth_headers)
    item_id = resp.get_json()['id']

    update = client.put(f'/api/items/{item_id}', json={'name': 'Updated'}, headers=auth_headers)
    assert update.get_json()['name'] == 'Updated'

    delete = client.delete(f'/api/items/{item_id}', headers=auth_headers)
    assert delete.status_code == 200
    assert client.get(f'/api/items/{item_id}').status_code == 404

def test_pagination(client, auth_headers):
    for i in range(15):
        client.post('/api/items/', json={'name': f'Item {i}'}, headers=auth_headers)
    resp = client.get('/api/items/?page=1&per_page=10')
    data = resp.get_json()
    assert len(data['items']) == 10
    assert data['total'] == 15

# Run tests
pytest tests/ -v
pytest tests/ --cov=app --cov-report=term-missing

11. Deployment

Gunicorn (WSGI Server)

Never use Flask's built-in server in production. Gunicorn handles concurrent requests and worker management.

# wsgi.py
from app import create_app
app = create_app('production')

# gunicorn.conf.py
import multiprocessing
bind = "0.0.0.0:8000"
workers = multiprocessing.cpu_count() * 2 + 1
timeout = 120
accesslog = "-"
errorlog = "-"

# Run: gunicorn -c gunicorn.conf.py wsgi:app

Docker

# Dockerfile
FROM python:3.12-slim
WORKDIR /app
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
RUN adduser --disabled-password --no-create-home appuser
USER appuser
EXPOSE 8000
CMD ["gunicorn", "-c", "gunicorn.conf.py", "wsgi:app"]

# docker-compose.yml
services:
  api:
    build: .
    ports: ["8000:8000"]
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/apidb
      - SECRET_KEY=${SECRET_KEY}
      - FLASK_ENV=production
    depends_on:
      db: { condition: service_healthy }
  db:
    image: postgres:16
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: apidb
    volumes: [pgdata:/var/lib/postgresql/data]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user"]
      interval: 5s
      timeout: 5s
      retries: 5
volumes:
  pgdata:

Environment Variables

# .env (never commit this file)
SECRET_KEY=your-super-secret-key-here
DATABASE_URL=postgresql://user:pass@localhost:5432/apidb

# .flaskenv (safe to commit)
FLASK_APP=wsgi.py
FLASK_DEBUG=0

12. Best Practices and Common Patterns

Rate Limiting

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/auth/login', methods=['POST'])
@limiter.limit("5 per minute")
def login():
    ...

Request Logging

import logging
from logging.handlers import RotatingFileHandler

def configure_logging(app):
    handler = RotatingFileHandler('api.log', maxBytes=10_000_000, backupCount=5)
    handler.setFormatter(logging.Formatter(
        '[%(asctime)s] %(levelname)s in %(module)s: %(message)s'
    ))
    app.logger.addHandler(handler)
    app.logger.setLevel(logging.INFO)

    @app.after_request
    def log_response(response):
        app.logger.info('%s %s %s %s', request.remote_addr,
                        request.method, request.full_path, response.status_code)
        return response

API Versioning

v1 = Blueprint('v1', __name__, url_prefix='/api/v1')
v2 = Blueprint('v2', __name__, url_prefix='/api/v2')

@v1.route('/items')
def get_items_v1():
    return jsonify([item.to_dict() for item in Item.query.all()])

@v2.route('/items')
def get_items_v2():
    page = request.args.get('page', 1, type=int)
    pagination = Item.query.paginate(page=page, per_page=20)
    return jsonify({
        'data': [item.to_dict() for item in pagination.items],
        'meta': {'total': pagination.total, 'page': page}
    })

Health Check Endpoint

@app.route('/health')
def health():
    try:
        db.session.execute(db.text('SELECT 1'))
        db_status = 'connected'
    except Exception:
        db_status = 'disconnected'
    return jsonify({
        'status': 'healthy' if db_status == 'connected' else 'degraded',
        'database': db_status, 'version': '1.0.0'
    }), 200 if db_status == 'connected' else 503

Quick Reference: Essential Patterns