Python 5 min read

Building REST APIs with Flask and SQLAlchemy

Complete guide to building REST APIs using Flask, SQLAlchemy for ORM, and Marshmallow for serialization. Includes SQLite and MySQL setup.

MR

Moshiour Rahman

#python #flask #sqlalchemy #rest-api #marshmallow #docker

Advertisement

Overview

In this tutorial, we’ll build a complete REST API using Flask - a lightweight Python web framework. We’ll use SQLAlchemy for database operations and Marshmallow for object serialization.

Prerequisites

  • Python 3.7+
  • pip (Python package manager)
  • Basic understanding of REST APIs

Setting Up the Project

Installing Pipenv

Pipenv is a dependency manager for Python that combines pip and virtualenv. If you’re familiar with npm (Node.js) or bundler (Ruby), Pipenv serves a similar purpose.

pip3 install pipenv

Create a project folder and activate the virtual environment:

mkdir rest_flask && cd rest_flask
pipenv shell

Installing Dependencies

pipenv install flask flask-sqlalchemy flask-marshmallow marshmallow-sqlalchemy

This creates a Pipfile (similar to package.json):

[[source]]
name = "pypi"
url = "https://pypi.org/simple"
verify_ssl = true

[packages]
flask = "*"
flask-sqlalchemy = "*"
flask-marshmallow = "*"
marshmallow-sqlalchemy = "*"

[requires]
python_version = "3.11"

Hello World with Flask

Create app.py:

from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/', methods=['GET'])
def index():
    return jsonify({'message': 'Hello TechyOwls!'})

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

Run the server:

python app.py

Test it:

curl http://localhost:5000
# Output: {"message": "Hello TechyOwls!"}

Database Configuration

Option 1: SQLite (Simple Setup)

import os
from flask_sqlalchemy import SQLAlchemy

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

app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{os.path.join(basedir, "db.sqlite")}'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)

Option 2: MySQL with Docker

Create docker-compose.yml:

version: "3.8"
services:
  mysql:
    image: mysql:8.0
    container_name: mysql-flask
    environment:
      MYSQL_USER: techyowls
      MYSQL_PASSWORD: techyowls
      MYSQL_ROOT_PASSWORD: root
      MYSQL_DATABASE: flask_rest
    ports:
      - "3306:3306"
    volumes:
      - mysql-data:/var/lib/mysql
    restart: always

volumes:
  mysql-data:

Start MySQL:

docker-compose up -d

Install the MySQL driver:

pipenv install pymysql

Update the configuration:

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:root@localhost:3306/flask_rest'

Flask-Marshmallow Setup

Marshmallow handles object serialization/deserialization:

from flask_marshmallow import Marshmallow

ma = Marshmallow(app)

Creating the Model

Let’s create a Product model:

class Product(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), unique=True, nullable=False)
    description = db.Column(db.String(200))
    price = db.Column(db.Float, nullable=False)
    quantity = db.Column(db.Integer, default=0)

    def __init__(self, name, description, price, quantity):
        self.name = name
        self.description = description
        self.price = price
        self.quantity = quantity

    def __repr__(self):
        return f'<Product {self.name}>'

Creating the Schema

Define which fields to expose in the JSON response:

class ProductSchema(ma.Schema):
    class Meta:
        fields = ('id', 'name', 'description', 'price', 'quantity')

# Initialize schemas
product_schema = ProductSchema()
products_schema = ProductSchema(many=True)

Creating Tables

Open a Python shell in your project directory:

from app import db
db.create_all()

REST API Endpoints

Create Product (POST)

@app.route('/products', methods=['POST'])
def create_product():
    data = request.get_json()

    new_product = Product(
        name=data['name'],
        description=data.get('description', ''),
        price=data['price'],
        quantity=data.get('quantity', 0)
    )

    db.session.add(new_product)
    db.session.commit()

    return product_schema.jsonify(new_product), 201

Get All Products (GET)

@app.route('/products', methods=['GET'])
def get_all_products():
    products = Product.query.all()
    return products_schema.jsonify(products)

Get Single Product (GET)

@app.route('/products/<int:id>', methods=['GET'])
def get_product(id):
    product = Product.query.get_or_404(id)
    return product_schema.jsonify(product)

Update Product (PUT)

@app.route('/products/<int:id>', methods=['PUT'])
def update_product(id):
    product = Product.query.get_or_404(id)
    data = request.get_json()

    product.name = data.get('name', product.name)
    product.description = data.get('description', product.description)
    product.price = data.get('price', product.price)
    product.quantity = data.get('quantity', product.quantity)

    db.session.commit()
    return product_schema.jsonify(product)

