Database Integration

In this chapter, we'll persist our Task API data using SQLAlchemy. We'll cover:

• Introduction to SQLAlchemy
• Connecting to databases
• Models and migrations
• CRUD operations
• Testing with databases
• Using test databases and fixtures

Introduction to SQLAlchemy

SQLAlchemy is Python's most popular ORM (Object-Relational Mapper):

pip install flask-sqlalchemy

Setting up the database

Configuration

# app/config.py
import os


class Config:
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///app.db'
    SQLALCHEMY_TRACK_MODIFICATIONS = False


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

Initialize SQLAlchemy

# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()


def create_app(config_class=None):
    app = Flask(__name__)
    
    if config_class is None:
        from app.config import Config
        config_class = Config
    
    app.config.from_object(config_class)
    db.init_app(app)
    
    # Register routes
    from app.routes import register_routes
    register_routes(app)
    
    return app

Creating models

Write the test first

# tests/test_models.py
import pytest
from app import create_app, db
from app.models import Task
from app.config import TestConfig


@pytest.fixture
def app():
    app = create_app(TestConfig)
    with app.app_context():
        db.create_all()
        yield app
        db.drop_all()


def test_create_task(app):
    with app.app_context():
        task = Task(title='Test Task', description='A test')
        db.session.add(task)
        db.session.commit()
        
        assert task.id is not None
        assert task.title == 'Test Task'
        assert task.completed is False


def test_task_to_dict(app):
    with app.app_context():
        task = Task(title='Test Task')
        db.session.add(task)
        db.session.commit()
        
        data = task.to_dict()
        
        assert data['title'] == 'Test Task'
        assert 'id' in data
        assert 'created_at' in data

Create the model

# app/models.py
from datetime import datetime
from app import db


