Flask数据库操作与SQLAlchemy - Web开发-学习笔记-上海佼艾

一、Flask-SQLAlchemy 概述

1.1 ORM 概念

ORM（Object-Relational Mapping，对象关系映射）是一种将关系型数据库中的表结构映射为面向对象编程中类和对象的技术。通过ORM，开发者不必直接编写SQL语句，而是通过操作Python对象来间接操作数据库。这样做的好处是显而易见的：首先，代码可读性大大提升，业务逻辑与数据访问逻辑解耦；其次，应用程序与底层数据库解耦，更换数据库时只需修改配置而无需重写代码；最后，ORM框架自动处理SQL注入防护，提升了应用安全性。

1.2 SQLAlchemy 简介

SQLAlchemy是Python生态中最流行的ORM框架，以其灵活性和强大的功能而著称。它提供了两套核心API：一是Core层，提供对数据库连接的底层抽象，允许开发者使用SQL表达式语言；二是ORM层，在Core之上构建，提供完整的对象关系映射能力。SQLAlchemy支持几乎所有主流关系型数据库——包括SQLite、MySQL、PostgreSQL、Oracle、Microsoft SQL Server等，在生产环境中有着广泛的应用。

1.3 Flask-SQLAlchemy 集成插件

Flask-SQLAlchemy是Flask框架的官方扩展插件，它将SQLAlchemy与Flask应用深度集成，简化了配置管理和会话生命周期管理。该插件自动将Flask应用上下文与数据库会话绑定，确保每个请求拥有独立的数据库会话，请求结束后自动关闭连接。

1.4 安装与配置

安装Flask-SQLAlchemy非常简单，通过pip即可完成：

pip install flask-sqlalchemy
# 根据所选数据库安装对应驱动
pip install pymysql      # MySQL驱动
pip install psycopg2     # PostgreSQL驱动

在Flask应用中进行基本配置：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 设置数据库URI
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///data.db'
# 关闭SQLAlchemy自身的信号系统
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 设置成True可在终端输出SQL语句，调试时非常有用
app.config['SQLALCHEMY_ECHO'] = True

# 创建数据库实例
db = SQLAlchemy(app)

提示：在生产环境中应当始终将 SQLALCHEMY_ECHO 设置为 False，避免敏感信息泄露。

二、数据库连接配置

2.1 SQLite 配置

SQLite是开发阶段的首选数据库，它无需安装额外的数据库服务器，所有数据存储在单一文件中，非常适合学习和原型开发。其URI格式简单：

# 相对路径（相对于Flask实例文件夹）
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///data.db'

# 绝对路径
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////absolute/path/to/db/data.db'

# 内存数据库（不持久化，测试用）
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'

2.2 MySQL 配置

MySQL是生产环境中最常用的数据库之一，配合PyMySQL驱动使用，需要注意字符集的配置：

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://username:password@localhost:3306/db_name?charset=utf8mb4'

各参数含义如下：用户名（username）、密码（password）、主机地址（localhost）、端口号（3306是MySQL默认端口）、数据库名（db_name），charset=utf8mb4则确保支持完整的Unicode字符（包括emoji）。

2.3 PostgreSQL 配置

PostgreSQL以强大的事务支持和高级数据类型著称，在企业级应用中广泛使用：

app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://username:password@localhost:5432/db_name'

2.4 多数据库绑定（SQLALCHEMY_BINDS）

Flask-SQLAlchemy支持同时连接多个数据库，通过 SQLALCHEMY_BINDS 配置实现。这在需要读写分离、分库分表或集成第三方应用数据时非常有用：

app.config['SQLALCHEMY_BINDS'] = {
    'default': 'sqlite:///data.db',
    'users':   'mysql+pymysql://user:pass@localhost/users_db',
    'logs':    'postgresql://user:pass@localhost/logs_db'
}

在模型定义中，通过 __bind_key__ 属性指定模型属于哪个数据库：

class User(db.Model):
    __bind_key__ = 'users'
    id = db.Column(db.Integer, primary_key=True)

class Log(db.Model):
    __bind_key__ = 'logs'
    id = db.Column(db.Integer, primary_key=True)

2.5 连接池设置

生产环境中合理配置连接池可以大幅提升数据库访问性能：

app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
    'pool_size': 10,           # 连接池大小
    'pool_recycle': 3600,      # 连接回收时间（秒）
    'pool_pre_ping': True,     # 使用前检查连接是否有效
    'max_overflow': 20         # 连接池最大溢出数量
}

