---

快速上手 Flask-SQLAlchemy：打造你的第一个数据库驱动的Flask应用

在构建现代 Web 应用时，数据存储是不可或缺的一环。关系型数据库因其结构化、易于管理等特性，是许多应用的基石。而作为 Python 开发者，我们常常会使用 ORM（Object-Relational Mapper，对象关系映射）来与数据库交互，而非直接编写 SQL 语句。ORM 能够将数据库中的表映射成 Python 对象，将行映射成对象实例，将列映射成对象属性，极大地提高了开发效率和代码的可维护性。

对于 Flask 这个轻量级 Web 框架来说，Flask-SQLAlchemy 是官方推荐的、集成 SQLAlchemy 这个强大 Python ORM 的扩展。它简化了在 Flask 应用中使用 SQLAlchemy 的配置和常见操作。

本文将带你从零开始，快速掌握 Flask-SQLAlchemy 的基本使用，构建一个简单的数据库驱动的 Flask 应用。

1. 为什么选择 Flask-SQLAlchemy？

• Pythonic 风格： 使用 Python 对象和方法操作数据库，无需编写复杂的 SQL。
• 抽象数据库差异： SQLAlchemy 支持多种数据库后端（SQLite, PostgreSQL, MySQL等），无需关心底层 SQL 语法差异。
• 强大的功能： 提供了强大的查询 API、关系管理、事务处理等功能。
• 与 Flask 无缝集成： Flask-SQLAlchemy 简化了配置、上下文管理等工作，让 SQLAlchemy 在 Flask 中使用更加便捷。

2. 前置准备

在开始之前，请确保你已经安装了 Python，并了解基本的 Flask 框架用法。

我们需要安装 Flask 和 Flask-SQLAlchemy：

bash
pip install Flask Flask-SQLAlchemy

推荐使用虚拟环境（Virtual Environment）：为了避免项目之间的依赖冲突，强烈建议为每个项目创建一个独立的虚拟环境。

“`bash

创建虚拟环境 (命名为 venv)

python -m venv venv

激活虚拟环境

Windows

venv\Scripts\activate

macOS/Linux

source venv/bin/activate

在激活的环境中安装库

pip install Flask Flask-SQLAlchemy
“`

3. 基本配置

首先，我们需要创建一个 Flask 应用实例，并配置 Flask-SQLAlchemy。在一个名为 app.py 的文件中：

“`python
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import os # 用于构建文件路径

创建 Flask 应用实例

app = Flask(name)

配置数据库 URI

SQLite 是一个轻量级的基于文件的数据库，适合开发和小型应用。

在生产环境中，通常会使用 PostgreSQL, MySQL 等。

这里使用绝对路径来指定数据库文件位置

basedir = os.path.abspath(os.path.dirname(file))
app.config[‘SQLALCHEMY_DATABASE_URI’] = ‘sqlite:///’ + os.path.join(basedir, ‘data.sqlite’)

禁用 Flask-SQLAlchemy 事件通知系统，可以减少开销

如果你不需要跟踪对象的修改，设置为 False 更好

app.config[‘SQLALCHEMY_TRACK_MODIFICATIONS’] = False

创建 SQLAlchemy 实例，并与 Flask 应用关联

db = SQLAlchemy(app)

… 后续的模型定义和路由代码 …

if name == ‘main‘:
app.run(debug=True)
“`

配置说明：

• SQLALCHEMY_DATABASE_URI: 这是最重要的配置项，指定了数据库的连接地址。
  • sqlite:///data.sqlite: 连接当前目录下的 data.sqlite 文件。/// 表示绝对路径。
  • sqlite:////absolute/path/to/data.sqlite: 指定一个绝对路径。
  • postgresql://user:password@host:port/dbname: 连接 PostgreSQL 数据库。
  • mysql://user:password@host/dbname: 连接 MySQL 数据库（需要安装相应的数据库驱动，如 psycopg2 for PostgreSQL, PyMySQL or mysql-connector-python for MySQL）。
• SQLALCHEMY_TRACK_MODIFICATIONS: 设置为 False 可以关闭修改跟踪，除非你明确需要这个功能，否则建议关闭以提高性能。

