Python Flask Rest Api With Sqlalchemy

Written by

in

Building a modern web service often starts with a clear, maintainable API that can talk to a database efficiently. Python Flask paired with SQLAlchemy offers exactly that: a lightweight framework for routing HTTP requests and a powerful ORM for handling relational data. In this guide we’ll walk through every step needed to create a production‑ready REST API using Flask and SQLAlchemy, from project setup to testing and deployment. Whether you’re a beginner eager to see a working example or a seasoned developer looking for best‑practice tips, this tutorial has you covered.

Why Choose Flask and SQLAlchemy for a REST API?

Before diving into code, it’s worth understanding the advantages that make Flask + SQLAlchemy a popular stack for RESTful services:

  • Minimalistic core: Flask provides just the essentials—routing, request handling, and extensions—so you stay in control of your architecture.
  • Extensible ecosystem: Extensions like Flask‑RESTful, Flask‑Migrate, and Flask‑JWT‑Extended add features without bloat.
  • SQLAlchemy’s ORM: Write Python classes instead of raw SQL, enjoy automatic schema migrations, and benefit from a database‑agnostic layer.
  • Great community and documentation: Both projects have extensive tutorials, Stack Overflow answers, and production‑grade examples.

Project Structure

A clean folder layout helps you scale the API as it grows. Below is a recommended structure for a Flask‑SQLAlchemy project:

my_flask_api/
│
├── app/
│   ├── __init__.py        # Flask app factory
│   ├── models.py          # SQLAlchemy models
│   ├── resources.py       # API endpoint definitions
│   ├── schemas.py         # Marshmallow schemas (optional)
│   └── config.py          # Configuration settings
│
├── migrations/            # Alembic migration scripts
│
├── tests/
│   └── test_api.py
│
├── venv/                  # Virtual environment (optional)
├── requirements.txt
└── run.py                 # Entry point

Step‑by‑Step Implementation

1. Set Up the Development Environment

First, create a virtual environment and install the required packages.

python -m venv venv
source venv/bin/activate   # On Windows: venv\Scripts\activate
pip install Flask Flask‑SQLAlchemy Flask‑RESTful Flask‑Migrate Flask‑JWT‑Extended marshmallow
pip freeze > requirements.txt

2. Create the Flask Application Factory

Using an application factory makes testing and configuration easier.

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

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

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

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

    # Register API resources
    api = Api(app)
    from .resources import UserListResource, UserResource
    api.add_resource(UserListResource, '/api/users')
    api.add_resource(UserResource, '/api/users/<int:user_id>')

    return app

3. Define Configuration Settings

Keep secret keys and database URLs out of source control.

# app/config.py
import os

class Config:
    SECRET_KEY = os.getenv('SECRET_KEY', 'super-secret-key')
    SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL', 'sqlite:///app.db')
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY', 'jwt-secret-key')

4. Model Your Data with SQLAlchemy

Below is a simple User model that includes password hashing using werkzeug.security.

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

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(128), nullable=False)

    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
        }

5. Create Marshmallow Schemas (Optional but Recommended)

Marshmallow handles serialization and validation cleanly.

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

class UserSchema(Schema):
    id = fields.Int(dump_only=True)
    username = fields.Str(required=True, validate=validate.Length(min=3))
    email = fields.Email(required=True)
    password = fields.Str(load_only=True, required=True, validate=validate.Length(min=6))

6. Implement RESTful Resources

Using Flask‑RESTful, define endpoints for CRUD operations.

# app/resources.py
from flask import request, jsonify
from flask_restful import Resource
from flask_jwt_extended import jwt_required, create_access_token
from . import db
from .models import User
from .schemas import UserSchema

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

class UserListResource(Resource):
    @jwt_required()
    def get(self):
        users = User.query.all()
        return users_schema.dump(users), 200

    def post(self):
        data = request.get_json()
        errors = user_schema.validate(data)
        if errors:
            return errors, 400

        user = User(
            username=data['username'],
            email=data['email']
        )
        user.set_password(data['password'])
        db.session.add(user)
        db.session.commit()
        return user_schema.dump(user), 201

class UserResource(Resource):
    @jwt_required()
    def get(self, user_id):
        user = User.query.get_or_404(user_id)
        return user_schema.dump(user), 200

    @jwt_required()
    def put(self, user_id):
        user = User.query.get_or_404(user_id)
        data = request.get_json()
        errors = user_schema.validate(data, partial=True)
        if errors:
            return errors, 400

        if 'username' in data:
            user.username = data['username']
        if 'email' in data:
            user.email = data['email']
        if 'password' in data:
            user.set_password(data['password'])

        db.session.commit()
        return user_schema.dump(user), 200

    @jwt_required()
    def delete(self, user_id):
        user = User.query.get_or_404(user_id)
        db.session.delete(user)
        db.session.commit()
        return {'message': 'User deleted'}, 204

class AuthResource(Resource):
    def post(self):
        data = request.get_json()
        user = User.query.filter_by(username=data.get('username')).first()
        if user and user.check_password(data.get('password')):
            access_token = create_access_token(identity=user.id)
            return {'access_token': access_token}, 200
        return {'message': 'Invalid credentials'}, 401

7. Register the Authentication Endpoint

Add the auth route in create_app:

# Inside app/__init__.py, after other resources
from .resources import AuthResource
api.add_resource(AuthResource, '/api/auth')

8. Run Database Migrations

Initialize Alembic and generate the first migration.

flask db init
flask db migrate -m "Create users table"
flask db upgrade

9. Create the Entry Point

The run.py file boots the application.

# run.py
from app import create_app

app = create_app()

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

Testing the API

Automated tests ensure your API behaves as expected. Below is a minimal pytest example using Flask’s test client.

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

@pytest.fixture
def client():
    app = create_app('app.config.Config')
    app.config['TESTING'] = True
    app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'

    with app.test_client() as client:
        with app.app_context():
            db.create_all()
            # Create a test user
            user = User(username='tester', email='test@example.com')
            user.set_password('password123')
            db.session.add(user)
            db.session.commit()
        yield client
        with app.app_context():
            db.drop_all()

def get_token(client):
    response = client.post('/api/auth', json={'username': 'tester', 'password': 'password123'})
    return json.loads(response.data)['access_token']

def test_get_users(client):
    token = get_token(client)
    resp = client.get('/api/users', headers={'Authorization': f'Bearer {token}'})
    assert resp.status_code == 200
    data = json.loads(resp.data)
    assert isinstance(data, list)

Best Practices for Production‑Ready APIs

  • <

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *