Skip to main content

Database

Flask provides excellent database integration through Flask-SQLAlchemy, an extension that adds SQLAlchemy support to Flask applications.

Setup and Configuration​

Installation​

pip install Flask-SQLAlchemy
pip install Flask-Migrate  # For migrations

Basic Configuration​

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

app = Flask(__name__)

# Database configuration
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# Initialize extensions
db = SQLAlchemy(app)
migrate = Migrate(app, db)

Database URLs​

# SQLite
'sqlite:///app.db'

# PostgreSQL
'postgresql://username:password@localhost/mydatabase'

# MySQL
'mysql://username:password@localhost/mydatabase'

# Environment-based configuration
import os
app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('DATABASE_URL', 'sqlite:///app.db')

Models​

Basic Model​

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)
    password_hash = db.Column(db.String(120), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    def __repr__(self):
        return f'<User {self.username}>'

Column Types​

class Example(db.Model):
    # Numeric types
    integer_col = db.Column(db.Integer)
    big_integer_col = db.Column(db.BigInteger)
    small_integer_col = db.Column(db.SmallInteger)
    float_col = db.Column(db.Float)
    numeric_col = db.Column(db.Numeric(10, 2))  # precision, scale

    # String types
    string_col = db.Column(db.String(100))
    text_col = db.Column(db.Text)
    unicode_col = db.Column(db.Unicode(100))

    # Date/Time types
    date_col = db.Column(db.Date)
    time_col = db.Column(db.Time)
    datetime_col = db.Column(db.DateTime)

    # Other types
    boolean_col = db.Column(db.Boolean)
    binary_col = db.Column(db.LargeBinary)
    json_col = db.Column(db.JSON)  # PostgreSQL only

Column Constraints​

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(
        db.String(80),
        unique=True,
        nullable=False,
        index=True
    )
    email = db.Column(
        db.String(120),
        unique=True,
        nullable=False,
        index=True
    )
    age = db.Column(db.Integer, db.CheckConstraint('age >= 0'))
    status = db.Column(
        db.String(20),
        default='active',
        server_default='active'
    )

Relationships​

One-to-Many​

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), nullable=False)
    posts = db.relationship('Post', backref='author', lazy=True)

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    content = db.Column(db.Text)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

Many-to-Many​

# Association table
tags = db.Table('post_tags',
    db.Column('post_id', db.Integer, db.ForeignKey('post.id'), primary_key=True),
    db.Column('tag_id', db.Integer, db.ForeignKey('tag.id'), primary_key=True)
)

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    tags = db.relationship('Tag', secondary=tags, backref='posts')

class Tag(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), nullable=False)

One-to-One​

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), nullable=False)
    profile = db.relationship('Profile', backref='user', uselist=False)

class Profile(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    bio = db.Column(db.Text)
    avatar = db.Column(db.String(200))
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

Self-Referential Relationships​

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), nullable=False)

    # Manager relationship
    manager_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    subordinates = db.relationship(
        'User',
        backref=db.backref('manager', remote_side=[id])
    )

Database Operations​

Create Database​

# Create all tables
with app.app_context():
    db.create_all()

# Drop all tables
with app.app_context():
    db.drop_all()

CRUD Operations​

Create​

# Single record
user = User(username='john', email='john@example.com')
db.session.add(user)
db.session.commit()

# Multiple records
users = [
    User(username='alice', email='alice@example.com'),
    User(username='bob', email='bob@example.com')
]
db.session.add_all(users)
db.session.commit()

# Get the ID of newly created record
print(user.id)  # Available after commit

Read​

# Get by primary key
user = User.query.get(1)
user = User.query.get_or_404(1)  # Raises 404 if not found

# Get first
user = User.query.first()
user = User.query.first_or_404()

# Get all
users = User.query.all()

# Filter
users = User.query.filter_by(username='john').all()
user = User.query.filter_by(username='john').first()

# Complex filters
users = User.query.filter(User.username.like('%john%')).all()
users = User.query.filter(User.age > 18).all()

Update​

# Update single record
user = User.query.get(1)
user.email = 'newemail@example.com'
db.session.commit()

# Bulk update
User.query.filter_by(status='inactive').update({'status': 'active'})
db.session.commit()

Delete​

# Delete single record
user = User.query.get(1)
db.session.delete(user)
db.session.commit()

# Bulk delete
User.query.filter_by(status='inactive').delete()
db.session.commit()

Querying​

Basic Queries​

# Filter operations
User.query.filter_by(username='john')
User.query.filter(User.username == 'john')
User.query.filter(User.age > 18)
User.query.filter(User.username.like('%john%'))
User.query.filter(User.username.ilike('%JOHN%'))  # Case insensitive

# Multiple conditions
User.query.filter(User.age > 18, User.status == 'active')
User.query.filter((User.age > 18) & (User.status == 'active'))
User.query.filter((User.age < 18) | (User.age > 65))

Ordering and Limiting​

# Order by
User.query.order_by(User.username).all()
User.query.order_by(User.username.desc()).all()
User.query.order_by(User.age, User.username).all()

# Limit and offset
User.query.limit(10).all()
User.query.offset(20).limit(10).all()

# Pagination
users = User.query.paginate(
    page=1,
    per_page=10,
    error_out=False
)

Joins​

# Inner join
posts = db.session.query(Post).join(User).all()

# Left join
posts = db.session.query(Post).outerjoin(User).all()

# Join with filter
posts = db.session.query(Post).join(User).filter(User.username == 'john').all()

# Multiple joins
results = db.session.query(Post).join(User).join(Category).all()

Subqueries​

# Subquery
subquery = db.session.query(Post.user_id).filter(Post.published == True).subquery()
users_with_posts = User.query.filter(User.id.in_(subquery)).all()

# Exists
from sqlalchemy import exists
users = User.query.filter(
    exists().where(Post.user_id == User.id)
).all()

Aggregation​

from sqlalchemy import func

# Count
user_count = User.query.count()
post_count_by_user = db.session.query(
    User.username,
    func.count(Post.id)
).join(Post).group_by(User.id).all()

# Other aggregations
avg_age = db.session.query(func.avg(User.age)).scalar()
max_age = db.session.query(func.max(User.age)).scalar()
min_age = db.session.query(func.min(User.age)).scalar()
sum_ages = db.session.query(func.sum(User.age)).scalar()

Migrations​

Flask-Migrate Setup​

# Initialize migration repository
flask db init

# Create migration
flask db migrate -m "Initial migration"

# Apply migration
flask db upgrade

# Downgrade
flask db downgrade

Migration Commands​

# Generate migration automatically
flask db migrate -m "Add user table"

# Create empty migration
flask db revision -m "Custom migration"

# Show current migration
flask db current

# Show migration history
flask db history

# Upgrade to specific revision
flask db upgrade <revision>

# Stamp database without running migrations
flask db stamp head

Custom Migration​

# migrations/versions/xxx_custom_migration.py
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column('user', sa.Column('last_login', sa.DateTime()))
    op.create_index('ix_user_last_login', 'user', ['last_login'])

def downgrade():
    op.drop_index('ix_user_last_login', 'user')
    op.drop_column('user', 'last_login')

Session Management​

Session Basics​

# Manual session management
try:
    user = User(username='john')
    db.session.add(user)
    db.session.commit()
except Exception as e:
    db.session.rollback()
    raise e

Context Managers​

from contextlib import contextmanager

@contextmanager
def db_transaction():
    try:
        db.session.begin()
        yield db.session
        db.session.commit()
    except Exception:
        db.session.rollback()
        raise

# Usage
with db_transaction() as session:
    user = User(username='john')
    session.add(user)

Raw SQL​

Execute Raw SQL​

# Raw query
result = db.session.execute(
    'SELECT * FROM user WHERE age > :age',
    {'age': 18}
)
users = result.fetchall()

# Using text()
from sqlalchemy import text
result = db.session.execute(
    text('SELECT * FROM user WHERE age > :age'),
    {'age': 18}
)

SQL Expressions​

from sqlalchemy import and_, or_, not_

# Complex conditions
users = User.query.filter(
    and_(
        User.age > 18,
        or_(
            User.status == 'active',
            User.status == 'pending'
        )
    )
).all()

Performance Optimization​

Eager Loading​

# N+1 problem solution
posts = Post.query.options(db.joinedload(Post.author)).all()

# Subquery load
posts = Post.query.options(db.subqueryload(Post.comments)).all()

# Select in load
posts = Post.query.options(db.selectinload(Post.tags)).all()

Query Optimization​

# Only select specific columns
users = db.session.query(User.id, User.username).all()

# Use exists instead of count
has_posts = db.session.query(
    exists().where(Post.user_id == user_id)
).scalar()

# Bulk operations
db.session.bulk_insert_mappings(User, user_data)
db.session.bulk_update_mappings(User, user_updates)

Indexing​

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), index=True)

    # Composite index
    __table_args__ = (
        db.Index('idx_name_email', 'username', 'email'),
    )

Connection Pooling​

Pool Configuration​

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
    'poolclass': QueuePool,
    'pool_size': 10,
    'pool_recycle': 3600,
    'pool_pre_ping': True
}

Error Handling​

Common Exceptions​

from sqlalchemy.exc import IntegrityError, DataError

try:
    user = User(username='existing_user')
    db.session.add(user)
    db.session.commit()
except IntegrityError:
    db.session.rollback()
    # Handle duplicate key error
except DataError:
    db.session.rollback()
    # Handle data type error

Testing​

Test Database Setup​

import tempfile
import os

class TestConfig:
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
    # Or temporary file
    # SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(tempfile.gettempdir(), 'test.db')

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

Database Fixtures​

@pytest.fixture
def user():
    user = User(username='testuser', email='test@example.com')
    db.session.add(user)
    db.session.commit()
    return user

def test_user_creation(user):
    assert user.username == 'testuser'
    assert User.query.count() == 1