注意：SQLite 不支持连接池的多线程并发，因此 pool_size 和 pool_pre_ping 对 SQLite 无效。在开发环境中使用 SQLite 时无需配置连接池参数。

三、模型定义

3.1 db.Model 基类与表名

所有模型类都必须继承 db.Model，并可选择定义 __tablename__ 指定数据库表名。如果不指定，SQLAlchemy将自动将类名小写并转换为复数形式作为表名：

class User(db.Model):
    __tablename__ = 'users'  # 显式指定表名

    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)
    is_active = db.Column(db.Boolean, default=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

3.2 列类型详解

SQLAlchemy提供了丰富的字段类型映射，下表列出了最常用的列类型及其对应的Python类型和数据库类型：

SQLAlchemy类型	Python类型	数据库类型	说明
db.Integer	int	INTEGER	整数字段
db.String(n)	str	VARCHAR(n)	可变长字符串，n为最大长度
db.Text	str	TEXT	长文本，无长度限制
db.Float	float	FLOAT	浮点数
db.Boolean	bool	BOOLEAN / INTEGER	布尔值，SQLite中存储为0/1
db.DateTime	datetime	DATETIME / TIMESTAMP	日期时间
db.Date	date	DATE	日期（不含时间）
db.JSON	dict / list	JSON / TEXT	JSON数据，需数据库支持
db.Enum	enum.Enum	ENUM / VARCHAR	枚举类型
db.LargeBinary	bytes	BLOB / BYTEA	二进制数据

3.3 列约束详解

列的约束条件定义了数据的完整性和行为规则：

class Product(db.Model):
    __tablename__ = 'products'

    id = db.Column(db.Integer, primary_key=True)                     # 主键
    sku = db.Column(db.String(50), unique=True, nullable=False)      # 唯一且非空
    name = db.Column(db.String(200), nullable=False)                 # 非空
    price = db.Column(db.Float, default=0.0)                         # 默认值0.0
    description = db.Column(db.Text, nullable=True)                  # 可为空
    category_id = db.Column(db.Integer, db.ForeignKey('categories.id'))  # 外键
    is_published = db.Column(db.Boolean, default=False, index=True)  # 索引
    created_at = db.Column(db.DateTime, default=datetime.utcnow)     # 默认当前时间

3.4 创建与删除表

# 创建所有表（基于定义的所有模型）
with app.app_context():
    db.create_all()

# 删除所有表（谨慎使用！会删除所有数据）
with app.app_context():
    db.drop_all()

# 创建单个表
with app.app_context():
    User.__table__.create(db.engine)

推荐做法：在开发环境中，使用 db.create_all() 快速创建表结构；在生产环境中，应使用数据库迁移工具（Flask-Migrate）来管理表结构的版本变更，避免数据丢失。

四、CRUD 操作

4.1 创建（Create）

创建新记录的基本流程：实例化模型对象 -> 添加到会话 -> 提交事务。三个步骤缺一不可：

from datetime import datetime

# 创建一个新用户
new_user = User(
    username='zhangsan',
    email='zhangsan@example.com',
    is_active=True,
    created_at=datetime.utcnow()
)

# 添加到数据库会话
db.session.add(new_user)

# 批量添加多个对象
db.session.add_all([user1, user2, user3])

# 提交事务——只有执行commit后数据才会真正写入数据库
db.session.commit()

# 提交后可以获取自增主键
print(new_user.id)  # 输出自动生成的ID

核心要点：commit() 是必不可少的。如果在add之后没有调用commit()，当请求结束或会话关闭时，所有未提交的更改会丢失。建议将数据库操作放在try/except块中，出错时调用 db.session.rollback() 回滚事务。

4.2 查询（Read）

SQLAlchemy提供了丰富多样的查询方法，是日常开发中使用频率最高的操作：

# 查询所有记录
all_users = User.query.all()

# 根据主键查询
user = User.query.get(1)

# 精确查询（filter_by - 使用关键字参数）
user = User.query.filter_by(username='zhangsan').first()

# 条件查询（filter - 使用比较表达式）
users = User.query.filter(User.age > 18).all()
users = User.query.filter(User.username != 'admin').all()

# 模糊查询
users = User.query.filter(User.username.like('%zhang%')).all()

# IN 查询
users = User.query.filter(User.id.in_([1, 3, 5])).all()

# 范围查询
users = User.query.filter(User.age.between(18, 60)).all()

# 包含查询
users = User.query.filter(User.email.contains('example')).all()

# 多条件组合（AND）
users = User.query.filter(
    User.age >= 18,
    User.is_active == True
).all()

# OR 条件（需要导入 or_）
from sqlalchemy import or_
users = User.query.filter(
    or_(User.username == 'admin', User.is_admin == True)
).all()

4.3 排序、限制与偏移

# 排序（升序）
users = User.query.order_by(User.created_at).all()

# 排序（降序）
users = User.query.order_by(User.created_at.desc()).all()

# 多字段排序
users = User.query.order_by(
    User.is_active.desc(),
    User.created_at.desc()
).all()

# 限制条数 + 偏移（用于分页）
users = User.query.order_by(User.id).limit(10).offset(20).all()

# 更优雅的分页方式
page = User.query.paginate(page=1, per_page=10, error_out=False)
print(page.items)      # 当前页数据
print(page.total)      # 总记录数
print(page.pages)      # 总页数
print(page.has_prev)   # 是否有上一页
print(page.has_next)   # 是否有下一页

4.4 更新（Update）

更新操作非常直观——先查询到对象，修改属性，然后提交：

# 方式一：查询后修改
user = User.query.get(1)
user.email = 'newemail@example.com'
user.is_active = False
db.session.commit()

# 方式二：批量更新（不加载对象，直接执行SQL）
User.query.filter(User.is_active == False).update(
    {'is_active': True}
)
db.session.commit()

4.5 删除（Delete）

# 方式一：查询后删除
user = User.query.get(1)
db.session.delete(user)
db.session.commit()

# 方式二：条件删除
User.query.filter(User.is_active == False).delete()
db.session.commit()

4.6 聚合查询

from sqlalchemy import func

# 计数
count = db.session.query(func.count(User.id)).scalar()

# 求和
total = db.session.query(func.sum(Order.amount)).scalar()

# 平均值
avg = db.session.query(func.avg(Order.amount)).scalar()

# 最大值 / 最小值
max_price = db.session.query(func.max(Product.price)).scalar()
min_price = db.session.query(func.min(Product.price)).scalar()

# 分组统计
from sqlalchemy import func
results = db.session.query(
    Category.name,
    func.count(Product.id).label('product_count')
).join(Product).group_by(Category.id).all()

五、模型关系（重点）

5.1 一对多关系

一对多关系是数据库中最常见的关系类型，例如一个作者可以有多篇文章。实现方式是在"多"的一方定义外键，在"一"的一方定义 relationship：

class Author(db.Model):
    __tablename__ = 'authors'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)

    # 定义关系 - 通过 backref 自动为 Post 模型创建 author 属性
    posts = db.relationship('Post', backref='author', lazy='dynamic')

    def __repr__(self):
        return f''