4. 定义数据库模型（Models）

数据库模型是 Python 类，它们映射到数据库表。在模型类中，我们定义属性来映射表的列。这些模型类需要继承自 db.Model。

“`python
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import os
from datetime import datetime # 用于日期时间字段

basedir = os.path.abspath(os.path.dirname(file))
app = Flask(name)
app.config[‘SQLALCHEMY_DATABASE_URI’] = ‘sqlite:///’ + os.path.join(basedir, ‘data.sqlite’)
app.config[‘SQLALCHEMY_TRACK_MODIFICATIONS’] = False
db = SQLAlchemy(app)

定义一个用户模型

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) # 邮箱，字符串，唯一，非空
# created_at = db.Column(db.DateTime, default=datetime.utcnow) # 创建时间，默认当前UTC时间

# 定义关系 (可选): 一个用户可以有很多文章
# 'Post' 是关联的模型类名
# lazy='dynamic' 意味着 posts 属性会返回一个查询对象，而不是列表，适合处理大量关联对象
posts = db.relationship('Post', backref='author', lazy='dynamic')

# 定义一个方便调试的方法
def __repr__(self):
    return f'<User {self.username}>'

定义一个文章模型

class Post(db.Model):
id = db.Column(db.Integer, primary_key=True) # 主键
title = db.Column(db.String(120), nullable=False) # 标题
body = db.Column(db.Text, nullable=False) # 内容，Text 类型适合存储较长的文本
timestamp = db.Column(db.DateTime, default=datetime.utcnow) # 发布时间
# 外键：关联到 users 表的 id 列
# user_id = db.Column(db.Integer, db.ForeignKey(‘user.id’), nullable=False) # ‘user.id’ 注意这里是表名.列名

# # backref 已经在 User 模型中定义了，这里的 author 属性会自动生成
# author = db.relationship('User', backref=db.backref('posts', lazy='dynamic'))

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

… 后续的路由代码 …

if name == ‘main‘:
# 在应用启动时创建数据库表（仅用于开发环境，生产环境请使用迁移工具）
with app.app_context():
db.create_all() # 在应用上下文中创建所有表
app.run(debug=True)

“`

字段类型（部分常用）：

• db.Integer: 整型
• db.String(length): 字符串，需要指定最大长度
• db.Text: 长文本
• db.Boolean: 布尔型
• db.DateTime: 日期时间
• db.Float: 浮点型
• db.Enum: 枚举类型
• db.PickleType: 用于存储序列化的 Python 对象

字段选项（部分常用）：

• primary_key=True: 设置为主键
• unique=True: 设置为唯一约束
• nullable=False: 设置为非空约束
• default=value: 设置默认值
• index=True: 为该列创建索引，加快查询速度
• ForeignKey('tablename.columnname'): 设置为外键

关系 (db.relationship) 说明：

db.relationship 用于在一个模型中定义与另一个模型之间的关系。

• 在 User 模型中定义 posts：表示一个 User 对象可以通过 user.posts 访问其关联的 Post 对象列表。
• 'Post': 指定关联的模型类名。
• backref='author': 在关联的模型 (Post) 中创建一个名为 author 的属性，通过 post.author 可以访问该文章所属的 User 对象。
• lazy='dynamic': 对于一对多关系，如果关联对象可能很多，设置为 dynamic 可以让 user.posts 返回一个查询对象，而不是加载所有关联对象到内存中，从而提高效率。你可以继续在这个查询对象上进行过滤、排序等操作。

外键 (db.ForeignKey) 说明：

db.ForeignKey('tablename.columnname') 用于定义外键约束。它指向另一个表的哪个列是其主键。

• 在 Post 模型中定义 user_id：db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) 表示 Post 表的 user_id 列是一个外键，它指向 user 表的 id 列。注意，ForeignKey 中引用的是表名（通常是模型类名的小写）和列名。

5. 创建数据库表

定义好模型后，我们需要根据模型创建实际的数据库表。这通常通过运行 db.create_all() 来完成。

重要： db.create_all() 只能在应用上下文中运行。推荐使用 Flask 提供的 flask shell 命令来执行数据库操作。

