SQLAlchemy ORM (Flask-SQLAlchemy): Say goodbye to native SQL and use Python objects to manage Flask databases

📂 Stage: Stage 2 - Interaction and Data (Core) 🔗 Related chapters: 数据库关系设计 · environment-setup

1. What is ORM? A comparison picture to understand in seconds

If you've ever written code like this, concatenating strings until it crashes:

-- 原生 SQL：字符串拼凑，容易出错，且存在 SQL 注入风险
INSERT INTO users (username, email, password_hash)
VALUES ("alice", "alice@example.com", "hashed_pwd123");

That ORM (Object Relational Mapping) will make it a lot easier for you. You just need to manipulate Python classes and objects and let it do the rest:

# ORM：用 Python 对象操作数据库
user = User(username="alice", email="alice@example.com", password_hash="hashed_pwd123")
db.session.add(user)
db.session.commit()

# 框架会自动生成**安全、符合数据库语法**的 SQL 语句

ORM’s biggest benefit: database independence. If you want to replace SQLite with PostgreSQL in the future, you almost don’t need to change the business code, just change the configuration connection string.

1.1 Quick installation

pip install flask-sqlalchemy

If you use PostgreSQL or MySQL, remember to install the corresponding database driver:

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install pymysql

2. Flask-SQLAlchemy basic configuration

2.1 Minimum available configuration

In order to avoid circular import, it is recommended to extract the extension instance to a separate file (for exampleapp/extensions.py）：

# app/extensions.py
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()  # 全局 db 对象，后续在应用工厂中初始化

Then bind the configuration anddbObject:

# app/__init__.py
from flask import Flask
from app.extensions import db

def create_app():
    app = Flask(__name__)

    # 核心配置
    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///myblog.db"   # 数据库文件保存在项目根目录
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False           # 关闭对象追踪，提高性能

    db.init_app(app)  # 把 app 和 db 关联起来
    return app

2.2 Supported database and connection pool configurations

Different databases only need to be modifiedSQLALCHEMY_DATABASE_URIThat’s it:

Database TypeConfiguration URI Example
SQLite (for development)sqlite:///myblog.db(local file) orsqlite:///:memory:(memory temporary library)
PostgreSQLpostgresql://user:pass@localhost:5432/myblog
MySQLmysql+pymysql://user:pass@localhost:3306/myblog?charset=utf8mb4

In production environments, it is strongly recommended to configure a connection pool to avoid frequent creation/destruction of database connections:

app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {
    "pool_size": 10,        # 连接池容纳的常规连接数
    "pool_recycle": 3600,   # 3600 秒后强制回收连接（防止数据库主动断开）
    "pool_timeout": 30,     # 等待连接的超时时间（秒）
    "max_overflow": 20,     # 连接池满时可以临时创建的额外连接数
}

3. Model definition: turn database tables into Python classes

The model is actually a mapping: Python class → database table, class attribute → table field, instance of the class → a row in the table.

3.1 User model (UserMixin with Flask-Login)

# app/models/user.py
from datetime import datetime
from app.extensions import db
from flask_login import UserMixin

class User(UserMixin, db.Model):
    """用户信息表"""
    __tablename__ = "users"  # 自定义表名，不写的话默认为小写类名

    # 字段定义
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), unique=True, nullable=False, index=True)  # 唯一、非空、加索引
    email = db.Column(db.String(120), unique=True, nullable=False, index=True)
    password_hash = db.Column(db.String(256), nullable=False)  # 永远不要存明文密码！
    bio = db.Column(db.String(500), default="")
    avatar = db.Column(db.String(200), default="default_avatar.jpg")
    is_active = db.Column(db.Boolean, default=True)
    is_admin = db.Column(db.Boolean, default=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)                    # utc 避免时区困扰
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # 一对多关系：一个用户拥有多篇文章
    posts = db.relationship("Post", back_populates="author", lazy="dynamic")

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

3.2 Article model (with foreign keys and cascade deletion)

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