class Post(db.Model):
    __tablename__ = 'posts'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    content = db.Column(db.Text, nullable=False)
    author_id = db.Column(db.Integer, db.ForeignKey('authors.id'), nullable=False)

    def __repr__(self):
        return f''

使用示例：

# 通过作者获取所有文章
author = Author.query.get(1)
posts = author.posts.all()  # lazy='dynamic' 返回查询对象，可继续链式调用

# 通过文章获取作者
post = Post.query.get(1)
print(post.author.name)  # backref 自动添加了 author 属性

5.2 一对多关系的替代方式：back_populates

相比 backref 的隐式创建，back_populates 要求双方显式声明关系，可读性更好，是大型项目中的推荐做法：

class Author(db.Model):
    __tablename__ = 'authors'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)
    posts = db.relationship('Post', back_populates='author')


class Post(db.Model):
    __tablename__ = 'posts'
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    author_id = db.Column(db.Integer, db.ForeignKey('authors.id'))
    author = db.relationship('Author', back_populates='posts')

5.3 多对多关系

多对多关系需要一张关联表来实现，例如学生和课程之间的关系——一个学生可以选择多门课程，一门课程也可以被多个学生选择。关联表中不包含业务数据，仅存储两方的外键：

# 关联表（不对应模型类，仅作为SQLAlchemy的表对象）
student_course = db.Table('student_course',
    db.Column('student_id', db.Integer, db.ForeignKey('students.id'), primary_key=True),
    db.Column('course_id', db.Integer, db.ForeignKey('courses.id'), primary_key=True)
)


