Flask and SQLAlchemy form a powerful combination for building robust REST APIs in Python. Flask provides a lightweight and flexible web framework, while SQLAlchemy offers a sophisticated Object-Relational Mapping (ORM) system that simplifies database operations. Together, they enable developers to create efficient, maintainable, and scalable web applications.

Benefits of using SQLAlchemy with Flask

• Simplified database operations through Python objects
• Powerful query interface
• Database agnostic development
• Automatic schema management
• Rich relationship mapping

In this tutorial, we’ll build a complete REST API with CRUD operations, incorporating best practices for security, performance, and maintainability.

Prerequisites

• Python 3.7+
• Basic understanding of Python
• Familiarity with REST APIs and Flask
• Basic SQL knowledge

Project Setup and Configuration

Project Structure Overview

Our project follows a modular structure for better organization:

project/
├── app/
│   ├── __init__.py
│   ├── models.py
│   ├── routes.py
│   └── config.py
├── run.py
└── requirements.txt


This structure separates concerns by placing models, routes, and configuration in different files, making the codebase easier to maintain as it grows. The __init__.py file serves as the application factory, models.py contains database models and schemas, routes.py defines API endpoints, and config.py handles different environment configurations. The run.py at the root level serves as the entry point for running the application.

Installing Required Dependencies

First, let’s create our project directory and set up our environment:

# Create project directory
mkdir flask-sqlalchemy-api
cd flask-sqlalchemy-api

# Create app package directory
mkdir -p app

# Create a virtual environment and activate it
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install required packages
pip install flask flask-sqlalchemy flask-marshmallow marshmallow-sqlalchemy flask-jwt-extended flask-cors flask-restx pytest


We’re installing:

• flask: The web framework itself that handles routing, request processing, and responses
• flask-sqlalchemy: Flask extension that adds SQLAlchemy ORM support with Flask-specific helpers
• flask-marshmallow: Integration layer for Flask and Marshmallow for object serialization/deserialization
• marshmallow-sqlalchemy: SQLAlchemy integration with Marshmallow that provides automatic schema generation
• flask-jwt-extended: For JSON Web Token authentication to secure API endpoints
• flask-cors: For handling Cross-Origin Resource Sharing, allowing controlled frontend access
• flask-restx: For API documentation with Swagger UI and enhanced API development tools
• pytest: For running tests

Basic Flask Application Setup

Let’s start by creating our Flask application with SQLAlchemy integration:

# app/__init__.py
from flask import Flask
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()

def create_app(config=None):
    app = Flask(__name__)
    app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///database.db'
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    app.config['JWT_SECRET_KEY'] = 'your-secret-key'  # Change this in production!
    
    # Initialize extensions
    db.init_app(app)
    ma.init_app(app)
    jwt = JWTManager(app)
    
    # Configure CORS
    CORS(app, resources={
        r"/api/*": {
            "origins": ["http://localhost:3000"],
            "methods": ["GET", "POST", "PUT", "DELETE"],
            "allow_headers": ["Content-Type"]
        }
    })
    
    # Create a test route
    @app.route('/ping')
    def ping():
        return {'message': 'pong'}
    
    return app


Let’s create a minimal run file to test our setup:

# run.py
from app import create_app

app = create_app()

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


Test Your Basic Setup

Let’s make sure our basic application works:

# Run the Flask application
python run.py


You should see output similar to:

* Serving Flask app...
* Debug mode: on
* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)


Now, open a new terminal window and test the ping endpoint:

curl http://localhost:5000/ping


You should see the response:

{"message":"pong"}


If you see this response, congratulations! Your basic Flask application is running. Press Ctrl+C in the terminal running your application to stop it before continuing.

Database Models and Schema Design

Creating SQLAlchemy Models

Let’s create the models.py file with our User model:

# app/models.py
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)
    
    def __init__(self, username, email):
        self.username = username
        self.email = email


This User model:

• Inherits from db.Model to get SQLAlchemy ORM functionality
• Defines an auto-incrementing integer primary key (id)
• Specifies a username field with max length 80, marked as unique and required
• Creates an email field with max length 120, also unique and required
• Provides a constructor to easily instantiate User objects with username and email

Implementing Marshmallow Schemas

Now, let’s add a serialization schema to our models.py file:

# app/models.py - add to the existing file
from app import ma
from marshmallow import validates, ValidationError, fields, validate

