Building a REST API with Python and Flask: Intermediate Guide

This tutorial walks through building a production-minded REST API using Python and Flask, with an emphasis on intermediate concerns: application structure, configuration, request validation, error handling, database persistence, migrations, authentication, pagination, testing, and deployment basics.

You will build a small API for managing “tasks” (a classic CRUD domain), but the patterns apply to most REST services.

Table of Contents

• Prerequisites
• Project Setup
• Recommended Project Structure
• Creating the Flask App (Application Factory)
• Configuration (Dev/Test/Prod)
• Database Integration with SQLAlchemy
• Migrations with Flask-Migrate
• Defining Models
• Designing RESTful Endpoints
• Request Validation and Serialization
• Error Handling (Consistent JSON Errors)
• Pagination, Filtering, and Sorting
• Authentication with JWT
• Testing with Pytest
• Running the API Locally
• Deployment Notes (Gunicorn)
• Next Steps

Prerequisites

You should already be comfortable with:

• Python basics (functions, classes, virtual environments)
• HTTP concepts (methods, status codes, headers)
• JSON
• Basic SQL concepts

You will need installed:

• Python 3.10+ (3.11+ recommended)
• pip
• SQLite (bundled with Python) for local development

Project Setup

Create a project directory and a virtual environment.

mkdir flask-rest-api-intermediate
cd flask-rest-api-intermediate

python -m venv .venv

Activate the environment:

• macOS/Linux:

source .venv/bin/activate

• Windows (PowerShell):

.\.venv\Scripts\Activate.ps1

Install dependencies:

pip install Flask==3.0.2 Flask-SQLAlchemy==3.1.1 Flask-Migrate==4.0.7 PyJWT==2.8.0 python-dotenv==1.0.1 pytest==8.0.2

Optional but strongly recommended for formatting/linting:

pip install black==24.2.0 ruff==0.2.2

Freeze requirements:

pip freeze > requirements.txt

Recommended Project Structure

A common mistake is starting with a single app.py file and letting it grow until it becomes unmaintainable. Instead, use an application factory and split concerns into modules.

Create this structure:

flask-rest-api-intermediate/
  app/
    __init__.py
    config.py
    extensions.py
    models.py
    auth.py
    routes/
      __init__.py
      tasks.py
  migrations/
  tests/
    test_tasks.py
  .env
  wsgi.py
  requirements.txt

Key ideas:

• app/__init__.py contains create_app() (the factory).
• extensions.py holds initialized extensions (SQLAlchemy, Migrate) to avoid circular imports.
• routes/ contains blueprints for endpoints.
• wsgi.py provides an entry point for production servers like Gunicorn.

Creating the Flask App (Application Factory)

Create app/extensions.py:

from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

db = SQLAlchemy()
migrate = Migrate()

Create app/__init__.py:

from flask import Flask

from .config import Config
from .extensions import db, migrate

def create_app(config_object: type[Config] = Config) -> Flask:
    app = Flask(__name__)
    app.config.from_object(config_object)

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

    # Register blueprints
    from .routes.tasks import tasks_bp
    app.register_blueprint(tasks_bp, url_prefix="/api")

    from .auth import auth_bp
    app.register_blueprint(auth_bp, url_prefix="/api")

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

    return app

Notice the pattern:

• The app is created inside a function.
• Extensions are initialized after app creation.
• Blueprints are registered inside the factory to avoid import cycles.

Create wsgi.py at the project root:

from app import create_app

app = create_app()

This file is what Gunicorn will point at (e.g., gunicorn wsgi:app).

Configuration (Dev/Test/Prod)

Create app/config.py:

import os

class Config:
    SECRET_KEY = os.getenv("SECRET_KEY", "dev-secret-change-me")
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    SQLALCHEMY_DATABASE_URI = os.getenv("DATABASE_URL", "sqlite:///app.db")
    JWT_ISSUER = os.getenv("JWT_ISSUER", "flask-rest-api")
    JWT_EXP_SECONDS = int(os.getenv("JWT_EXP_SECONDS", "3600"))

class TestConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"

Create a .env file in the project root:

SECRET_KEY=change-this-in-real-life
DATABASE_URL=sqlite:///app.db
JWT_ISSUER=flask-rest-api
JWT_EXP_SECONDS=3600

To load .env automatically, you can set environment variables in your shell, or you can rely on Flask CLI with python-dotenv (Flask will load .env for you in many setups). If needed, you can explicitly load it, but keep the app factory clean unless you have a specific reason.

Database Integration with SQLAlchemy

SQLAlchemy provides:

• A Pythonic way to define tables as classes (ORM models).
• A session for writing/reading data.
• Query building.

In Flask-SQLAlchemy, db is a wrapper that exposes:

• db.Model base class
• db.session for transactions
• Column types and helpers

Migrations with Flask-Migrate

Migrations help you evolve your database schema safely over time.

Initialize migrations:

export FLASK_APP=wsgi:app
flask db init

If you are on Windows PowerShell:

$env:FLASK_APP="wsgi:app"
flask db init

You will create models first, then run:

flask db migrate -m "create tasks table"
flask db upgrade

Defining Models

Create app/models.py:

from datetime import datetime
from .extensions import db

class Task(db.Model):
    __tablename__ = "tasks"

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    done = db.Column(db.Boolean, nullable=False, default=False)

    created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow)

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "title": self.title,
            "done": self.done,
            "created_at": self.created_at.isoformat() + "Z",
            "updated_at": self.updated_at.isoformat() + "Z",
        }

Notes:

• nullable=False enforces a database-level constraint.
• default=datetime.utcnow sets timestamps automatically.
• onupdate=datetime.utcnow updates updated_at when the row changes.

Now create migrations:

flask db migrate -m "create tasks table"
flask db upgrade

If everything is correct, a SQLite file app.db will appear (depending on your working directory and DATABASE_URL).

Designing RESTful Endpoints

A clean REST design uses:

• Nouns for resources: /tasks
• HTTP methods for actions:
  • GET /tasks list
  • POST /tasks create
  • GET /tasks/<id> retrieve
  • PUT /tasks/<id> replace
  • PATCH /tasks/<id> partial update
  • DELETE /tasks/<id> delete

You also want consistent status codes:

• 200 OK for successful reads/updates
• 201 Created for creation
• 204 No Content for deletion (optional)
• 400 Bad Request for invalid JSON/validation failures
• 401 Unauthorized for missing/invalid auth
• 404 Not Found for missing resources

Request Validation and Serialization

Flask does not enforce schemas by default. At intermediate level, you should at least:

• Validate required fields
• Validate types
• Reject unknown fields (optional but helpful)
• Return consistent error responses

You can do this manually or with a library (Marshmallow, Pydantic). Here we’ll implement a small manual validator to keep dependencies minimal.

Create app/routes/tasks.py:

from flask import Blueprint, request, jsonify
from sqlalchemy import select

from ..extensions import db
from ..models import Task
from ..security import require_auth

tasks_bp = Blueprint("tasks", __name__)

ALLOWED_FIELDS = {"title", "done"}

def parse_json() -> dict:
    data = request.get_json(silent=True)
    if data is None:
        raise ValueError("Request body must be valid JSON")
    if not isinstance(data, dict):
        raise ValueError("JSON body must be an object")
    unknown = set(data.keys()) - ALLOWED_FIELDS
    if unknown:
        raise ValueError(f"Unknown fields: {', '.join(sorted(unknown))}")
    return data

@tasks_bp.get("/tasks")
@require_auth
def list_tasks():
    # Pagination parameters
    page = int(request.args.get("page", 1))
    page_size = int(request.args.get("page_size", 20))
    page_size = max(1, min(page_size, 100))

    stmt = select(Task).order_by(Task.id.desc())
    pagination = db.paginate(stmt, page=page, per_page=page_size, error_out=False)

    return jsonify({
        "items": [t.to_dict() for t in pagination.items],
        "page": page,
        "page_size": page_size,
        "total": pagination.total,
    }), 200