class Post(db.Model):
    """文章信息表"""
    __tablename__ = "posts"

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    slug = db.Column(db.String(200), unique=True, index=True)   # 用于友好 URL
    content = db.Column(db.Text, nullable=False)
    summary = db.Column(db.String(500))
    cover_image = db.Column(db.String(200))
    views = db.Column(db.Integer, default=0)
    is_published = 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)

    # 外键约束
    author_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
    category_id = db.Column(db.Integer, db.ForeignKey("categories.id", ondelete="SET NULL"))

    # 反向关系
    author = db.relationship("User", back_populates="posts")
    category = db.relationship("Category", back_populates="posts")
    comments = db.relationship("Comment", back_populates="post", lazy="dynamic",
                               cascade="all, delete-orphan")  # 删除文章时，级联删除所有评论

    def __repr__(self):
        return f"<Post {self.title}>"

3.3 Quick check of common field types

SQLAlchemy typesPython typesDescription
db.IntegerintInteger, the most commonly used primary key
db.BigIntegerintA larger range of integers
db.String(n)strVariable-length string, the maximum length must be specified
db.Textstrlong text (article content, comments, etc.)
db.BooleanboolBoolean type (yes/no)
db.DateTimedatetimedatetime
db.FloatfloatFloating point number (do not use to save amount)
db.Numeric(10,2)DecimalFixed point decimal, suitable for amounts and fractions
db.JSONdict/listJSON data, flexible storage configuration or tags

4. Getting Started with CRUD: Quick Start with Add, Delete, Modify and Check

💡 All operations must be performed within the Flask application context. It is automatically included in the view function. If you want to test in the terminal, please addwith app.app_context():。

4.1 Create

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

# 1. 创建单条记录
user = User(username="bob", email="bob@example.com", password_hash="hashed_bob_pwd")
db.session.add(user)
db.session.commit()   # 真正写入数据库（出错的话可以回滚：db.session.rollback()）

# 2. 批量创建（推荐 add_all，比循环 add 更高效）
users = [
    User(username="charlie", email="charlie@ex.com", password_hash="..."),
    User(username="david", email="david@ex.com", password_hash="..."),
]
db.session.add_all(users)
db.session.commit()

# 3. 大量数据的纯迁移场景，可以用超高性能批量插入（不会触发 ORM 事件）
db.session.bulk_save_objects(users)
db.session.commit()

4.2 Read

The most commonly used query postures:

# 按主键查询 —— 最快
user = db.session.get(User, 1)          # 查 id=1，找不到返回 None

# 单条件精确查询
user = User.query.filter_by(username="bob").first()   # 返回第一条符合条件的记录

# 多条件组合、排序、限制
from datetime import datetime
active_users = User.query.filter(
    User.is_active == True,
    User.created_at > datetime(2025, 1, 1)
).order_by(User.created_at.desc()).limit(10).all()

Safety practices when writing native SQL (not recommended for frequent use):

from sqlalchemy import text

result = db.session.execute(
    text("SELECT * FROM users WHERE created_at > :date"),
    {"date": datetime(2025, 1, 1)}   # 参数化查询，防止 SQL 注入
)
users = result.mappings().all()      # 转成字典列表

4.3 Update

# 单条更新
user = db.session.get(User, 1)
if user:
    user.bio = "Bob 的新简介"
    user.is_active = False
    db.session.commit()

# 批量更新（性能更好，不会触发每条记录的 ORM 事件）
User.query.filter_by(is_active=False).update({"is_admin": False})
db.session.commit()

4.4 Delete

# 单条删除
user = db.session.get(User, 1)
if user:
    db.session.delete(user)
    db.session.commit()

# 批量删除（注意级联设置，否则可能留下孤儿数据）
Post.query.filter_by(is_published=False).delete()
db.session.commit()

5. Practical query skills

# 区分大小写
User.query.filter(User.username.like("%ob%"))   # 包含 "ob" 的用户名

# 不区分大小写（PostgreSQL 推荐 ilike）
User.query.filter(User.username.ilike("%OB%"))

5.2 IN query

User.query.filter(User.id.in_([1, 3, 5])).all()

5.3 AND / OR combination

from sqlalchemy import and_, or_

# AND：活跃且不是管理员
User.query.filter(and_(User.is_active == True, User.is_admin == False)).all()