Delete Product (DELETE)

@app.route('/products/<int:id>', methods=['DELETE'])
def delete_product(id):
    product = Product.query.get_or_404(id)
    db.session.delete(product)
    db.session.commit()
    return '', 204

Complete Application

Here’s the full app.py:

import os
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow

# Initialize app
app = Flask(__name__)

# Database configuration
basedir = os.path.abspath(os.path.dirname(__file__))
app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{os.path.join(basedir, "db.sqlite")}'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# Initialize extensions
db = SQLAlchemy(app)
ma = Marshmallow(app)

# Model
class Product(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), unique=True, nullable=False)
    description = db.Column(db.String(200))
    price = db.Column(db.Float, nullable=False)
    quantity = db.Column(db.Integer, default=0)

    def __init__(self, name, description, price, quantity):
        self.name = name
        self.description = description
        self.price = price
        self.quantity = quantity

# Schema
class ProductSchema(ma.Schema):
    class Meta:
        fields = ('id', 'name', 'description', 'price', 'quantity')

product_schema = ProductSchema()
products_schema = ProductSchema(many=True)

# Routes
@app.route('/products', methods=['POST'])
def create_product():
    data = request.get_json()
    new_product = Product(
        name=data['name'],
        description=data.get('description', ''),
        price=data['price'],
        quantity=data.get('quantity', 0)
    )
    db.session.add(new_product)
    db.session.commit()
    return product_schema.jsonify(new_product), 201

@app.route('/products', methods=['GET'])
def get_all_products():
    return products_schema.jsonify(Product.query.all())

@app.route('/products/<int:id>', methods=['GET'])
def get_product(id):
    return product_schema.jsonify(Product.query.get_or_404(id))

@app.route('/products/<int:id>', methods=['PUT'])
def update_product(id):
    product = Product.query.get_or_404(id)
    data = request.get_json()
    product.name = data.get('name', product.name)
    product.description = data.get('description', product.description)
    product.price = data.get('price', product.price)
    product.quantity = data.get('quantity', product.quantity)
    db.session.commit()
    return product_schema.jsonify(product)

@app.route('/products/<int:id>', methods=['DELETE'])
def delete_product(id):
    product = Product.query.get_or_404(id)
    db.session.delete(product)
    db.session.commit()
    return '', 204

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
    app.run(debug=True)

Testing the API

Create a Product

curl -X POST http://localhost:5000/products \
  -H "Content-Type: application/json" \
  -d '{"name": "iPhone 15", "description": "Latest iPhone", "price": 999.99, "quantity": 100}'

Get All Products

curl http://localhost:5000/products

Update a Product

curl -X PUT http://localhost:5000/products/1 \
  -H "Content-Type: application/json" \
  -d '{"price": 899.99}'

Delete a Product

curl -X DELETE http://localhost:5000/products/1

Conclusion

You’ve learned how to build a complete REST API with Flask, including:

  • Setting up Flask with SQLAlchemy and Marshmallow
  • Creating models and schemas
  • Implementing CRUD operations
  • Testing with curl

Source Code: GitHub Repository

In upcoming tutorials, we’ll add authentication, pagination, and error handling.

Advertisement

MR

Moshiour Rahman

Software Architect & AI Engineer

Share:
MR

Moshiour Rahman

Software Architect & AI Engineer

Enterprise software architect with deep expertise in financial systems, distributed architecture, and AI-powered applications. Building large-scale systems at Fortune 500 companies. Specializing in LLM orchestration, multi-agent systems, and cloud-native solutions. I share battle-tested patterns from real enterprise projects.

Related Articles

Python

FastAPI Tutorial: Build Modern Python APIs

Master FastAPI for building high-performance Python APIs. Learn async endpoints, validation, authentication, database integration, and deployment.

Python

FastAPI Tutorial Part 7: CRUD Operations - Build a Complete REST API

Build production-ready CRUD APIs with FastAPI. Learn RESTful patterns, pagination, filtering, bulk operations, and best practices for real-world applications.

Python

FastAPI Tutorial Part 6: Database Integration with SQLAlchemy

Connect FastAPI to databases using SQLAlchemy. Learn ORM models, relationships, migrations with Alembic, and async database operations.

Comments

Comments are powered by GitHub Discussions.

Configure Giscus at giscus.app to enable comments.