Flask数据库操作与迁移教程

引言

Flask是一个轻量级的Python Web框架，常用于快速开发Web应用。在Flask中，数据库操作是核心功能之一，通常通过扩展如SQLAlchemy集成。本教程将深入介绍如何在Flask中进行数据库操作，包括CRUD、查询集进阶、数据库迁移以及事务处理，适合新人从零开始学习。

1. 数据库基础与Flask集成

在Flask中，推荐使用SQLAlchemy作为ORM（对象关系映射）工具来管理数据库。首先，安装Flask-SQLAlchemy扩展：

pip install Flask-SQLAlchemy

然后在Flask应用中配置数据库连接，例如使用SQLite：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'  # SQLite数据库路径
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False  # 禁用跟踪修改以提升性能
db = SQLAlchemy(app)

# 定义数据模型
class User(db.Model):
    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)

2. CRUD操作详解

CRUD代表创建（Create）、查询（Read）、更新（Update）和删除（Delete），是数据库的基本操作。

2.1 创建（Create）

使用db.session.add()添加新记录到会话，然后提交到数据库：

# 创建新用户
new_user = User(username='john', email='john@example.com')
db.session.add(new_user)  # 添加到会话
db.session.commit()  # 提交到数据库
print("用户创建成功")

2.2 查询（Read）

使用query()方法查询数据，可以检索单个或多个记录：

# 查询所有用户
all_users = User.query.all()
for user in all_users:
    print(user.username, user.email)

# 查询单个用户
user = User.query.filter_by(username='john').first()  # 使用filter_by
if user:
    print(f"找到用户: {user.username}")

2.3 更新（Update）

更新现有记录：先查询到对象，修改属性，然后提交：

# 更新用户邮箱
user = User.query.filter_by(username='john').first()
if user:
    user.email = 'newjohn@example.com'
    db.session.commit()
    print("用户邮箱更新成功")

2.4 删除（Delete）

删除记录：查询对象后，使用db.session.delete()：

# 删除用户
user = User.query.filter_by(username='john').first()
if user:
    db.session.delete(user)
    db.session.commit()
    print("用户删除成功")

3. 查询集（Query）进阶与过滤条件

查询集是SQLAlchemy中Query对象的实例，允许链式调用和复杂过滤。

3.1 常用过滤方法

• filter(): 用于复杂条件，如等于、大于等。
• filter_by(): 用于简单键值对过滤。
• order_by(): 排序结果。
• limit() 和 offset(): 分页查询。

示例代码：

# 使用filter进行条件查询
users = User.query.filter(User.email.endswith('@example.com')).all()  # 邮箱以@example.com结尾的用户

# 链式调用
users = User.query.filter(User.id > 1).order_by(User.username.desc()).limit(5).all()  # ID大于1的用户，按用户名降序，限制5个

# 使用filter_by
user = User.query.filter_by(username='john', email='john@example.com').first()  # 同时过滤用户名和邮箱

3.2 高级过滤条件

SQLAlchemy支持运算符如==, !=, >, <, in_, like等。例如：

from sqlalchemy import or_

# 使用or_组合条件
users = User.query.filter(or_(User.username == 'john', User.email.like('%example%'))).all()  # 用户名是john或邮箱包含example的用户

4. 数据库迁移（Flask-Migrate）使用

数据库迁移用于管理数据库模式变更，避免手动修改表结构。Flask-Migrate是基于Alembic的扩展。

4.1 安装与配置

安装Flask-Migrate：

pip install Flask-Migrate

在Flask应用中初始化：

from flask_migrate import Migrate

migrate = Migrate(app, db)  # 关联app和db

4.2 使用迁移命令

在终端中，首先初始化迁移环境：

flask db init  # 创建迁移文件夹

然后生成迁移脚本，当模型有变化时：

flask db migrate -m "Initial migration"  # 生成迁移脚本，描述变更

最后应用迁移到数据库：

flask db upgrade  # 升级数据库到最新版本

如果出错，可以回滚：

flask db downgrade  # 降级到上一个版本

示例：修改User模型添加年龄字段：

# 在模型中添加
age = db.Column(db.Integer, nullable=True)

然后运行flask db migrate -m "Add age to User"和flask db upgrade。

5. 事务管理与异常处理

事务确保数据库操作的原子性，要么全部成功，要么全部回滚。在Flask-SQLAlchemy中，会话（session）管理事务。

5.1 事务基础

默认情况下，db.session.commit()提交事务。如果操作失败，使用db.session.rollback()回滚：

try:
    user1 = User(username='alice', email='alice@example.com')
    user2 = User(username='bob', email='bob@example.com')
    db.session.add(user1)
    db.session.add(user2)
    db.session.commit()  # 提交事务
    print("事务提交成功")
except Exception as e:
    db.session.rollback()  # 回滚事务，避免部分数据写入
    print(f"事务失败，已回滚: {e}")

5.2 异常处理最佳实践

• 使用try-except块捕获数据库操作异常。
• 确保在异常时调用rollback()。
• 使用Flask的错误处理装饰器如@app.errorhandler全局处理错误。

示例：处理唯一性约束错误：

from sqlalchemy.exc import IntegrityError

try:
    user = User(username='john', email='john@example.com')
    db.session.add(user)
    db.session.commit()
except IntegrityError:
    db.session.rollback()
    print("用户名或邮箱已存在，请检查")

总结

本教程覆盖了Flask中数据库操作与迁移的核心概念。从基础的CRUD操作到查询集进阶，再到使用Flask-Migrate进行数据库迁移，以及事务管理和异常处理，帮助新人构建稳健的Flask应用。建议实践项目加深理解，并查阅官方文档获取更多信息。

通过以上步骤，你可以高效地在Flask中进行数据库开发，确保数据的安全性和一致性。