@tasks_bp.post("/tasks")
@require_auth
def create_task():
    data = parse_json()
    title = data.get("title")
    done = data.get("done", False)

    if not title or not isinstance(title, str):
        return jsonify({"error": {"code": "validation_error", "message": "title is required and must be a string"}}), 400
    if not isinstance(done, bool):
        return jsonify({"error": {"code": "validation_error", "message": "done must be a boolean"}}), 400

    task = Task(title=title.strip(), done=done)
    db.session.add(task)
    db.session.commit()

    return jsonify(task.to_dict()), 201

@tasks_bp.get("/tasks/<int:task_id>")
@require_auth
def get_task(task_id: int):
    task = db.session.get(Task, task_id)
    if not task:
        return jsonify({"error": {"code": "not_found", "message": "Task not found"}}), 404
    return jsonify(task.to_dict()), 200

@tasks_bp.patch("/tasks/<int:task_id>")
@require_auth
def update_task(task_id: int):
    task = db.session.get(Task, task_id)
    if not task:
        return jsonify({"error": {"code": "not_found", "message": "Task not found"}}), 404

    data = parse_json()

    if "title" in data:
        if not isinstance(data["title"], str) or not data["title"].strip():
            return jsonify({"error": {"code": "validation_error", "message": "title must be a non-empty string"}}), 400
        task.title = data["title"].strip()

    if "done" in data:
        if not isinstance(data["done"], bool):
            return jsonify({"error": {"code": "validation_error", "message": "done must be a boolean"}}), 400
        task.done = data["done"]

    db.session.commit()
    return jsonify(task.to_dict()), 200

@tasks_bp.delete("/tasks/<int:task_id>")
@require_auth
def delete_task(task_id: int):
    task = db.session.get(Task, task_id)
    if not task:
        return jsonify({"error": {"code": "not_found", "message": "Task not found"}}), 404

    db.session.delete(task)
    db.session.commit()
    return "", 204

What’s happening here:

• parse_json() ensures the request body is a JSON object and rejects unknown fields.
• We validate title and done types.
• We use db.session.get() for primary key lookup (efficient and clear).
• We return JSON consistently.

Error Handling (Consistent JSON Errors)

Right now, some errors are returned inline, and parse_json() raises ValueError. If you don’t catch it, Flask will produce an HTML error page (not ideal for APIs).

Create app/errors.py:

from flask import Flask, jsonify
from werkzeug.exceptions import HTTPException

def register_error_handlers(app: Flask) -> None:
    @app.errorhandler(ValueError)
    def handle_value_error(err: ValueError):
        return jsonify({"error": {"code": "bad_request", "message": str(err)}}), 400

    @app.errorhandler(HTTPException)
    def handle_http_exception(err: HTTPException):
        # Converts default HTML errors into JSON
        return jsonify({
            "error": {
                "code": err.name.replace(" ", "_").lower(),
                "message": err.description,
            }
        }), err.code or 500

    @app.errorhandler(Exception)
    def handle_unexpected(err: Exception):
        # In production you would log err with stack trace.
        return jsonify({"error": {"code": "internal_error", "message": "An unexpected error occurred"}}), 500

This gives you:

• JSON errors for common HTTP exceptions
• A safe catch-all for unknown errors
• A specific handler for ValueError from validation

Pagination, Filtering, and Sorting

You already implemented pagination in GET /tasks. Let’s extend it with filtering and sorting.

Update list_tasks() in app/routes/tasks.py:

