CRUD Operations

Difficulty: Intermediate

CRUD stands for Create, Read, Update, and Delete, the four basic operations for persistent data. Flask-SQLAlchemy provides an intuitive API for all CRUD operations through the db.session and the Model.query interface.

Creating records involves instantiating a model, adding it to the session, and committing. The session is a staging area for changes. You can add multiple objects before committing them all in a single transaction. If anything goes wrong, you call db.session.rollback() to undo uncommitted changes.

Reading records uses the query interface. Model.query returns a BaseQuery object with methods for filtering, ordering, and limiting results. Common methods include .all() (returns all results as a list), .first() (returns the first result or None), .get(id) (returns by primary key or None), .filter() (SQL-like filtering), .filter_by() (keyword argument filtering), .order_by(), .limit(), and .offset().

The filter() method uses SQLAlchemy column expressions like User.name == 'Alice' or User.age > 18. The filter_by() method uses simpler keyword arguments like filter_by(name='Alice'). filter() is more powerful, supporting complex expressions with and_(), or_(), like(), in_(), and between(). filter_by() is more concise for simple equality checks.

Updating records is straightforward: query the object, modify its attributes, and commit. SQLAlchemy tracks changes to loaded objects and generates UPDATE statements automatically. For bulk updates without loading objects, use Model.query.filter(...).update({...}).

Deleting records uses db.session.delete(object) followed by commit. For bulk deletes, use Model.query.filter(...).delete(). Be careful with cascading deletes: if a User has Posts with cascade='all, delete-orphan', deleting the user also deletes all their posts.

Pagination is essential for large datasets. Flask-SQLAlchemy provides a paginate() method that returns a Pagination object with items, pages, total, has_next, has_prev, next_num, and prev_num attributes. This makes implementing paginated API responses and web pages straightforward.

Always use try/except around database operations and call db.session.rollback() in the except block. Database operations can fail due to unique constraint violations, connection issues, or data integrity problems. Rolling back ensures the session is in a clean state.

Code examples

Create Operations

from app.extensions import db
from app.models import User, Post

# Create a single record
user = User(username='alice', email='alice@example.com')
db.session.add(user)
db.session.commit()

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

# Create with relationship
post = Post(title='Hello World', body='My first post', author=user)
db.session.add(post)
db.session.commit()

# Safe creation with error handling
try:
    user = User(username='alice', email='alice@example.com')
    db.session.add(user)
    db.session.commit()
except Exception:
    db.session.rollback()
    raise

Add objects to the session with add() or add_all(), then commit(). Objects with relationships can be created together. Always handle errors with rollback() to keep the session clean.

Read Operations (Queries)

from app.models import User

# Get all users
users = User.query.all()  # [<User alice>, <User bob>]

# Get by primary key
user = User.query.get(1)  # <User alice> or None

# Get first match
user = User.query.filter_by(username='alice').first()  # <User alice> or None

# Get one or 404
user = User.query.get_or_404(1)  # <User alice> or abort(404)

# Filter with expressions
active_users = User.query.filter(User.is_active == True).all()
recent = User.query.filter(User.created_at > some_date).all()
search = User.query.filter(User.username.like('%ali%')).all()

# Order and limit
top_users = User.query.order_by(User.created_at.desc()).limit(10).all()

# Count
total = User.query.count()
active_count = User.query.filter_by(is_active=True).count()

# Chaining
results = User.query\
    .filter(User.is_active == True)\
    .filter(User.role == 'admin')\
    .order_by(User.username)\
    .limit(20)\
    .all()

The query API is chainable. filter_by() uses keyword args for simple equality. filter() uses expressions for complex conditions. Methods like all(), first(), count() execute the query.

Update and Delete Operations

from app.extensions import db
from app.models import User

# Update a single record
user = User.query.get(1)
user.email = 'newemail@example.com'
user.is_active = False
db.session.commit()  # SQLAlchemy detects changes automatically

# Bulk update (efficient, no loading)
User.query.filter(User.is_active == False)\
    .update({'role': 'inactive'})
db.session.commit()

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

# Bulk delete
User.query.filter(User.last_login < cutoff_date).delete()
db.session.commit()

# Delete with cascade (deletes user AND their posts)
user = User.query.get(1)
db.session.delete(user)  # Posts auto-deleted if cascade set
db.session.commit()

Update by modifying attributes and committing. Bulk update/delete uses filter().update()/delete() without loading objects. SQLAlchemy tracks attribute changes automatically.

Pagination

from app.models import Post

# Paginate results (page 1, 20 per page)
pagination = Post.query.order_by(Post.created_at.desc())\
    .paginate(page=1, per_page=20, error_out=False)

# Pagination object attributes
posts = pagination.items      # List of posts for this page
total = pagination.total       # Total number of posts
pages = pagination.pages       # Total number of pages
has_next = pagination.has_next
has_prev = pagination.has_prev
next_num = pagination.next_num  # Next page number

# In a view function
@app.route('/api/posts')
def list_posts():
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)

    pagination = Post.query.order_by(Post.created_at.desc())\
        .paginate(page=page, per_page=min(per_page, 100))

    return {
        'posts': [p.to_dict() for p in pagination.items],
        'total': pagination.total,
        'pages': pagination.pages,
        'page': page,
        'has_next': pagination.has_next
    }

paginate() returns a Pagination object with items and metadata. error_out=False returns an empty page instead of 404 for out-of-range pages. Always cap per_page to prevent abuse.

Key points

Concepts covered

Create, Read, Update, Delete, db.session, Query API, Filtering