class Student(db.Model):
    __tablename__ = 'students'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)

    # secondary 指向关联表
    courses = db.relationship('Course', secondary=student_course,
                              backref='students', lazy='dynamic')


class Course(db.Model):
    __tablename__ = 'courses'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)

多对多操作示例：

# 创建学生和课程
s1 = Student(name='小明')
c1 = Course(title='Python基础')
c2 = Course(title='数据结构')

# 关联学生和课程
s1.courses.append(c1)
s1.courses.append(c2)
db.session.commit()

# 查询某个学生的所有课程
for course in s1.courses.all():
    print(course.title)

# 查询某门课程的所有学生
course = Course.query.get(1)
for student in course.students:
    print(student.name)

5.4 一对一关系

一对一关系可以看作是一对多的特例，通过设置 uselist=False 来实现。例如一个用户有一个身份证信息：

class User(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)

    # uselist=False 表示这是一个一对一关系
    profile = db.relationship('Profile', backref='user', uselist=False)


class Profile(db.Model):
    __tablename__ = 'profiles'

    id = db.Column(db.Integer, primary_key=True)
    id_number = db.Column(db.String(18), unique=True, nullable=False)
    phone = db.Column(db.String(20))
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), unique=True)

5.5 延迟加载策略（lazy）

lazy 参数控制关联数据的加载时机和方式。合理选择加载策略对应用性能影响巨大：

lazy 值	加载时机	说明	适用场景
'select'	首次访问时	默认值，访问关系时发送一条SELECT查询	大多数场景
'joined'	查询父对象时	使用JOIN一次性加载关联数据	确定需要访问关联数据时
'subquery'	查询父对象时	使用子查询加载关联数据	复杂查询的替代方案
'dynamic'	不自动加载	返回Query对象，支持链式过滤	关联数据量大，需过滤时
True	首次访问时	已废弃，等效于'select'	兼容旧代码
False	查询父对象时	已废弃，等效于'joined'	兼容旧代码

性能建议：当一对多关系中的"多"方数据量很大时，优先使用 lazy='dynamic'，以便可以链式添加过滤条件。例如 author.posts.filter(Post.is_published == True).all()。如果确定每次都要访问关联数据，使用 lazy='joined' 可以避免N+1查询问题。

5.6 级联操作（cascade）

级联操作定义了父对象上的操作如何传播到关联的子对象，是关系管理中的重要概念：

class Author(db.Model):
    __tablename__ = 'authors'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)

    # cascade='all, delete-orphan' 表示：
    # - save-update：添加作者时自动保存文章
    # - delete：删除作者时删除所有文章
    # - delete-orphan：从作者的文章列表中移除的文章自动删除
    posts = db.relationship('Post', backref='author',
                            cascade='all, delete-orphan', lazy='dynamic')


# 测试级联删除
author = Author.query.get(1)
db.session.delete(author)  # 级联删除该作者的所有文章
db.session.commit()

常用的级联策略组合：

级联策略	行为说明
'save-update'	父对象add到session时，关联的子对象也自动add
'delete'	删除父对象时，同时删除关联的子对象
'delete-orphan'	解除关联的子对象自动被删除
'all'	包含save-update、merge、refresh-expire、expunge、delete
'all, delete-orphan'	最常用组合，父对象完全拥有子对象生命周期

六、数据库迁移

6.1 Flask-Migrate 简介

在项目迭代过程中，数据模型经常需要变更——增加字段、修改约束、新建表等。Flask-Migrate是基于Alembic的Flask扩展，提供了数据库迁移能力，能够自动生成迁移脚本、追踪数据库的版本变更历史，并支持团队协作中的迁移同步。Alembic会在数据库中维护一张 alembic_version 表，记录当前数据库所处的迁移版本。

6.2 安装与初始化

# 安装
pip install flask-migrate

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

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///data.db'
db = SQLAlchemy(app)

# 初始化 Flask-Migrate
migrate = Migrate(app, db)

初始化迁移环境（在终端执行）：

# 创建迁移目录（生成 migrations/ 文件夹）
flask db init

执行后项目目录中会生成 migrations/ 文件夹，包含：

migrations/env.py — 迁移环境配置，自动集成Flask应用上下文
migrations/alembic.ini — Alembic配置文件
migrations/versions/ — 存放迁移脚本的目录
migrations/script.py.mako — 迁移脚本模板