@tasks_bp.get("/tasks")
@require_auth
def list_tasks():
    page = int(request.args.get("page", 1))
    page_size = int(request.args.get("page_size", 20))
    page_size = max(1, min(page_size, 100))

    done_param = request.args.get("done")
    sort = request.args.get("sort", "id_desc")

    stmt = select(Task)

    # Filtering
    if done_param is not None:
        if done_param.lower() in ("true", "1", "yes"):
            stmt = stmt.where(Task.done.is_(True))
        elif done_param.lower() in ("false", "0", "no"):
            stmt = stmt.where(Task.done.is_(False))
        else:
            return jsonify({"error": {"code": "validation_error", "message": "done must be true or false"}}), 400

    # Sorting
    if sort == "id_desc":
        stmt = stmt.order_by(Task.id.desc())
    elif sort == "id_asc":
        stmt = stmt.order_by(Task.id.asc())
    elif sort == "created_desc":
        stmt = stmt.order_by(Task.created_at.desc())
    else:
        return jsonify({"error": {"code": "validation_error", "message": "Invalid sort value"}}), 400

    pagination = db.paginate(stmt, page=page, per_page=page_size, error_out=False)

    return jsonify({
        "items": [t.to_dict() for t in pagination.items],
        "page": page,
        "page_size": page_size,
        "total": pagination.total,
    }), 200

This approach:

• Keeps filtering logic close to the endpoint.
• Validates query parameters.
• Avoids returning massive datasets by default.

For more complex APIs, consider a dedicated query parsing layer or a library, but this is a solid intermediate baseline.

Authentication with JWT

Many REST APIs use stateless authentication with JSON Web Tokens (JWT). The server issues a signed token; the client sends it in the Authorization header.

You will implement:

• POST /api/login to issue a token
• A require_auth decorator to protect endpoints

Security note

This tutorial uses a simple username/password check stored in environment variables for demonstration. In real systems you should store users in a database, hash passwords using a strong algorithm (bcrypt/argon2), enforce rotation, and consider OAuth2/OpenID Connect when appropriate.

Create app/security.py:

import time
from functools import wraps
from flask import request, jsonify, current_app
import jwt

def create_token(subject: str) -> str:
    now = int(time.time())
    payload = {
        "iss": current_app.config["JWT_ISSUER"],
        "sub": subject,
        "iat": now,
        "exp": now + current_app.config["JWT_EXP_SECONDS"],
    }
    secret = current_app.config["SECRET_KEY"]
    return jwt.encode(payload, secret, algorithm="HS256")

def decode_token(token: str) -> dict:
    secret = current_app.config["SECRET_KEY"]
    return jwt.decode(
        token,
        secret,
        algorithms=["HS256"],
        issuer=current_app.config["JWT_ISSUER"],
        options={"require": ["exp", "iat", "iss", "sub"]},
    )

