How to Create a REST API with Flask and SQLAlchemy

What You Will Build

In this tutorial, you will learn how to build a complete REST API using Flask and SQLAlchemy. By the end, you will have a working API for managing a collection of books — supporting full CRUD operations (Create, Read, Update, Delete) backed by a relational database.

Why Flask and SQLAlchemy?

Flask is a lightweight Python web framework that gives you the freedom to structure your application exactly how you want it. Unlike larger frameworks, Flask does not impose a rigid project layout or include an ORM by default. This makes it an excellent choice for APIs where you want full control without unnecessary overhead.

SQLAlchemy is the most popular Python ORM (Object Relational Mapper). It allows you to interact with your database using Python classes and objects instead of writing raw SQL queries. When combined with Flask-SQLAlchemy — a Flask extension that bridges the two — you get seamless database integration with minimal boilerplate.

Together, Flask and SQLAlchemy give you a fast, readable, and maintainable way to build REST APIs that can scale from a personal project to a production application.

Prerequisites

• Python 3.8 or higher installed on your machine
• Basic understanding of HTTP methods (GET, POST, PUT, DELETE)
• Familiarity with Python and virtual environments
• A code editor (VS Code, PyCharm, or any text editor)

Setting Up Your Environment

Create a new project directory and set up a virtual environment to keep your dependencies isolated:

mkdir flask-rest-api
cd flask-rest-api
python -m venv venv

Activate the virtual environment:

On macOS/Linux:

source venv/bin/activate

On Windows:

venv\Scripts\activate

Install the required packages:

pip install flask flask-sqlalchemy flask-marshmallow marshmallow-sqlalchemy

Here is what each package does:

• flask — The web framework for defining routes and handling requests
• flask-sqlalchemy — Flask extension that simplifies SQLAlchemy configuration
• flask-marshmallow — Flask integration for Marshmallow serialization
• marshmallow-sqlalchemy — Automatic serializer generation from SQLAlchemy models

Create a requirements file so others can recreate your environment:

pip freeze > requirements.txt

Project Structure

A clean project structure helps you stay organized as your API grows. Create the following layout:

flask-rest-api/
├── venv/
├── app.py
├── models.py
├── schemas.py
├── routes.py
├── config.py
└── requirements.txt

Here is what each file will contain:

• app.py — The Flask application entry point
• models.py — SQLAlchemy database models
• schemas.py — Marshmallow serialization schemas
• routes.py — API route definitions (endpoints)
• config.py — Configuration variables (database URI, settings)

Configuration

Start by defining your configuration in config.py. This keeps database credentials and settings in one place:

# config.py
import os

basedir = os.path.abspath(os.path.dirname(__file__))

class Config:
    SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'books.db')
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    JSON_SORT_KEYS = False

Using SQLite for development is convenient because it requires no external database server. The database file books.db will be created automatically in your project folder. For production, you would swap this to a PostgreSQL or MySQL connection string.

SQLALCHEMY_TRACK_MODIFICATIONS is set to False to suppress a warning about the legacy modification tracking system.

Defining the Database Model

Create models.py and define a Book model. This class maps directly to a database table:

# models.py
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

db = SQLAlchemy()

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(150), nullable=False)
    isbn = db.Column(db.String(20), unique=True, nullable=False)
    published_year = db.Column(db.Integer)
    genre = db.Column(db.String(100))
    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""

Key points about the model:

• __tablename__ explicitly sets the table name to books
• primary_key=True marks id as the unique identifier
• nullable=False ensures required fields cannot be empty
• unique=True on isbn prevents duplicate ISBN entries
• The __repr__ method provides a human-readable representation for debugging

Creating Serialization Schemas

Marshmallow schemas convert your Python model objects into JSON (serialization) and validate incoming JSON data (deserialization). Create schemas.py:

# schemas.py
from marshmallow import fields, validate, ValidationError
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema
from models import Book

class BookSchema(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=150))
    isbn = fields.String(required=True, validate=validate.Length(min=10, max=20))