6.3 生成迁移脚本

当模型定义更新后，执行以下命令自动生成迁移脚本：

# 自动检测模型变更并生成迁移脚本
# -m 参数提供描述信息，便于追踪版本
flask db migrate -m "add email field to User model"

生成的迁移脚本位于 migrations/versions/ 目录下，包含 upgrade() 和 downgrade() 两个函数：

"""add email field to User model

Revision ID: abc123def456
Revises: xyz789abc123
Create Date: 2026-05-05 23:14:00.000000
"""
from alembic import op
import sqlalchemy as sa

# revision identifiers
revision = 'abc123def456'
down_revision = 'xyz789abc123'


def upgrade():
    """应用迁移：新增 email 字段"""
    op.add_column('users', sa.Column('email', sa.String(length=120), nullable=True))


def downgrade():
    """回滚迁移：删除 email 字段"""
    op.drop_column('users', 'email')

6.4 应用与回滚迁移

# 应用所有待执行的迁移（升级到最新版本）
flask db upgrade

# 回滚一个版本
flask db downgrade

# 回滚到指定版本
flask db downgrade abc123def456

# 查看当前版本和历史
flask db history      # 查看所有迁移历史
flask db current      # 查看当前数据库所处的版本
flask db show         # 显示当前版本的详细信息

6.5 迁移最佳实践

迁移工作流：

修改模型定义（新增/修改/删除字段或表）
运行 flask db migrate -m "描述信息" 生成迁移脚本
务必审查生成的迁移脚本，确认自动检测是否正确
在开发环境运行 flask db upgrade 测试迁移
将迁移脚本提交到版本控制（Git）
在测试/生产环境拉取代码后运行 flask db upgrade

重要警告：

迁移脚本一旦提交到版本控制，禁止修改已合并到主分支的迁移脚本。如需调整，应创建新的迁移
在 flask db migrate 之前，确保当前数据库已通过 flask db upgrade 更新到最新状态
添加非空字段时，需要为新字段指定默认值，否则迁移会失败：
db.Column(db.String(100), nullable=False, server_default='')
重命名表或字段时，Alembic无法自动检测，需要手动编辑迁移脚本

七、核心要点总结

本章学习要点回顾：

ORM 的核心思想是通过对象操作数据库，降低代码耦合度，提高开发效率
Flask-SQLAlchemy 的配置围绕 SQLALCHEMY_DATABASE_URI 展开，支持多数据库绑定
模型定义涵盖丰富的列类型和约束条件，是数据库设计的基石
CRUD 操作遵循"增删改查"四类基本操作，其中查询操作最为灵活多样
模型关系包括一对多（ForeignKey + relationship）、多对多（secondary关联表）、一对一（uselist=False）
延迟加载策略（lazy）的选择直接影响查询性能，需要根据场景合理配置
级联操作（cascade）管理父子对象的生命周期联动
Flask-Migrate 是生产环境管理数据库版本变更的标准工具，迁移脚本务必纳入版本控制
事务管理遵循"操作 -> commit / rollback"的基本模式，确保数据一致性

八、进一步思考与练习

1. N+1 查询问题：在使用 lazy='select' 策略时，循环访问每个父对象的关联属性会产生N+1条SQL语句。请思考如何通过 eager loading（joined/subquery）或者 selectinload 来解决这个问题。

2. 事务隔离级别：SQLAlchemy支持设置事务隔离级别（READ UNCOMMITTED / READ COMMITTED / REPEATABLE READ / SERIALIZABLE）。尝试在代码中通过 db.session.connection(execution_options={'isolation_level': 'SERIALIZABLE'}) 设置，并观察在高并发场景下的不同表现。

3. 混合属性（Hybrid Attributes）：SQLAlchemy提供的 @hybrid_property 装饰器允许定义同时作用于Python层和SQL层的计算属性，是复用业务逻辑的强大工具。

4. 综合练习：设计一个博客系统的数据模型，包含用户（User）、文章（Article）、分类（Category）、标签（Tag）四个实体。要求：用户与文章为一对多，文章与分类为多对一，文章与标签为多对多。实现文章的分页查询、按标签筛选、按分类统计等功能。

5. 性能优化练习：在一个包含10万条用户记录和50万条订单记录的数据库中，比较使用 lazy='joined'、lazy='select' 和 db.session.query().options(db.joinedload()) 三种方式查询用户及其最新订单的性能差异。