def require_auth(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        auth = request.headers.get("Authorization", "")
        if not auth.startswith("Bearer "):
            return jsonify({"error": {"code": "unauthorized", "message": "Missing Bearer token"}}), 401
        token = auth.removeprefix("Bearer ").strip()
        try:
            decode_token(token)
        except jwt.ExpiredSignatureError:
            return jsonify({"error": {"code": "unauthorized", "message": "Token expired"}}), 401
        except jwt.InvalidTokenError:
            return jsonify({"error": {"code": "unauthorized", "message": "Invalid token"}}), 401
        return fn(*args, **kwargs)
    return wrapper

Create app/auth.py:

import os
from flask import Blueprint, request, jsonify

from .security import create_token

auth_bp = Blueprint("auth", __name__)

@auth_bp.post("/login")
def login():
    data = request.get_json(silent=True) or {}
    username = data.get("username")
    password = data.get("password")

    expected_user = os.getenv("API_USER", "admin")
    expected_pass = os.getenv("API_PASS", "admin")

    if username != expected_user or password != expected_pass:
        return jsonify({"error": {"code": "unauthorized", "message": "Invalid credentials"}}), 401

    token = create_token(subject=username)
    return jsonify({"access_token": token, "token_type": "Bearer"}), 200

Add to .env:

API_USER=admin
API_PASS=admin

Now your tasks endpoints require a valid token.

Running the API Locally

Set Flask app entry point and run:

export FLASK_APP=wsgi:app
export FLASK_ENV=development
flask run --port 5000

PowerShell:

$env:FLASK_APP="wsgi:app"
flask run --port 5000

Get a token

curl -s -X POST http://127.0.0.1:5000/api/login \
  -H "Content-Type: application/json" \
  -d '{"username":"admin","password":"admin"}'

You should receive JSON like:

{"access_token":"<token>","token_type":"Bearer"}

Store it in a shell variable (macOS/Linux):

TOKEN="$(curl -s -X POST http://127.0.0.1:5000/api/login \
  -H "Content-Type: application/json" \
  -d '{"username":"admin","password":"admin"}' | python -c "import sys, json; print(json.load(sys.stdin)['access_token'])")"
echo "$TOKEN"

Create a task

curl -i -X POST http://127.0.0.1:5000/api/tasks \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"title":"Write Flask REST API tutorial","done":false}'

List tasks (with pagination)

curl -s "http://127.0.0.1:5000/api/tasks?page=1&page_size=10&sort=id_desc" \
  -H "Authorization: Bearer $TOKEN" | python -m json.tool

Filter tasks by done status

curl -s "http://127.0.0.1:5000/api/tasks?done=false" \
  -H "Authorization: Bearer $TOKEN" | python -m json.tool

Update a task

curl -i -X PATCH http://127.0.0.1:5000/api/tasks/1 \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"done":true}'

Delete a task

curl -i -X DELETE http://127.0.0.1:5000/api/tasks/1 \
  -H "Authorization: Bearer $TOKEN"

Testing with Pytest

Testing is where intermediate APIs start to feel professional. You want:

• A test app configured for an isolated database
• Tests that call endpoints and assert status codes and JSON

Create tests/test_tasks.py:

import pytest
from app import create_app
from app.config import TestConfig
from app.extensions import db

@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()

def login(client):
    res = client.post("/api/login", json={"username": "admin", "password": "admin"})
    assert res.status_code == 200
    return res.get_json()["access_token"]

def auth_header(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}

def test_create_and_list_tasks(client):
    token = login(client)

    res = client.post("/api/tasks", headers=auth_header(token), json={"title": "Test task", "done": False})
    assert res.status_code == 201
    created = res.get_json()
    assert created["title"] == "Test task"
    assert created["done"] is False

    res = client.get("/api/tasks", headers=auth_header(token))
    assert res.status_code == 200
    data = res.get_json()
    assert data["total"] == 1
    assert len(data["items"]) == 1

def test_requires_auth(client):
    res = client.get("/api/tasks")
    assert res.status_code == 401

Run tests:

pytest -q

Why this setup works:

• TestConfig uses an in-memory SQLite database.
• db.create_all() creates tables for tests without migrations.
• Each test run starts clean.

For larger projects, you might run migrations in tests too, but create_all() is acceptable for many intermediate workflows.

Deployment Notes (Gunicorn)

Flask’s built-in server is for development only. For production, use Gunicorn (Linux/macOS) or a platform-specific server.

Install Gunicorn:

pip install gunicorn==21.2.0

Run:

gunicorn -w 2 -b 0.0.0.0:8000 wsgi:app

Common production considerations:

• Put a reverse proxy in front (Nginx, Caddy, or a managed platform).
• Use a real database (PostgreSQL) instead of SQLite for concurrency.
• Store secrets in environment variables (not .env committed to git).
• Add structured logging and request IDs.
• Consider rate limiting and CORS policies if serving browsers.

Next Steps

To push this API further (still within Flask), consider adding:

1. Pydantic or Marshmallow schemas for robust validation and serialization.
2. OpenAPI documentation (Swagger UI) using libraries like flask-smorest or apispec.
3. Role-based access control (admin vs user).
4. ETags / caching headers for efficient reads.
5. Background jobs (Celery/RQ) for long-running tasks.
6. Observability: metrics (Prometheus), tracing (OpenTelemetry), structured logs.

If you want, share your target domain (e.g., inventory, blog, IoT devices), and the database you plan to use (SQLite/PostgreSQL), and I can adapt this structure into a more domain-specific API with a stronger validation layer and OpenAPI docs.