1. 设置 Flask 应用环境变量: 在命令行中（或你的终端配置文件如 .bashrc, .zshrc 中）设置 FLASK_APP 环境变量指向你的应用文件。

   “`bash

   假设你的应用文件是 app.py

   export FLASK_APP=app.py
   “`

   或者直接在运行 flask 命令时指定：flask --app app.py shell

2. 启动 Flask shell: 激活你的虚拟环境，然后运行：

   bash
flask shell

   进入 shell 后，Flask 会自动加载你的应用上下文，你可以在其中访问 db 对象和你的模型类。

3. 创建表: 在 shell 中输入以下命令：

   “`python

   from app import db # 导入 db 实例
   db.create_all() # 执行创建所有表的命令

   如果需要删除所有表（仅在开发环境调试时使用！）

   db.drop_all()

   exit() # 退出 shell
   “`

运行 db.create_all() 后，你应该会在你的项目目录下看到一个 data.sqlite 文件（如果你配置的是 SQLite）。

关于数据库迁移： db.create_all() 只能创建表，无法处理模型的修改（例如添加新列、修改列类型等）。在生产环境中或项目迭代过程中，你需要使用数据库迁移工具，最常用的是 Flask-Migrate (它集成了 Alembic)。本文作为快速上手指南不详细展开，但你需要知道它的存在。

6. 数据库基本操作：CRUD (创建、读取、更新、删除)

Flask-SQLAlchemy 提供了一个 db.session 对象，它是与数据库进行所有交互的“暂存区”。所有对数据库的修改（创建、更新、删除）都需要通过 session 进行添加、删除操作，并通过 session.commit() 来最终确认提交到数据库。查询操作也通过模型类的 .query 属性进行。

我们可以在 Flask shell 中演示这些操作，或者在 Flask 路由函数中执行。

在 Flask shell 中演示 CRUD

首先，确保你已经运行了 db.create_all() 并且数据库文件存在。然后再次进入 flask shell。

bash
flask shell

创建 (Create)

“`python

from app import db, User, Post # 导入 db 实例和模型类

创建一个新的用户实例

user1 = User(username=’john_doe’, email=’[email protected]’)
user2 = User(username=’jane_smith’, email=’[email protected]’)

创建一篇新文章并关联作者 (通过 backref 生成的 author 属性)

post1 = Post(title=’My First Post’, body=’This is the content of my first post.’, author=user1)
post2 = Post(title=’Another Post’, body=’Content of the second post.’, author=user1)
post3 = Post(title=’Jane\’s Post’, body=’Content by Jane.’, author=user2)

将新创建的对象添加到 session 中

db.session.add(user1)
db.session.add(user2)
db.session.add(post1)
db.session.add(post2)
db.session.add(post3)

或者可以使用 add_all 添加多个

db.session.add_all([user1, user2, post1, post2, post3])

提交 session，将所有变更保存到数据库

db.session.commit()

print(“Users and posts added successfully!”)
“`

读取 (Read)

查询是使用模型类的 .query 属性进行的。

“`python

from app import User, Post # 导入模型类

查询所有用户

all_users = User.query.all()
print(“所有用户:”, all_users) # 会调用模型的 repr 方法显示

根据主键查询单个用户 (推荐使用 get())

user_by_id = User.query.get(1)
print(“ID 为 1 的用户:”, user_by_id)

根据条件查询第一个匹配的用户 (使用 filter_by)

john = User.query.filter_by(username=’john_doe’).first()
print(“用户名是 john_doe 的用户:”, john)

根据条件查询所有匹配的用户 (使用 filter)

users_example = User.query.filter(User.email.endswith(‘@example.com’)).all()
print(“邮箱后缀为 @example.com 的用户:”, users_example)

使用关系查询用户的文章

if john:
… john_posts = john.posts.all() # 因为 lazy=’dynamic’，所以需要调用 .all() 获取结果列表
… print(f”{john.username} 的文章:”, john_posts)

查询所有的文章

all_posts = Post.query.all()
print(“所有文章:”, all_posts)

查询特定用户的文章 (通过外键或关系)

posts_by_john = Post.query.filter_by(author=john).all() # 使用关系查询
print(f”通过关系查询 {john.username} 的文章:”, posts_by_john)

或者使用外键ID: Post.query.filter_by(user_id=john.id).all()

排序

sorted_users = User.query.order_by(User.username).all()
print(“按用户名排序的用户:”, sorted_users)

倒序排序

sorted_users_desc = User.query.order_by(User.username.desc()).all()

限制和偏移 (分页)

first_user = User.query.limit(1).first() # 获取第一个
users_after_first = User.query.offset(1).all() # 跳过第一个，获取剩余的
print(“第一个用户:”, first_user)
print(“跳过第一个的用户:”, users_after_first)

结合使用：User.query.offset((page-1)*per_page).limit(per_page).all() 实现分页

“`

更新 (Update)

更新一个对象只需先查询到它，然后修改其属性，最后提交 session。

“`python

from app import db, User

查询要更新的用户

user_to_update = User.query.filter_by(username=’john_doe’).first()

检查用户是否存在

if user_to_update:
… # 修改用户的邮箱
… user_to_update.email = ‘[email protected]’
… # 将变更添加到 session (实际上 SQLAlchemy 已经跟踪了修改，这步可以省略，但明确写出有助于理解)
… # db.session.add(user_to_update)
… # 提交 session 保存变更
… db.session.commit()
… print(f”用户 {user_to_update.username} 的邮箱已更新。”)
… else:
… print(“用户未找到。”)

验证更新

updated_user = User.query.filter_by(username=’john_doe’).first()
print(“更新后的用户信息:”, updated_user.email)
“`

删除 (Delete)

删除一个对象只需先查询到它，然后使用 db.session.delete() 删除，最后提交 session。

“`python

from app import db, User, Post

查询要删除的用户 (及其关联的文章)

user_to_delete = User.query.filter_by(username=’john_doe’).first()

if user_to_delete:
… # 注意：删除用户时，默认情况下其关联的文章会保留（外键约束可能不允许）。
… # 通常需要在关系中配置 cascade=’all, delete-orphan’ 来实现级联删除。
… # 这里我们手动删除与该用户相关的文章以避免外键冲突（如果外键是非空的）
… # Alternatively, update the foreign key to null or a default user.
… # For this example, let’s assume we manually handle related posts or they are deleted.
… # Simple case: Assuming the relationship doesn’t block deletion or posts are handled elsewhere.
… # If you configured cascade delete on the relationship, deleting the user would delete posts.
… # Let’s manually delete posts for demonstration if needed, or just delete the user if FK allows NULL
… # Or if cascade=’all,delete’ was on the relationship:
… # db.session.delete(user_to_delete) # This would delete user and potentially posts depending on cascade

… # Let’s demonstrate deleting a post
… post_to_delete = Post.query.filter_by(title=’Another Post’).first()
… if post_to_delete:
… db.session.delete(post_to_delete)
… db.session.commit()
… print(f”文章 ‘{post_to_delete.title}’ 已删除。”)
… else:
… print(“文章未找到。”)

… # Now delete the user (assuming related posts are handled or FK is nullable/cascade is set)
… db.session.delete(user_to_delete)
… db.session.commit()
… print(f”用户 {user_to_delete.username} 已删除。”)

else:
… print(“用户未找到。”)

验证删除

deleted_user = User.query.filter_by(username=’john_doe’).first()
print(“删除后的用户:”, deleted_user) # 应该输出 None
deleted_post = Post.query.filter_by(title=’Another Post’).first()
print(“删除后的文章:”, deleted_post) # 应该输出 None
“`

关于 db.session 和事务：

• db.session 是一个数据库会话，代表了当前操作的暂存区。
• db.session.add(obj), db.session.delete(obj) 等操作只是将对象标记为待处理的状态。
• db.session.commit() 将 session 中的所有待处理操作批量提交到数据库，形成一个原子事务。如果提交过程中发生错误，整个事务会回滚（rollback），之前的操作都不会生效。
• db.session.rollback() 用于回滚当前 session 中未提交的变更，回到上次提交或 session 开始时的状态。
• Flask-SQLAlchemy 会在请求结束时自动关闭 session。在路由函数中，你通常只需要关心 add/delete 和 commit。

在 Flask 路由函数中使用 CRUD

将数据库操作集成到 Flask 路由中，就可以构建动态的 Web 应用。

继续编辑 app.py，添加路由函数：

“`python
from flask import Flask, render_template_string, request, redirect, url_for
from flask_sqlalchemy import SQLAlchemy
import os
from datetime import datetime

basedir = os.path.abspath(os.path.dirname(file))
app = Flask(name)
app.config[‘SQLALCHEMY_DATABASE_URI’] = ‘sqlite:///’ + os.path.join(basedir, ‘data.sqlite’)
app.config[‘SQLALCHEMY_TRACK_MODIFICATIONS’] = False
db = SQLAlchemy(app)

定义模型 (User 和 Post，如前所示)

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)
posts = db.relationship(‘Post’, backref=’author’, lazy=’dynamic’)
def repr(self):
return f’‘

class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(120), nullable=False)
body = db.Text(nullable=False)
timestamp = db.Column(db.DateTime, default=datetime.utcnow)
user_id = db.Column(db.Integer, db.ForeignKey(‘user.id’), nullable=False) # Foreign key link

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

— 数据库操作集成到路由 —

首页：显示所有用户和他们的文章数量

@app.route(‘/’)
def index():
users = User.query.all()
# 一个简单的 HTML 模板字符串
html = “””

用户列表

{% for user in users %}
• {{ user.username }} ({{ user.email }}) – 文章数: {{ user.posts.count() }}
  | 查看文章
  | 删除用户

{% endfor %}

添加新用户

用户名:
邮箱:

---

文章列表

{% for post in posts %}
• {{ post.title }} by {{ post.author.username if post.author else ‘未知用户’ }} ({{ post.timestamp.strftime(‘%Y-%m-%d %H:%M’) }})
  | 删除文章

{% endfor %}

添加新文章

作者ID (例如: 1):
标题:
内容:

“””
posts = Post.query.order_by(Post.timestamp.desc()).all() # 按时间倒序获取文章
return render_template_string(html, users=users, posts=posts)

添加新用户

@app.route(‘/add_user’, methods=[‘POST’])
def add_user():
username = request.form[‘username’]
email = request.form[’email’]
# 检查用户名或邮箱是否已存在
existing_user = User.query.filter_by(username=username).first() or User.query.filter_by(email=email).first()
if existing_user:
return “用户或邮箱已存在！”, 400 # 返回错误信息和状态码

new_user = User(username=username, email=email)
db.session.add(new_user)
try:
    db.session.commit()
    return redirect(url_for('index')) # 添加成功后重定向回首页
except Exception as e:
    db.session.rollback() # 如果提交失败，回滚事务
    return f"添加用户失败: {e}", 500

查看某个用户的文章

@app.route(‘/user//posts’)
def view_user_posts(user_id):
user = User.query.get_or_404(user_id) # 获取用户，如果不存在返回 404
# 使用关系获取该用户的所有文章
posts = user.posts.order_by(Post.timestamp.desc()).all()
html = “””

{{ user.username }} 的文章

{% for post in posts %}
• {{ post.title }} ({{ post.timestamp.strftime(‘%Y-%m-%d %H:%M’) }})

{% else %}

• 该用户还没有文章。

{% endfor %}

返回首页

“””
return render_template_string(html, user=user, posts=posts)

添加新文章

@app.route(‘/add_post’, methods=[‘POST’])
def add_post():
user_id = request.form[‘user_id’]
title = request.form[‘title’]
body = request.form[‘body’]

author = User.query.get(user_id) # 获取作者用户
if not author:
    return "作者用户不存在！", 400

new_post = Post(title=title, body=body, author=author) # 通过关系设置作者
# 或者直接设置外键ID: new_post = Post(title=title, body=body, user_id=user_id)

db.session.add(new_post)
try:
    db.session.commit()
    return redirect(url_for('index'))
except Exception as e:
    db.session.rollback()
    return f"添加文章失败: {e}", 500

删除用户

@app.route(‘/delete_user/‘)
def delete_user(user_id):
user_to_delete = User.query.get_or_404(user_id)

try:
    # 删除关联的文章 (取决于你的外键设置和级联删除配置)
    # 如果外键是 ON DELETE CASCADE，则删除用户时会自动删除关联文章
    # 如果不是，你需要手动删除或将外键设为 NULL (如果允许)
    # For this example, let's assume ON DELETE CASCADE is configured or we manually handle it.
    # Example manual deletion of posts before user deletion:
    # for post in user_to_delete.posts.all():
    #     db.session.delete(post)

    db.session.delete(user_to_delete)
    db.session.commit()
    return redirect(url_for('index'))
except Exception as e:
    db.session.rollback()
    # 根据实际情况处理外键约束冲突等错误
    return f"删除用户失败: {e}", 500 # 可能因为关联文章无法删除

删除文章

@app.route(‘/delete_post/‘)
def delete_post(post_id):
post_to_delete = Post.query.get_or_404(post_id)

try:
    db.session.delete(post_to_delete)
    db.session.commit()
    return redirect(url_for('index'))
except Exception as e:
    db.session.rollback()
    return f"删除文章失败: {e}", 500

— 应用启动 —

if name == ‘main‘:
# 在应用启动时创建数据库表（仅用于开发环境，生产环境请使用迁移工具）
with app.app_context():
# 可以在这里检查表是否存在，避免重复创建
# if not db.engine.dialect.has_table(db.engine, ‘user’): # 检查 user 表是否存在
db.create_all()
print(“数据库表已创建或已存在。”) # 启动时输出提示信息

app.run(debug=True)

“`

运行这个应用：

bash
export FLASK_APP=app.py
flask run

打开浏览器访问 http://127.0.0.1:5000/，你就可以通过简单的表单和链接进行用户和文章的 CRUD 操作了。

7. 常见问题与进阶

• 应用上下文与请求上下文： Flask-SQLAlchemy 的 db 实例和 db.session 都依赖于 Flask 的上下文。db.create_all() 等操作需要在应用上下文 (with app.app_context():) 中执行。在处理 Web 请求的路由函数中，请求上下文会自动激活，所以可以直接使用 db.session。
• 错误处理与回滚： 在执行 db.session.commit() 时，数据库可能会抛出异常（例如唯一约束冲突、外键约束失败等）。捕获这些异常并在必要时调用 db.session.rollback() 是非常重要的，以确保数据库状态的一致性。
• 数据库迁移： 对于生产环境的应用，模型的任何修改都需要通过数据库迁移工具（如 Flask-Migrate）来同步到数据库结构。这比 db.create_all() 更安全和可控。
• 性能优化： 对于大型应用，需要考虑查询性能优化，例如索引的创建、避免 N+1 查询问题（可以使用 joinedload, subqueryload 等加载策略）、合理使用 lazy 参数等。
• 关系的高级用法： SQLAlchemy 的关系功能非常强大，支持多对多关系、自引用关系、级联操作 (cascade)、被动删除 (passive_deletes) 等。
• 查询进阶： Query 对象提供了丰富的过滤、联接 (join)、分组 (group_by)、聚合 (func) 等操作。

8. 总结

通过本文，你已经了解了 Flask-SQLAlchemy 的基本概念、如何进行安装和配置、如何定义映射到数据库表的模型、如何创建数据库表，以及如何在 Flask 应用中进行基本的数据创建、读取、更新和删除操作。

Flask-SQLAlchemy 大大简化了在 Flask 中使用 SQLAlchemy 的过程，让你能够更加专注于业务逻辑而非底层的数据库细节。这是一个强大的工具，值得深入学习。

下一步学习方向：

• 深入研究 Flask-SQLAlchemy 文档。
• 学习如何使用 Flask-Migrate 进行数据库迁移。
• 探索 SQLAlchemy 高级的查询技巧和性能优化方法。
• 学习如何在生产环境中部署带有数据库的 Flask 应用。

希望这篇详细的文章能帮助你快速上手 Flask-SQLAlchemy，为你的 Flask 应用插上数据存储的翅膀！祝你编程愉快！

---