class UserSchema(ma.Schema):
    # Explicitly define fields to avoid KeyError
    id = fields.Integer(dump_only=True)  # Read-only field
    username = fields.String(required=True, validate=[
        validate.Length(min=3, error="Username must be at least {min} characters"),
        validate.Regexp(r'^[a-zA-Z0-9_]+$', error="Username can only contain letters, numbers, and underscores")
    ], error_messages={
        'required': 'Username is required',
        'invalid': 'Invalid username format'
    })
    email = fields.Email(required=True, error_messages={
        'required': 'Email is required',
        'invalid': 'Not a valid email address'
    })
    
    class Meta:
        fields = ('id', 'username', 'email')

user_schema = UserSchema()
users_schema = UserSchema(many=True)


This schema:

• Explicitly defines fields to avoid KeyError exceptions
• Uses Marshmallow’s built-in validators for more comprehensive validation
• Sets the ID field as dump_only (for serialization output only)
• Creates both single object and collection serializers

For more information on Flask-Marshmallow read our post Using Flask and Marshmallow.

Initialize the Database

Let’s modify our run.py file to create the database tables on startup:

# run.py
from app import create_app, db

app = create_app()

# Create a context for database operations
with app.app_context():
    db.create_all()
    print("Database tables created!")

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


Run this script to create your database:

python run.py


You should see “Database tables created!” in your console. You can stop the server again with Ctrl+C.

Implementing CRUD Operations

Setting Up the Routes File

Let’s create our routes file with a blueprint and helper functions:

# app/routes.py
from flask import Blueprint, request, jsonify
from app import db
from app.models import User, user_schema, users_schema

# Create a blueprint
api_bp = Blueprint('api', __name__)

# Helper functions
def process_request():
    if not request.is_json:
        return jsonify({'error': 'Content-Type must be application/json'}), 400
        
    data = request.get_json()
    if not all(k in data for k in ('username', 'email')):
        return jsonify({'error': 'Missing required fields'}), 400
        
    return data

def create_response(data=None, message=None, error=None, status=200):
    response = {
        'success': error is None,
        'message': message,
        'data': data,
        'error': error
    }
    return jsonify(response), status


Create Operation

Add the create endpoint to our routes.py file:

# app/routes.py - add this function
@api_bp.route('/user', methods=['POST'])
def create_user():
    try:
        # Validate request data
        result = process_request()
        if isinstance(result, tuple):  # Is an error response
            return result
            
        username = request.json['username']
        email = request.json['email']
        
        new_user = User(username=username, email=email)
        db.session.add(new_user)
        db.session.commit()
        
        return user_schema.jsonify(new_user)
    except Exception as e:
        return jsonify({'error': str(e)}), 400


Read Operations

Add the read endpoints to our routes.py file:

# app/routes.py - add these functions
# Get all users
@api_bp.route('/users', methods=['GET'])
def get_users():
    users = User.query.all()
    return users_schema.jsonify(users)

# Get single user
@api_bp.route('/user/<id>', methods=['GET'])
def get_user(id):
    user = User.query.get_or_404(id)
    return user_schema.jsonify(user)


Update Operation

Add the update endpoint to our routes.py file:

# app/routes.py - add this function
@api_bp.route('/user/<id>', methods=['PUT'])
def update_user(id):
    user = User.query.get_or_404(id)
    
    username = request.json.get('username', user.username)
    email = request.json.get('email', user.email)
    
    user.username = username
    user.email = email
    
    db.session.commit()
    return user_schema.jsonify(user)


Delete Operation

Add the delete endpoint to our routes.py file:

# app/routes.py - add this function
@api_bp.route('/user/<id>', methods=['DELETE'])
def delete_user(id):
    user = User.query.get_or_404(id)
    db.session.delete(user)
    db.session.commit()
    return jsonify({'message': 'User deleted successfully'})


Error Handlers

Add error handlers to our routes.py file:

# app/routes.py - add these functions
@api_bp.errorhandler(404)
def not_found(error):
    return create_response(error='Resource not found', status=404)

@api_bp.errorhandler(400)
def bad_request(error):
    return create_response(error='Bad request', status=400)

@api_bp.errorhandler(500)
def server_error(error):
    return create_response(error='Internal server error', status=500)


Register the Blueprint

Now update the init.py file to register our blueprint:

# app/__init__.py - modify the create_app function
def create_app(config=None):
    app = Flask(__name__)
    app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///database.db'
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    app.config['JWT_SECRET_KEY'] = 'your-secret-key'  # Change this in production!
    
    # Initialize extensions
    db.init_app(app)
    ma.init_app(app)
    jwt = JWTManager(app)
    
    # Configure CORS
    CORS(app, resources={
        r"/api/*": {
            "origins": ["http://localhost:3000"],
            "methods": ["GET", "POST", "PUT", "DELETE"],
            "allow_headers": ["Content-Type"]
        }
    })
    
    # Import and register blueprints
    from app.routes import api_bp
    app.register_blueprint(api_bp, url_prefix='/api')
    
    return app


Testing Your CRUD Endpoints

Let’s start the application and test our endpoints:

python run.py


Now, in a new terminal, let’s test each endpoint:

1. Create a user:

curl -X POST http://localhost:5000/api/user \
     -H "Content-Type: application/json" \
     -d '{"username": "john_doe", "email": "john@example.com"}'


You should see a response with the created user, including the auto-generated ID:

{"email":"john@example.com","id":1,"username":"john_doe"}


2. Get all users:

curl http://localhost:5000/api/users


Response:

[{"email":"john@example.com","id":1,"username":"john_doe"}]


3. Get a specific user:

curl http://localhost:5000/api/user/1


Response:

{"email":"john@example.com","id":1,"username":"john_doe"}


4. Update a user:

curl -X PUT http://localhost:5000/api/user/1 \
     -H "Content-Type: application/json" \
     -d '{"username": "john_updated", "email": "john_new@example.com"}'


Response:

{"email":"john_new@example.com","id":1,"username":"john_updated"}


5. Delete a user:

curl -X DELETE http://localhost:5000/api/user/1


Response:

{"message":"User deleted successfully"}


6. Verify deletion:

curl http://localhost:5000/api/users


Response should be an empty array:

[]


Great! Your CRUD operations are working as expected. Press Ctrl+C to stop the server.

Security Implementation

Authentication Setup

Let’s add a login endpoint and a protected route to our routes.py:

# app/routes.py - add these imports and functions
from flask_jwt_extended import jwt_required, create_access_token

@api_bp.route('/login', methods=['POST'])
def login():
    if not request.is_json:
        return jsonify({'error': 'Missing JSON in request'}), 400
        
    username = request.json.get('username', None)
    password = request.json.get('password', None)
    
    if not username or not password:
        return jsonify({'error': 'Missing username or password'}), 400
    
    # In a real application, you would validate against a database
    # For this tutorial, we'll accept any credentials
    
    access_token = create_access_token(identity=username)
    return jsonify({'access_token': access_token})

@api_bp.route('/protected', methods=['GET'])
@jwt_required()
def protected():
    return jsonify({'message': 'This is a protected endpoint'})


Testing Authentication

Let’s run the application again:

python run.py


In a new terminal:

1. Login to get a token:

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


You should receive a JWT token:

{"access_token":"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9..."}


2. Access a protected endpoint:

Use the token from the previous step:

curl http://localhost:5000/api/protected \
     -H "Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9..."


You should see:

{"message":"This is a protected endpoint"}


3. Try accessing without a token:

curl http://localhost:5000/api/protected


You should get an error:

{"msg":"Missing Authorization Header"}


Great! Your authentication system is working. Press Ctrl+C to stop the server.

API Documentation with Swagger

Let’s integrate Flask-RESTX to get Swagger documentation automatically. Create a new file called restx.py:

# app/restx.py
from flask import Blueprint
from flask_restx import Api, Resource, fields
from app import db
from app.models import User, user_schema, users_schema
from flask_jwt_extended import jwt_required

# Create a blueprint
api_bp = Blueprint('api_doc', __name__)

# Create Flask-RESTX API
api = Api(
    api_bp, 
    version='1.0', 
    title='User Management API',
    description='A simple User API with SQLAlchemy',
    doc='/docs'
)

# Create namespaces
ns_users = api.namespace('users', description='User operations')
ns_auth = api.namespace('auth', description='Authentication operations')

# Models for documentation
user_model = api.model('User', {
    'id': fields.Integer(readonly=True, description='User identifier'),
    'username': fields.String(required=True, description='Username'),
    'email': fields.String(required=True, description='Email address')
})

user_input_model = api.model('UserInput', {
    'username': fields.String(required=True, description='Username'),
    'email': fields.String(required=True, description='Email address')
})

login_model = api.model('Login', {
    'username': fields.String(required=True, description='Username'),
    'password': fields.String(required=True, description='Password')
})

