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

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

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

原生 SQL 开发者的日常，可能是盯着一堆字符串拼语法：

原始 SQL：
INSERT INTO users (username, email, password_hash)
VALUES ("alice", "alice@example.com", "hashed_pwd123");

用 ORM（对象关系映射） 后，直接操作 Python 类和对象就行：

# ORM 操作
user = User(username="alice", email="alice@example.com", password_hash="hashed_pwd123")
db.session.add(user)
db.session.commit()

# 底层自动生成**安全的、符合数据库语法的 SQL**

它最大的好处是数据库无关性——换数据库（比如从 SQLite 切到 PostgreSQL）时，几乎不用改业务代码，只需要调整配置 URI 即可。

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 对象

然后在应用工厂里配置并绑定：

# 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)  # 绑定应用
    return app

2.2 支持的数据库与进阶配置

数据库类型配置 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,   # 1小时后强制回收旧连接（避免数据库断开）
    "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"  # 自定义表名（默认类名转小写下划线，如 User→user）

    # 字段定义
    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 别名（如 my-first-post）
    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大整数（如存储大 ID）
db.String(n)str可变长度字符串（加最大长度限制）
db.Textstr长文本（文章内容、评论等）
db.Booleanbool布尔值（是/否、激活/禁用）
db.DateTimedatetime日期时间
db.Floatfloat浮点数（不推荐存金额）
db.Numeric(10,2)Decimal精确小数（金额、分数等）
db.JSONdict/listJSON 数据（灵活存储配置、标签列表）

4. CRUD 操作速查（核心！）

CRUD 是数据库最常用的增 Create、查 Read、改 Update、删 Delete 操作。

💡 前提：必须在 Flask 应用上下文中执行——Flask 请求处理时自动有，手动测试时加 with app.app_context():。

4.1 创建（Create）

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

# 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()  # 查第一条符合条件的，查不到返回 None

# 高级查询（支持条件组合、排序、分页）
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()  # 查1月1日后的前10个活跃用户，按注册时间倒序

原生 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）

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

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

4.4 删除（Delete）

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

# 2. 批量删除（注意：级联删除需要在模型里设置 cascade）
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 查询

# 查 id 为 1、3、5 的用户
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_users = db.session.query(func.count(User.id)).scalar()
# 或者更简单的写法
total_users = User.query.count()

# 统计每个分类下的已发布文章数
from app.models.post import Category

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. 分页功能（Flask-SQLAlchemy 内置超好用）

分页是博客、列表页的标配，不用自己写 LIMIT/OFFSET。

6.1 视图函数中的分页

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

@bp.route("/users")
def list_users():
    # 获取页码（默认第1页）
    page = request.args.get("page", 1, type=int)
    per_page = 10  # 每页显示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)        # 跳过
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 也别拼字符串，用 text() + 字典传参，防止 SQL 注入。
  3. 用 UTC 时间：datetime.utcnow 存储，前端再转成用户本地时间。
  4. 抽离模型：每个模型单独一个文件（放在 app/models/ 目录），避免代码臃肿。
  5. 加索引：unique=True 自动加唯一索引，常用查询字段（如 username、email）手动加 index=True，提升查询速度。

🔗 扩展阅读