book_schema = BookSchema()
books_schema = BookSchema(many=True)

Notice two schema instances:

• book_schema — Handles a single book object
• books_schema — Handles a list of books (many=True)

The load_instance=True option tells Marshmallow to deserialize directly into a Book model instance, which is extremely convenient when creating or updating records.

Building the API Routes

Now create routes.py — the heart of your API. Each function corresponds to an HTTP endpoint:

# routes.py
from flask import Blueprint, request, jsonify, abort
from models import db, Book
from schemas import book_schema, books_schema
from marshmallow import ValidationError

api = Blueprint('api', __name__)

A Blueprint allows you to organize routes in a modular way. You can later register this blueprint on your Flask app and even prefix all routes with /api.

GET All Books

Retrieve the entire collection of books:

@api.route('/books', methods=['GET'])
def get_books():
    books = Book.query.all()
    result = books_schema.dump(books)
    return jsonify(result)

Book.query.all() fetches every row from the books table. The dump() method serializes the list of model objects into a list of dictionaries, and jsonify() converts that to a proper JSON response with the correct Content-Type header.

GET a Single Book

Fetch one book by its ID:

@api.route('/books/', methods=['GET'])
def get_book(id):
    book = Book.query.get_or_404(id)
    result = book_schema.dump(book)
    return jsonify(result)

get_or_404(id) is a Flask-SQLAlchemy shortcut that automatically looks up the record and calls abort(404) if it does not exist. This keeps your code clean and avoids manual if book is None checks.

POST Create a New Book

Create a new book from JSON data sent in the request body:

@api.route('/books', methods=['POST'])
def create_book():
    json_data = request.get_json()

    if not json_data:
        abort(400, description="No JSON data provided")

    try:
        book = book_schema.load(json_data)
    except ValidationError as err:
        return jsonify({"errors": err.messages}), 400

    db.session.add(book)
    db.session.commit()

    result = book_schema.dump(book)
    return jsonify(result), 201

Step-by-step breakdown:

• request.get_json() parses the incoming JSON body
• If no JSON is sent, a 400 Bad Request response is returned
• book_schema.load() validates and deserializes the JSON into a Book instance
• If validation fails, Marshmallow raises ValidationError — we catch it and return the specific error messages with a 400 status code
• db.session.add() stages the new book for insertion
• db.session.commit() writes the transaction to the database
• The created book is serialized back to JSON and returned with status 201 Created

PUT Update an Existing Book

Update one or more fields of an existing book:

@api.route('/books/', methods=['PUT'])
def update_book(id):
    book = Book.query.get_or_404(id)
    json_data = request.get_json()

    if not json_data:
        abort(400, description="No JSON data provided")

    try:
        updated_book = book_schema.load(json_data, instance=book, partial=True)
    except ValidationError as err:
        return jsonify({"errors": err.messages}), 400

    db.session.commit()

    result = book_schema.dump(updated_book)
    return jsonify(result)

Important details:

• The instance=book parameter tells Marshmallow to apply the incoming data to an existing model instance rather than creating a new one
• partial=True allows partial updates — clients can send only the fields they want to change without needing to include every required field
• After committing, the updated book is returned with a 200 status code (the default for Flask responses)

DELETE a Book

Remove a book from the database:

@api.route('/books/', methods=['DELETE'])
def delete_book(id):
    book = Book.query.get_or_404(id)
    db.session.delete(book)
    db.session.commit()
    return jsonify({"message": f"Book '{book.title}' deleted successfully"}), 200

The response includes a confirmation message with the title of the deleted book so the client knows exactly which resource was removed.

Full routes.py File

Here is the complete routes file for reference:

# routes.py
from flask import Blueprint, request, jsonify, abort
from models import db, Book
from schemas import book_schema, books_schema
from marshmallow import ValidationError

api = Blueprint('api', __name__)

@api.route('/books', methods=['GET'])
def get_books():
    books = Book.query.all()
    result = books_schema.dump(books)
    return jsonify(result)