token_model = api.model('Token', {
    'access_token': fields.String(description='JWT Access token')
})

# User routes
@ns_users.route('/')
class UserList(Resource):
    @ns_users.doc('list_users')
    @ns_users.marshal_list_with(user_model)
    def get(self):
        """List all users"""
        users = User.query.all()
        return users
        
    @ns_users.doc('create_user')
    @ns_users.expect(user_input_model)
    @ns_users.marshal_with(user_model, code=201)
    @ns_users.response(400, 'Validation Error')
    def post(self):
        """Create a new user"""
        try:
            data = api.payload
            new_user = User(username=data['username'], email=data['email'])
            db.session.add(new_user)
            db.session.commit()
            
            return new_user, 201
        except Exception as e:
            api.abort(400, str(e))

@ns_users.route('/<int:id>')
@ns_users.param('id', 'The user identifier')
@ns_users.response(404, 'User not found')
class UserResource(Resource):
    @ns_users.doc('get_user')
    @ns_users.marshal_with(user_model)
    def get(self, id):
        """Get a specific user by ID"""
        user = User.query.get_or_404(id)
        return user
        
    @ns_users.doc('update_user')
    @ns_users.expect(user_input_model)
    @ns_users.marshal_with(user_model)
    def put(self, id):
        """Update a user"""
        user = User.query.get_or_404(id)
        
        data = api.payload
        if 'username' in data:
            user.username = data['username']
        if 'email' in data:
            user.email = data['email']
        
        db.session.commit()
        return user
    
    @ns_users.doc('delete_user')
    @ns_users.response(204, 'User deleted')
    def delete(self, id):
        """Delete a user"""
        user = User.query.get_or_404(id)
        db.session.delete(user)
        db.session.commit()
        return '', 204

# Authentication routes
@ns_auth.route('/login')
class Login(Resource):
    @ns_auth.doc('user_login')
    @ns_auth.expect(login_model)
    @ns_auth.marshal_with(token_model)
    @ns_auth.response(400, 'Invalid credentials')
    def post(self):
        """Log in and receive an access token"""
        from flask_jwt_extended import create_access_token
        
        data = api.payload
        username = data.get('username')
        password = data.get('password')
        
        # In a real app, validate credentials against database
        # For this tutorial, we accept any credentials
        
        access_token = create_access_token(identity=username)
        return {'access_token': access_token}


Now, update your init.py file to register this blueprint:

# app/__init__.py - modify create_app to include the new blueprint
def create_app(config=None):
    # Existing app setup code...
    
    # Import and register blueprints
    from app.routes import api_bp
    app.register_blueprint(api_bp, url_prefix='/api')
    
    # Import and register the RESTx blueprint
    from app.restx import api_bp as api_doc_bp
    app.register_blueprint(api_doc_bp, url_prefix='/api/v1')
    
    return app


Testing the Swagger UI

Start your application:

python run.py


Open your browser and navigate to:

http://localhost:5000/api/v1/docs


You should see the Swagger UI where you can:

• Explore all your API endpoints
• Test them interactively
• See detailed documentation
• Understand what models are expected for each endpoint

Try creating a user directly from the Swagger UI:

1. Expand the POST /api/v1/users/ endpoint
2. Click “Try it out”
3. Enter a username and email in the request body
4. Click “Execute”
5. Observe the response

Press Ctrl+C to stop the server when you’re done exploring.

Unit Testing

Let’s create a test file to test our API:

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

@pytest.fixture
def client():
    """Create a test client for the app."""
    app = create_app('testing')
    app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
    app.config['TESTING'] = True
    
    with app.test_client() as client:
        with app.app_context():
            db.create_all()
        yield client
        
        # teardown
        with app.app_context():
            db.drop_all()

def test_create_user(client):
    """Test creating a user."""
    response = client.post(
        '/api/user',
        data=json.dumps({'username': 'testuser', 'email': 'test@example.com'}),
        content_type='application/json'
    )
    data = json.loads(response.data)
    assert response.status_code == 200
    assert data['username'] == 'testuser'
    assert data['email'] == 'test@example.com'

def test_get_users(client):
    """Test getting all users."""
    # First create a user
    client.post(
        '/api/user',
        data=json.dumps({'username': 'testuser', 'email': 'test@example.com'}),
        content_type='application/json'
    )
    
    # Then get all users
    response = client.get('/api/users')
    data = json.loads(response.data)
    assert response.status_code == 200
    assert isinstance(data, list)
    assert len(data) == 1
    assert data[0]['username'] == 'testuser'