class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    description = db.Column(db.Text, default='')
    completed = db.Column(db.Boolean, default=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    def to_dict(self):
        return {
            'id': self.id,
            'title': self.title,
            'description': self.description,
            'completed': self.completed,
            'created_at': self.created_at.isoformat(),
            'updated_at': self.updated_at.isoformat()
        }

CRUD operations

Create

# app/services/task_service.py
from app import db
from app.models import Task


class TaskService:
    def create(self, title, description=''):
        task = Task(title=title, description=description)
        db.session.add(task)
        db.session.commit()
        return task

Read

def get_all(self):
    return Task.query.all()

def get_by_id(self, task_id):
    return Task.query.get(task_id)

def filter_by_completion(self, completed):
    return Task.query.filter_by(completed=completed).all()

Update

def update(self, task_id, **updates):
    task = self.get_by_id(task_id)
    if task:
        for key, value in updates.items():
            if hasattr(task, key):
                setattr(task, key, value)
        db.session.commit()
    return task

Delete

def delete(self, task_id):
    task = self.get_by_id(task_id)
    if task:
        db.session.delete(task)
        db.session.commit()
        return True
    return False

Testing with databases

Test fixtures

# tests/conftest.py
import pytest
from app import create_app, db
from app.config import TestConfig
from app.services.task_service import TaskService


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


@pytest.fixture
def task_service(app):
    with app.app_context():
        yield TaskService()

Testing the service

# tests/test_task_service.py
def test_create_task(app, task_service):
    with app.app_context():
        task = task_service.create('My Task', 'Description')
        
        assert task.id is not None
        assert task.title == 'My Task'


def test_get_all_tasks(app, task_service):
    with app.app_context():
        task_service.create('Task 1')
        task_service.create('Task 2')
        
        tasks = task_service.get_all()
        
        assert len(tasks) == 2


def test_update_task(app, task_service):
    with app.app_context():
        task = task_service.create('Original')
        
        updated = task_service.update(task.id, title='Updated', completed=True)
        
        assert updated.title == 'Updated'
        assert updated.completed is True


def test_delete_task(app, task_service):
    with app.app_context():
        task = task_service.create('To Delete')
        
        result = task_service.delete(task.id)
        
        assert result is True
        assert task_service.get_by_id(task.id) is None

Updating the API

# app/routes.py
from flask import request, jsonify
from app import db
from app.models import Task
from app.services.task_service import TaskService


def register_routes(app):
    task_service = TaskService()
    
    @app.route('/api/tasks')
    def get_tasks():
        completed = request.args.get('completed')
        
        if completed is not None:
            is_completed = completed.lower() == 'true'
            tasks = task_service.filter_by_completion(is_completed)
        else:
            tasks = task_service.get_all()
        
        return jsonify([t.to_dict() for t in tasks])
    
    @app.route('/api/tasks', methods=['POST'])
    def create_task():
        data = request.json
        
        if not data or not data.get('title', '').strip():
            return jsonify({'error': 'Title is required'}), 400
        
        task = task_service.create(
            data['title'].strip(),
            data.get('description', '')
        )
        return jsonify(task.to_dict()), 201
    
    @app.route('/api/tasks/<int:task_id>')
    def get_task(task_id):
        task = task_service.get_by_id(task_id)
        if task:
            return jsonify(task.to_dict())
        return jsonify({'error': 'Task not found'}), 404
    
    @app.route('/api/tasks/<int:task_id>', methods=['PUT'])
    def update_task(task_id):
        data = request.json
        task = task_service.update(task_id, **data)
        if task:
            return jsonify(task.to_dict())
        return jsonify({'error': 'Task not found'}), 404
    
    @app.route('/api/tasks/<int:task_id>', methods=['DELETE'])
    def delete_task(task_id):
        if task_service.delete(task_id):
            return '', 204
        return jsonify({'error': 'Task not found'}), 404

Relationships

Adding notes to tasks

# app/models.py
class Note(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.Text, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    task_id = db.Column(db.Integer, db.ForeignKey('task.id'), nullable=False)
    
    task = db.relationship('Task', backref=db.backref('notes', lazy=True))
    
    def to_dict(self):
        return {
            'id': self.id,
            'content': self.content,
            'created_at': self.created_at.isoformat(),
            'task_id': self.task_id
        }


class Task(db.Model):
    # ... existing fields ...
    
    def to_dict(self, include_notes=False):
        data = {
            'id': self.id,
            'title': self.title,
            'description': self.description,
            'completed': self.completed,
            'created_at': self.created_at.isoformat(),
            'updated_at': self.updated_at.isoformat()
        }
        if include_notes:
            data['notes'] = [n.to_dict() for n in self.notes]
        return data

Testing relationships

def test_task_with_notes(app):
    with app.app_context():
        task = Task(title='Task with Notes')
        db.session.add(task)
        db.session.commit()
        
        note = Note(content='A note', task_id=task.id)
        db.session.add(note)
        db.session.commit()
        
        # Reload task
        task = Task.query.get(task.id)
        assert len(task.notes) == 1
        assert task.notes[0].content == 'A note'

Migrations with Flask-Migrate

For production, use migrations:

pip install flask-migrate

# app/__init__.py
from flask_migrate import Migrate

migrate = Migrate()


def create_app(config_class=None):
    app = Flask(__name__)
    # ... config ...
    
    db.init_app(app)
    migrate.init_app(app, db)
    
    # ...

Commands:

flask db init          # Initialize migrations
flask db migrate -m "Add tasks table"  # Create migration
flask db upgrade       # Apply migrations
flask db downgrade     # Rollback

Transaction handling

Rollback on error

def create_task_with_notes(self, title, notes_content):
    try:
        task = Task(title=title)
        db.session.add(task)
        db.session.flush()  # Get task.id without committing
        
        for content in notes_content:
            note = Note(content=content, task_id=task.id)
            db.session.add(note)
        
        db.session.commit()
        return task
    except Exception as e:
        db.session.rollback()
        raise

Testing transactions

def test_rollback_on_error(app, task_service):
    with app.app_context():
        initial_count = len(task_service.get_all())
        
        try:
            # This should fail and rollback
            task_service.create_task_with_notes('Task', [None])  # Invalid note
        except Exception:
            pass
        
        # No task should be created
        assert len(task_service.get_all()) == initial_count

Query optimization

Eager loading

# Avoid N+1 queries
def get_tasks_with_notes(self):
    return Task.query.options(db.joinedload(Task.notes)).all()

Pagination

def get_paginated(self, page=1, per_page=20):
    return Task.query.paginate(
        page=page,
        per_page=per_page,
        error_out=False
    )

Usage:

@app.route('/api/tasks')
def get_tasks():
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    
    pagination = task_service.get_paginated(page, per_page)
    
    return jsonify({
        'items': [t.to_dict() for t in pagination.items],
        'total': pagination.total,
        'page': pagination.page,
        'pages': pagination.pages,
        'has_next': pagination.has_next,
        'has_prev': pagination.has_prev
    })

Using a real database for tests

For integration tests with a real database:

# tests/conftest.py
import os
import pytest


@pytest.fixture(scope='session')
def real_db_app():
    """Use for integration tests that need a real database."""
    os.environ['DATABASE_URL'] = 'postgresql://user:pass@localhost/test_db'
    
    app = create_app()
    with app.app_context():
        db.create_all()
        yield app
        db.drop_all()

Wrapping up

We've covered:

• SQLAlchemy setup - Configuring Flask-SQLAlchemy
• Models - Defining database tables as Python classes
• CRUD operations - Create, read, update, delete with the ORM
• Relationships - Foreign keys and backref
• Testing - Using in-memory SQLite for fast tests
• Migrations - Version control for database schema
• Transactions - Handling commits and rollbacks
• Query optimization - Eager loading and pagination

Key takeaways

1. Use an in-memory SQLite database for unit tests
2. Always commit or rollback transactions
3. Use to_dict() methods for JSON serialization
4. Handle relationships carefully to avoid N+1 queries
5. Use migrations for production database changes

In the next chapter, we'll build a command-line interface for our API.