在 Flask 中，最主流的数据库操作方式是使用 ORM（对象关系映射）。它将数据库中的“表”映射为 Python 的“类”，将“行”映射为“对象”。这样你就可以用 Python 代码而不是 SQL 语句来操作数据库。
业界标准是 Flask-SQLAlchemy，它是 SQLAlchemy 的 Flask 扩展。
🛠️ 1. 安装与配置
pip install flask-sqlalchemy基础配置 (app/__init__.py 或 config.py)
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
# 配置数据库 URI (这里以 SQLite 为例，生产环境常用 MySQL 或 PostgreSQL)
# SQLite: sqlite:///database.db
# MySQL: mysql+pymysql://user:password@localhost/db_name
# PostgreSQL: postgresql://user:password@localhost/db_name
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False # 关闭不必要的信号追踪，节省内存
# 初始化数据库实例
db = SQLAlchemy(app)🏗️ 2. 定义模型 (Model)
模型是一个 Python 类，继承自 db.Model。类的属性对应数据库的列。
# models.py
from app import db
from datetime import datetime
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)
password_hash = db.Column(db.String(256))
created_at = db.Column(db.DateTime, default=datetime.utcnow)
# 建立一对多关系：一个用户可以有多篇文章
# backref='author' 允许通过 Post.author 访问用户对象
posts = db.relationship('Post', backref='author', lazy=True)
def __repr__(self):
return f'<User {self.username}>'
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)
date_posted = db.Column(db.DateTime, default=datetime.utcnow)
# 外键：关联到 users 表的 id
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
def __repr__(self):
return f'<Post {self.title}>'常用字段类型：
db.Integer: 整数db.String(n): 变长字符串db.Text: 长文本db.Float: 浮点数db.Boolean: 布尔值db.DateTime: 日期时间
常用约束：
primary_key=True: 主键unique=True: 唯一索引nullable=False: 不允许为空default=value: 默认值
🔄 3. 数据库迁移 (Flask-Migrate)
千万不要手动修改数据库结构！ 当你的 Model 发生变化时（如添加新字段），需要使用迁移工具自动更新数据库表结构。
安装
pip install flask-migrate初始化
在 app/__init__.py 中集成：
from flask_migrate import Migrate
migrate = Migrate(app, db)常用命令
在项目根目录下执行：
- 初始化迁移文件夹（只需执行一次）：
flask db init- 生成迁移脚本（每次修改 Model 后执行）：
flask db migrate -m "添加用户邮箱字段"- 应用迁移到数据库：
flask db upgrade- 回滚迁移（如果出错了）：
flask db downgrade💻 4. CRUD 操作实战
A. 创建 (Create)
from app import db
from app.models import User, Post
# 创建新用户
new_user = User(username='john', email='john@example.com')
db.session.add(new_user)
db.session.commit() # 必须提交才能写入数据库
# 创建文章并关联用户
new_post = Post(title='Hello Flask', content='This is my first post.', author=new_user)
db.session.add(new_post)
db.session.commit()B. 查询 (Read)
SQLAlchemy 提供了强大的查询接口。
# 1. 查询所有用户
all_users = User.query.all()
# 2. 根据主键查询
user = User.query.get(1)
# 3. 过滤查询 (Filter)
# SELECT * FROM users WHERE username = 'john'
user_john = User.query.filter_by(username='john').first()
# 复杂过滤
# SELECT * FROM users WHERE username LIKE '%j%'
users_like_j = User.query.filter(User.username.like('%j%')).all()
# 4. 排序
posts = Post.query.order_by(Post.date_posted.desc()).all()
# 5. 分页 (Web 开发最常用)
# 获取第 1 页，每页 10 条
page_obj = Post.query.paginate(page=1, per_page=10)
posts_on_page = page_obj.items
total_pages = page_obj.pagesC. 更新 (Update)
user = User.query.get(1)
user.email = 'new_email@example.com'
db.session.commit()D. 删除 (Delete)
user = User.query.get(1)
db.session.delete(user)
db.session.commit()🔗 5. 关系查询 (Relationships)
利用我们在 Model 中定义的 relationship，可以非常方便地进行关联查询。
# 1. 获取某个用户的所有文章
user = User.query.get(1)
for post in user.posts:
print(post.title)
# 2. 获取某篇文章的作者
post = Post.query.get(1)
print(post.author.username)
# 3. 联合查询 (Join)
# 查询所有标题包含 "Flask" 的文章及其作者名
results = db.session.query(Post.title, User.username)\
.join(User, Post.user_id == User.id)\
.filter(Post.title.like('%Flask%'))\
.all()⚠️ 常见陷阱与最佳实践
- 记得 commit：所有的增删改操作最后都必须调用
db.session.commit()，否则数据不会持久化。 - 事务回滚：如果发生异常，建议回滚事务以保持数据一致性。
try:
db.session.add(some_object)
db.session.commit()
except Exception as e:
db.session.rollback()
raise e- N+1 查询问题：在循环中访问关联对象会导致大量数据库查询。
错误示例：
posts = Post.query.all()
for post in posts:
print(post.author.username) # 每次循环都查一次数据库！优化：使用 joinedload 预加载。
from sqlalchemy.orm import joinedload
posts = Post.query.options(joinedload(Post.author)).all()- 不要在生产环境使用 SQLite：SQLite 是文件型数据库，并发性能差。生产环境请使用 PostgreSQL 或 MySQL。
📝 总结流程图
定义 Model (Python 类)
↓
flask db migrate (生成迁移脚本)
↓
flask db upgrade (更新数据库表结构)
↓
视图函数中:
db.session.add(obj) -> 新增
Model.query.filter(...) -> 查询
obj.field = value -> 修改
db.session.delete(obj) -> 删除
↓
db.session.commit() (提交事务)掌握了数据库操作，你的 Flask 应用就拥有了“记忆”。接下来，你想学习如何结合 Flask-Login 实现完整的用户认证系统，还是学习如何构建 RESTful API？