def test_get_user(client):
    """Test getting a specific user."""
    # First create a user
    client.post(
        '/api/user',
        data=json.dumps({'username': 'testuser', 'email': 'test@example.com'}),
        content_type='application/json'
    )
    
    # Then get the user
    response = client.get('/api/user/1')
    data = json.loads(response.data)
    assert response.status_code == 200
    assert data['username'] == 'testuser'

def test_update_user(client):
    """Test updating a user."""
    # First create a user
    client.post(
        '/api/user',
        data=json.dumps({'username': 'testuser', 'email': 'test@example.com'}),
        content_type='application/json'
    )
    
    # Then update the user
    response = client.put(
        '/api/user/1',
        data=json.dumps({'username': 'updated', 'email': 'updated@example.com'}),
        content_type='application/json'
    )
    data = json.loads(response.data)
    assert response.status_code == 200
    assert data['username'] == 'updated'
    assert data['email'] == 'updated@example.com'

def test_delete_user(client):
    """Test deleting a user."""
    # First create a user
    client.post(
        '/api/user',
        data=json.dumps({'username': 'testuser', 'email': 'test@example.com'}),
        content_type='application/json'
    )
    
    # Then delete the user
    response = client.delete('/api/user/1')
    assert response.status_code == 200
    
    # Verify the user was deleted
    response = client.get('/api/user/1')
    assert response.status_code == 404


Create a pytest.ini file at the root of your project:

# pytest.ini
[pytest]
testpaths = tests


Make sure to create a tests directory and make it a Python package:

mkdir -p tests
touch tests/__init__.py


Running the Tests

Now run your tests:

pytest -v


You should see all your tests pass with detailed output:

test_api.py::test_create_user PASSED
test_api.py::test_get_users PASSED
test_api.py::test_get_user PASSED
test_api.py::test_update_user PASSED
test_api.py::test_delete_user PASSED


Performance Optimization

Let’s add pagination to our users endpoint in both routes.py and restx.py:

# app/routes.py - modify the get_users function
@api_bp.route('/users', methods=['GET'])
def get_users():
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 10, type=int)
    
    pagination = User.query.paginate(page=page, per_page=per_page)
    users = pagination.items
    
    # Add pagination metadata
    meta = {
        'page': pagination.page,
        'per_page': pagination.per_page,
        'total': pagination.total,
        'pages': pagination.pages
    }
    
    result = users_schema.dump(users)
    return jsonify({
        'users': result,
        'meta': meta
    })


And update the Flask-RESTX implementation:

# app/restx.py - modify the UserList.get method
@ns_users.route('/')
class UserList(Resource):
    @ns_users.doc('list_users')
    @ns_users.param('page', 'Page number', type=int, default=1)
    @ns_users.param('per_page', 'Items per page', type=int, default=10)
    def get(self):
        """List all users with pagination"""
        from flask import request
        
        page = request.args.get('page', 1, type=int)
        per_page = request.args.get('per_page', 10, type=int)
        
        pagination = User.query.paginate(page=page, per_page=per_page)
        users = pagination.items
        
        # Create response with pagination metadata
        result = {
            'users': users_schema.dump(users),
            'meta': {
                'page': pagination.page,
                'per_page': pagination.per_page,
                'total': pagination.total,
                'pages': pagination.pages
            }
        }
        
        return result


Testing Pagination

Start your application:

python run.py


Let’s first create several users:

# Create 15 users in a loop
for i in {1..15}; do
  curl -X POST http://localhost:5000/api/user \
       -H "Content-Type: application/json" \
       -d "{\"username\": \"user$i\", \"email\": \"user$i@example.com\"}"
done


Now, test pagination with different parameters:

# Get the first page (default 10 items)
curl http://localhost:5000/api/users

# Get the second page
curl http://localhost:5000/api/users?page=2

# Get 5 items per page
curl http://localhost:5000/api/users?page=1&per_page=5


Check the response to see the pagination metadata along with the users for each request.

Press Ctrl+C to stop the server when done testing.

Conclusion

Congratulations! You’ve built a complete REST API using Flask and SQLAlchemy that includes:

• Database operations and modeling
• API design following RESTful principles
• Security and authentication with JWT
• Error handling and validation
• Interactive API documentation with Swagger
• Unit testing with pytest
• Performance optimization with pagination

Remember to always validate your data before saving it to the database and provide clear error messages to improve the developer experience of your API.