SQLAlchemy ORM (Flask-SQLAlchemy)：告别原生 SQL，用 Python 对象管理 Flask 数据库

📂 所属阶段：第二阶段 — 交互与数据（核心篇）
🔗 相关章节：数据库关系设计 · environment-setup

1. ORM 是什么？一张对比图秒懂

如果你曾经写过这样的代码，拼接字符串拼到崩溃：

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

那 ORM（对象关系映射）会让你轻松很多。你只需要操作 Python 类和对象，剩下的交给它：

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

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

ORM 最大的好处：数据库无关性。将来想把 SQLite 换成 PostgreSQL，几乎不用改业务代码，换个配置连接串就行。

1.1 快速安装

pip install flask-sqlalchemy

如果使用 PostgreSQL 或 MySQL，记得装上对应的数据库驱动：

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install pymysql

2. Flask-SQLAlchemy 基础配置

2.1 最小可用配置

为了避免循环导入，建议把扩展实例单独抽到一个文件（例如 app/extensions.py）：

# app/extensions.py
from flask_sqlalchemy import SQLAlchemy

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

然后在应用工厂中绑定配置和 db 对象：

# 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 支持的数据库与连接池配置

不同的数据库只需要修改 SQLALCHEMY_DATABASE_URI 即可：

数据库类型配置 URI 示例
SQLite（开发用）sqlite:///myblog.db（本地文件）或 sqlite:///:memory:（内存临时库）
PostgreSQLpostgresql://user:pass@localhost:5432/myblog
MySQLmysql+pymysql://user:pass@localhost:3306/myblog?charset=utf8mb4

生产环境强烈建议配置连接池，避免频繁创建/销毁数据库连接：

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

3. 模型定义：把数据库表变成 Python 类

模型其实就是一种映射： Python 类 → 数据库表，类属性 → 表字段，类的实例 → 表中的一行。

3.1 用户模型（搭配 Flask-Login 的 UserMixin）

# 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 文章模型（带外键与级联删除）

# 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 常用字段类型速查

SQLAlchemy 类型Python 类型说明
db.Integerint整数，主键最常用
db.BigIntegerint更大范围的整数
db.String(n)str变长字符串，必须指定最大长度
db.Textstr长文本（文章内容、评论等）
db.Booleanbool布尔型（是/否）
db.DateTimedatetime日期时间
db.Floatfloat浮点数（不要用来存金额）
db.Numeric(10,2)Decimal定点小数，适合金额、分数
db.JSONdict/listJSON 数据，灵活存储配置或标签

4. 上手 CRUD：增删改查快速入门

💡 所有操作都必须在 Flask 应用上下文中执行。在视图函数里自动有，如果要在终端测试，请加上 with 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）

最常用的几种查询姿势：

# 按主键查询 —— 最快
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()

需要写原生 SQL 时的安全做法（不推荐频繁使用）：

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. 实用查询技巧

5.1 模糊搜索

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

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

5.2 IN 查询

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

5.3 AND / OR 组合

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 统计与聚合

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. 分页查询：告别手动 OFFSET / LIMIT

在博客或列表页中，分页是刚需。Flask-SQLAlchemy 提供了非常方便的分页器。

6.1 视图函数中分页

# 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 Jinja2 模板中的分页 UI

<!-- 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 %}

上面的例子只展示了基本页码，实际项目中可以进一步优化（比如添加省略号、只显示前后几页）。

7. 小结与最佳实践

7.1 常用操作速记

把最常用的会话、查询、更新、删除方法记在这里，随时查阅：

# 会话管理
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 最佳实践建议

  1. 永远不要存储明文密码：使用 werkzeug.security 的 generate_password_hash 和 check_password_hash 处理密码。
  2. 杜绝 SQL 注入：即使写原生 SQL，也要用 text() + 字典传参，绝不拼接字符串。
  3. 时间统一用 UTC：后端存储一律使用 datetime.utcnow，展示时再根据用户时区转换。
  4. 模型拆分保存：将每个模型放在单独的文件（如 app/models/user.py），保持代码清爽。
  5. 善用数据库索引：在经常查询的字段上添加 index=True，unique=True 会自动创建唯一索引，有效提升查询性能。

🔗 扩展阅读