Flask 数据库集成完全指南：Flask-SQLAlchemy 实践

1. 引言

数据库是现代Web应用的核心组件，Flask通过Flask-SQLAlchemy扩展提供了强大的数据库集成能力。本文将全面介绍如何在Flask应用中使用Flask-SQLAlchemy进行数据库操作，涵盖从基础配置到高级查询模式的各个方面。
在这里插入图片描述

2. Flask-SQLAlchemy基础配置

2.1 安装与初始化

pip install flask-sqlalchemy

from flask import Flask
from flask_sqlalchemy import SQLAlchemyapp = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

关键配置：

SQLALCHEMY_DATABASE_URI：数据库连接字符串
- SQLite: sqlite:///database.db
- PostgreSQL: postgresql://user:password@localhost/mydatabase
- MySQL: mysql://user:password@localhost/mydatabase
SQLALCHEMY_TRACK_MODIFICATIONS：禁用事件系统以提升性能

2.2 应用工厂模式集成

from flask_sqlalchemy import SQLAlchemydb = SQLAlchemy()def create_app():app = Flask(__name__)app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db'db.init_app(app)return app

3. 定义数据模型

3.1 基础模型定义

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)def __repr__(self):return f'<User {self.username}>'

常用字段类型：

db.Integer：整数
db.String(size)：字符串
db.Text：长文本
db.DateTime：日期时间
db.Boolean：布尔值
db.Float：浮点数

3.2 关系模型

class Post(db.Model):id = db.Column(db.Integer, primary_key=True)title = db.Column(db.String(100), nullable=False)content = db.Column(db.Text, nullable=False)user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)author = db.relationship('User', backref=db.backref('posts', lazy=True))class Comment(db.Model):id = db.Column(db.Integer, primary_key=True)text = db.Column(db.Text, nullable=False)post_id = db.Column(db.Integer, db.ForeignKey('post.id'))user_id = db.Column(db.Integer, db.ForeignKey('user.id'))post = db.relationship('Post', backref=db.backref('comments', lazy=True))user = db.relationship('User', backref=db.backref('comments', lazy=True))

关系类型：

db.ForeignKey：外键约束
db.relationship：定义模型间关系
backref：反向引用
lazy：加载策略(select, joined, dynamic等)

4. 数据库操作

4.1 创建数据库表

with app.app_context():db.create_all()

4.2 增删改查(CRUD)操作

创建记录

new_user = User(username='john', email='john@example.com')
db.session.add(new_user)
db.session.commit()

查询记录

# 获取所有用户
users = User.query.all()# 获取单个用户
user = User.query.get(1)# 条件查询
admin = User.query.filter_by(username='admin').first()# 复杂查询
recent_users = User.query.order_by(User.id.desc()).limit(5).all()

更新记录

user = User.query.get(1)
user.email = 'new@example.com'
db.session.commit()

删除记录

user = User.query.get(1)
db.session.delete(user)
db.session.commit()

5. 高级查询技巧

5.1 分页查询

page = request.args.get('page', 1, type=int)
per_page = 10
pagination = Post.query.paginate(page=page, per_page=per_page)
posts = pagination.items

分页对象属性：

items：当前页记录
page：当前页码
per_page：每页记录数
total：总记录数
pages：总页数

5.2 聚合查询

from sqlalchemy import func# 计数
user_count = db.session.query(func.count(User.id)).scalar()# 分组统计
post_counts = db.session.query(User.username,func.count(Post.id)
).join(Post).group_by(User.id).all()

5.3 复杂过滤

from sqlalchemy import or_# 多条件查询
search = "flask"
posts = Post.query.filter(or_(Post.title.contains(search),Post.content.contains(search))
).all()

6. 数据库迁移

6.1 Flask-Migrate配置

pip install flask-migrate

from flask_migrate import Migrateapp = Flask(__name__)
# ...其他配置...
db = SQLAlchemy(app)
migrate = Migrate(app, db)

6.2 迁移命令

# 初始化迁移仓库
flask db init# 生成迁移脚本
flask db migrate -m "initial migration"# 应用迁移
flask db upgrade# 回滚迁移
flask db downgrade

7. 性能优化

7.1 连接池配置

app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {'pool_size': 10,'pool_recycle': 300,'pool_pre_ping': True
}

7.2 批量操作

# 批量插入
users = [User(username=f'user{i}') for i in range(100)]
db.session.bulk_save_objects(users)
db.session.commit()# 批量更新
User.query.filter(User.id > 10).update({'active': False})
db.session.commit()

7.3 延迟加载与预加载

# 延迟加载(默认)
posts = User.query.get(1).posts  # 额外查询# 预加载
users = User.query.options(db.joinedload(User.posts)).all()

8. 测试与调试

8.1 测试数据库配置

import pytest
from app import create_app, db as _db@pytest.fixture
def app():app = create_app({'TESTING': True, 'SQLALCHEMY_DATABASE_URI': 'sqlite:///:memory:'})with app.app_context():_db.create_all()yield app_db.drop_all()@pytest.fixture
def client(app):return app.test_client()

8.2 SQL日志调试

import logging
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

9. 安全最佳实践

9.1 防止SQL注入

# 安全 - 使用ORM
User.query.filter_by(username=request.form['username'])# 危险 - 直接拼接SQL
query = f"SELECT * FROM user WHERE username = '{request.form['username']}'"

9.2 敏感数据加密

from werkzeug.security import generate_password_hash, check_password_hashclass User(db.Model):# ...password_hash = db.Column(db.String(128))@propertydef password(self):raise AttributeError('password is not a readable attribute')@password.setterdef password(self, password):self.password_hash = generate_password_hash(password)def verify_password(self, password):return check_password_hash(self.password_hash, password)

10. 总结与最佳实践

10.1 关键要点总结

模型设计：
- 合理规划表结构与关系
- 使用合适的字段类型和约束
- 为常用查询添加索引
查询优化：
- 使用预加载减少查询次数
- 只查询需要的字段
- 合理使用缓存
事务管理：
- 保持事务短小
- 正确处理回滚
- 避免长时间持有事务

10.2 推荐项目结构

/myapp/models__init__.py     # 导出所有模型user.py         # 用户模型post.py         # 文章模型/migrations         # 迁移脚本/testsconftest.py     # 测试配置test_models.py  # 模型测试app.py              # 应用工厂config.py           # 配置