@api.route('/books/', methods=['GET'])
def get_book(id):
    book = Book.query.get_or_404(id)
    result = book_schema.dump(book)
    return jsonify(result)

@api.route('/books', methods=['POST'])
def create_book():
    json_data = request.get_json()
    if not json_data:
        abort(400, description="No JSON data provided")
    try:
        book = book_schema.load(json_data)
    except ValidationError as err:
        return jsonify({"errors": err.messages}), 400
    db.session.add(book)
    db.session.commit()
    result = book_schema.dump(book)
    return jsonify(result), 201

@api.route('/books/', methods=['PUT'])
def update_book(id):
    book = Book.query.get_or_404(id)
    json_data = request.get_json()
    if not json_data:
        abort(400, description="No JSON data provided")
    try:
        updated_book = book_schema.load(json_data, instance=book, partial=True)
    except ValidationError as err:
        return jsonify({"errors": err.messages}), 400
    db.session.commit()
    result = book_schema.dump(updated_book)
    return jsonify(result)

@api.route('/books/', methods=['DELETE'])
def delete_book(id):
    book = Book.query.get_or_404(id)
    db.session.delete(book)
    db.session.commit()
    return jsonify({"message": f"Book '{book.title}' deleted successfully"}), 200

Assembling the Flask Application

The app.py file ties everything together. It creates the Flask app, loads configuration, initializes the database, and registers the API blueprint:

# app.py
from flask import Flask
from config import Config
from models import db
from routes import api

def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)

    db.init_app(app)

    with app.app_context():
        db.create_all()

    app.register_blueprint(api, url_prefix='/api')

    return app

if __name__ == '__main__':
    app = create_app()
    app.run(debug=True)

Key elements explained:

• app.config.from_object(Config) loads all uppercase variables from the Config class
• db.init_app(app) binds SQLAlchemy to the Flask application
• db.create_all() runs inside an application context and creates all tables defined by your models if they do not exist yet
• url_prefix='/api' means all book endpoints will be accessible under /api/books rather than /books
• debug=True enables hot-reloading and an interactive debugger during development

Running and Testing the API

Start the development server:

python app.py

You should see output similar to:

* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)

Now test the endpoints using curl, Postman, or any HTTP client:

Create a new book (POST)

curl -X POST http://127.0.0.1:5000/api/books \
  -H "Content-Type: application/json" \
  -d '{"title": "1984", "author": "George Orwell", "isbn": "978-0451524935", "published_year": 1949, "genre": "Dystopian"}'

Expected response (201 Created):

{
  "id": 1,
  "title": "1984",
  "author": "George Orwell",
  "isbn": "978-0451524935",
  "published_year": 1949,
  "genre": "Dystopian",
  "created_at": "2025-01-15T14:30:00",
  "updated_at": "2025-01-15T14:30:00"
}

Get all books (GET)

curl http://127.0.0.1:5000/api/books

Get a single book (GET)

curl http://127.0.0.1:5000/api/books/1

Update a book (PUT)

curl -X PUT http://127.0.0.1:5000/api/books/1 \
  -H "Content-Type: application/json" \
  -d '{"published_year": 1950}'

Delete a book (DELETE)

curl -X DELETE http://127.0.0.1:5000/api/books/1

Adding Query Filtering and Pagination

Real-world APIs need filtering and pagination. Add a search endpoint that supports query parameters:

@api.route('/books/search', methods=['GET'])
def search_books():
    # Get query parameters with defaults
    author = request.args.get('author')
    genre = request.args.get('genre')
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 10, type=int)

    # Build the query dynamically
    query = Book.query

    if author:
        query = query.filter(Book.author.ilike(f'%{author}%'))
    if genre:
        query = query.filter(Book.genre == genre)

    # Paginate results
    pagination = query.paginate(page=page, per_page=per_page, error_out=False)
    books = pagination.items

    result = {
        'books': books_schema.dump(books),
        'total': pagination.total,
        'pages': pagination.pages,
        'current_page': page,
        'has_next': pagination.has_next,
        'has_prev': pagination.has_prev
    }

    return jsonify(result)