# OR：用户名为 bob 或邮箱是 bob@ex.com
User.query.filter(or_(User.username == "bob", User.email == "bob@ex.com")).all()

5.4 Statistics and Aggregation

from sqlalchemy import func

# 用户总数
total = User.query.count()

# 每个分类下已发布文章的数量
category_post_counts = db.session.query(
    Category.name, func.count(Post.id)
).join(Post, Post.category_id == Category.id, isouter=True).filter(
    Post.is_published == True
).group_by(Category.id).all()

6. Paging query: Say goodbye to manual OFFSET / LIMIT

In a blog or list page, pagination is a necessity. Flask-SQLAlchemy provides a very convenient pager.

6.1 Pagination in view functions

# app/routes/user.py
from flask import request, render_template
from app.models.user import User

@bp.route("/users")
def list_users():
    page = request.args.get("page", 1, type=int)
    per_page = 10

    pagination = User.query.order_by(User.created_at.desc()).paginate(
        page=page,
        per_page=per_page,
        error_out=False   # 页码超出范围时返回空数据，否则会 404
    )

    return render_template("users/list.html", pagination=pagination)

6.2 Pagination UI in Jinja2 template

<!-- templates/users/list.html -->
<ul class="user-list">
    {% for user in pagination.items %}
        <li>{{ user.username }} · {{ user.email }}</li>
    {% else %}
        <li>暂无用户</li>
    {% endfor %}
</ul>

<!-- 分页导航 -->
{% if pagination.pages > 1 %}
    <div class="pagination flex gap-2 mt-4">
        {% if pagination.has_prev %}
            <a href="{{ url_for('user.list_users', page=pagination.prev_num) }}"
               class="px-3 py-1 border rounded">上一页</a>
        {% endif %}

        {% for p in range(1, pagination.pages + 1) %}
            {% if p == pagination.page %}
                <span class="px-3 py-1 bg-blue-500 text-white rounded">{{ p }}</span>
            {% else %}
                <a href="{{ url_for('user.list_users', page=p) }}"
                   class="px-3 py-1 border rounded">{{ p }}</a>
            {% endif %}
        {% endfor %}

        {% if pagination.has_next %}
            <a href="{{ url_for('user.list_users', page=pagination.next_num) }}"
               class="px-3 py-1 border rounded">下一页</a>
        {% endif %}
    </div>
{% endif %}

The above example only shows basic page numbers. In actual projects, it can be further optimized (such as adding ellipses and only displaying the previous and next pages).

7. Summary and best practices

7.1 Common operation shorthand

Write down the most commonly used conversation, query, update, and delete methods here for reference at any time:

# 会话管理
db.session.add(obj)        # 加入单个对象
db.session.add_all(objs)   # 加入批量对象
db.session.commit()        # 提交事务
db.session.rollback()      # 回滚事务

# 查询
db.session.get(User, id)   # 按主键查
User.query.filter_by(x=y)  # 简单等值过滤
User.query.filter(Xxx)     # 复杂条件过滤
User.query.order_by(...)   # 排序
User.query.limit(n)        # 限制返回条数
User.query.offset(n)       # 跳过前 n 条
User.query.paginate(...)   # 分页
User.query.count()         # 计数
User.query.all()           # 返回所有结果
User.query.first()         # 返回第一条结果

# 更新 & 删除
User.query.filter(...).update(...)   # 批量更新
db.session.delete(obj)               # 删除单个对象
User.query.filter(...).delete()      # 批量删除

7.2 Best practice recommendations

  1. Never store clear text passwords: Usewerkzeug.securityofgenerate_password_hashandcheck_password_hashHandle passwords.
  2. Eliminate SQL injection: Even if you write native SQL, you must usetext()+ Dictionary passing parameters, never concatenate strings.
  3. Uniformly use UTC for time: all back-end storage is useddatetime.utcnow, and then converted according to the user's time zone when displayed.
  4. Model Split Saving: Put each model in a separate file (such asapp/models/user.py) to keep the code clean.
  5. Make good use of database indexes: Add fields to frequently queried fieldsindex=True，unique=TrueA unique index will be automatically created to effectively improve query performance.

🔗 Extended reading