This endpoint demonstrates several production-ready patterns:

• Optional query parameters with sensible defaults
• Dynamic query building — filters are only applied when the parameter is present
• ilike() enables case-insensitive partial matching for text searches
• Pagination metadata lets clients know how many total results exist and whether there are more pages

Error Handling Best Practices

A robust API should have consistent error responses. Add a global error handler in app.py:

@app.errorhandler(404)
def not_found(error):
    return jsonify({"error": "Resource not found", "status": 404}), 404

@app.errorhandler(400)
def bad_request(error):
    return jsonify({"error": str(error.description), "status": 400}), 400

@app.errorhandler(500)
def internal_error(error):
    db.session.rollback()
    return jsonify({"error": "Internal server error", "status": 500}), 500

Register these handlers inside the create_app() function after creating the app instance. The db.session.rollback() call in the 500 handler prevents partial database writes from leaving the database in an inconsistent state.

Database Migrations for Production

While db.create_all() works for development, production applications need proper database migrations to evolve the schema without data loss. Install Flask-Migrate (which wraps Alembic):

pip install flask-migrate

Initialize migrations in app.py:

from flask_migrate import Migrate

migrate = Migrate(app, db)

Then run these commands in your terminal:

flask db init
flask db migrate -m "Initial migration"
flask db upgrade

From this point forward, whenever you modify your models, run flask db migrate and flask db upgrade to apply changes incrementally.

Best Practices Summary

• Use Blueprints — Organize routes into logical groups (e.g., api, auth, admin)
• Validate input early — Marshmallow validation catches bad data before it reaches your database layer
• Return proper HTTP status codes — 201 for creation, 204 for no-content responses, 400 for client errors, 404 for missing resources
• Use consistent JSON responses — Always return JSON objects (or arrays) with predictable key names
• Handle exceptions globally — Centralized error handlers give clients uniform error formats
• Paginate large collections — Never return thousands of records in a single response; use page-based or cursor-based pagination
• Use environment variables for configuration — Keep database URIs and secrets out of source code with os.environ.get()
• Write tests — Use pytest and Flask's test client to verify your endpoints behave correctly
• Log requests — Add logging middleware to track incoming requests, response times, and errors
• Use proper HTTP semantics — GET requests should be idempotent and never modify data; PUT should fully replace a resource; PATCH should partially update

Complete app.py with Error Handlers

Here is the final app.py combining everything we have covered:

# app.py
from flask import Flask, jsonify
from config import Config
from models import db
from routes import api

def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)

    db.init_app(app)

    # Register error handlers
    @app.errorhandler(404)
    def not_found(error):
        return jsonify({"error": "Resource not found", "status": 404}), 404

    @app.errorhandler(400)
    def bad_request(error):
        return jsonify({"error": str(error.description), "status": 400}), 400

    @app.errorhandler(500)
    def internal_error(error):
        db.session.rollback()
        return jsonify({"error": "Internal server error", "status": 500}), 500

    with app.app_context():
        db.create_all()

    app.register_blueprint(api, url_prefix='/api')

    return app

if __name__ == '__main__':
    app = create_app()
    app.run(debug=True)

Conclusion

You have now built a fully functional REST API using Flask and SQLAlchemy. You learned how to define database models, serialize them with Marshmallow, implement all five CRUD endpoints (GET collection, GET single, POST, PUT, DELETE), handle validation errors gracefully, and structure your project for maintainability. You also explored advanced topics like query filtering, pagination, database migrations, and global error handling.

This foundation scales naturally — you can add authentication with Flask-JWT-Extended, implement rate limiting, containerize the application with Docker, and deploy it to cloud platforms like AWS, Google Cloud, or Render. The patterns you learned here (Blueprint organization, schema-based validation, consistent error responses) apply universally to any Flask API you build in the future. Start extending this base with your own models and endpoints, and you will quickly see how Flask's simplicity combined with SQLAlchemy's power creates an exceptionally productive development experience.