第139集 Flask数据库集成

1. 数据库集成概述

1.1 为什么需要数据库集成

在Web应用开发中，数据库是存储和管理数据的核心组件。Flask作为一个轻量级的Web框架，本身并不包含数据库支持，但它提供了灵活的扩展机制，允许我们轻松地集成各种数据库系统。

1.2 Flask中常用的数据库解决方案

Flask支持多种数据库集成方式：

1. 原生数据库API：直接使用Python的数据库API（如sqlite3、psycopg2等）
2. ORM框架：使用对象关系映射（ORM）框架，如SQLAlchemy
3. Flask扩展：使用专门为Flask设计的数据库扩展，如Flask-SQLAlchemy、Flask-MySQLdb等

在实际开发中，我们通常使用Flask-SQLAlchemy扩展，它提供了强大的ORM功能，使数据库操作更加简洁和安全。

2. Flask-SQLAlchemy扩展

2.1 Flask-SQLAlchemy简介

Flask-SQLAlchemy是Flask的一个扩展，它集成了SQLAlchemy ORM框架，为Flask应用提供了便捷的数据库访问接口。SQLAlchemy是Python中最流行的ORM框架之一，它允许我们使用Python对象来表示数据库表，而不需要直接编写SQL语句。

2.2 安装Flask-SQLAlchemy

pip install flask-sqlalchemy

3. 数据库连接配置

3.1 配置数据库URL

在Flask应用中，我们需要通过配置项来指定数据库连接信息。SQLAlchemy支持多种数据库后端，每种数据库有不同的URL格式：

3.2 初始化Flask-SQLAlchemy

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
# 配置数据库URL
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
# 禁用SQLAlchemy的事件通知系统，减少开销
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# 创建SQLAlchemy实例
db = SQLAlchemy(app)

4. 模型定义

4.1 什么是模型

在ORM中，模型是一个Python类，它映射到数据库中的一个表。模型类的每个属性对应表中的一个列。

4.2 定义模型

我们可以通过继承db.Model来定义数据库模型：

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 = db.Column(db.String(120), nullable=False)
    created_at = db.Column(db.DateTime, default=db.func.current_timestamp())
    
    def __repr__(self):
        return f'<User {self.username}>'

4.3 常用的数据类型

SQLAlchemy提供了多种数据类型，用于定义表的列：

4.4 常用的列选项

定义列时，我们可以使用以下常用选项：

5. 数据库操作

5.1 创建数据库和表

在定义好模型后，我们需要创建数据库和表：

# 创建所有表
db.create_all()

5.2 增加数据

# 创建一个新用户
new_user = User(username='admin', email='admin@example.com', password='password123')

# 将用户添加到会话
db.session.add(new_user)

# 提交会话，将更改保存到数据库
db.session.commit()

# 批量添加多个用户
users = [
    User(username='user1', email='user1@example.com', password='password1'),
    User(username='user2', email='user2@example.com', password='password2')
]
db.session.add_all(users)
db.session.commit()

5.3 查询数据

# 查询所有用户
all_users = User.query.all()

# 查询第一个用户
first_user = User.query.first()

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

# 条件查询
admin_user = User.query.filter_by(username='admin').first()

# 更复杂的条件查询
users_with_gmail = User.query.filter(User.email.like('%gmail.com')).all()

# 排序查询
users_ordered = User.query.order_by(User.username.asc()).all()

# 限制查询结果数量
limited_users = User.query.limit(5).all()

# 分页查询
page = 1
per_page = 10
pagination = User.query.paginate(page=page, per_page=per_page)
users = pagination.items

5.4 更新数据

# 先查询用户
user = User.query.filter_by(username='admin').first()

# 更新用户信息
user.email = 'new_admin@example.com'

# 提交更改
db.session.commit()

5.5 删除数据

# 先查询用户
user = User.query.filter_by(username='admin').first()

# 删除用户
db.session.delete(user)

# 提交更改
db.session.commit()

6. 关系映射

6.1 关系类型

SQLAlchemy支持多种关系类型：

1. 一对一：一个模型的实例与另一个模型的实例一一对应
2. 一对多：一个模型的实例与另一个模型的多个实例对应
3. 多对多：多个模型的实例与多个另一个模型的实例对应

6.2 一对多关系

class Post(db.Model):
    __tablename__ = 'posts'
    
    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('users.id'), nullable=False)
    
    # 定义关系
    user = db.relationship('User', backref=db.backref('posts', lazy=True))
    
    def __repr__(self):
        return f'<Post {self.title}>'

在这个例子中，User和Post之间是一对多关系：一个用户可以有多个帖子，一个帖子只能属于一个用户。

• db.ForeignKey('users.id')：定义外键，引用users表的id列
• db.relationship('User', backref=db.backref('posts', lazy=True))：定义关系，backref参数会在User模型中添加一个posts属性，用于访问该用户的所有帖子

6.3 多对多关系

多对多关系需要一个中间表来维护两个表之间的关系：

# 定义中间表
followers = db.Table('followers',
    db.Column('follower_id', db.Integer, db.ForeignKey('users.id'), primary_key=True),
    db.Column('followed_id', db.Integer, db.ForeignKey('users.id'), primary_key=True)
)

class User(db.Model):
    # ... 其他字段 ...
    
    # 定义多对多关系
    followed = db.relationship(
        'User', secondary=followers,
        primaryjoin=(followers.c.follower_id == id),
        secondaryjoin=(followers.c.followed_id == id),
        backref=db.backref('followers', lazy='dynamic'), lazy='dynamic'
    )

在这个例子中，User模型之间是多对多关系：一个用户可以关注多个用户，也可以被多个用户关注。

• secondary参数指定中间表
• primaryjoin参数指定主表与中间表的连接条件
• secondaryjoin参数指定从表与中间表的连接条件

7. 数据库迁移

7.1 为什么需要数据库迁移

在开发过程中，我们经常需要修改数据库模型（如添加新字段、修改字段类型等）。直接删除并重新创建表会导致数据丢失，因此我们需要使用数据库迁移工具来管理数据库模式的变化。

7.2 使用Flask-Migrate

Flask-Migrate是Flask的一个扩展，它集成了Alembic迁移工具，为Flask-SQLAlchemy应用提供了数据库迁移支持。

7.2.1 安装Flask-Migrate

pip install flask-migrate

7.2.2 初始化Flask-Migrate

from flask_migrate import Migrate

# 初始化迁移工具
migrate = Migrate(app, db)

7.2.3 创建迁移仓库

flask db init

7.2.4 创建迁移脚本

flask db migrate -m "Initial migration"

7.2.5 应用迁移

flask db upgrade

7.2.6 回滚迁移

flask db downgrade

8. 事务管理

8.1 什么是事务

事务是数据库操作的一个原子单位，它要么全部成功，要么全部失败。事务具有ACID特性：

• 原子性（Atomicity）：事务中的所有操作要么全部完成，要么全部不完成
• 一致性（Consistency）：事务完成后，数据库从一个一致性状态转换到另一个一致性状态
• 隔离性（Isolation）：多个事务并发执行时，它们之间不会互相影响
• 持久性（Durability）：事务完成后，对数据库的修改是永久的

8.2 在Flask-SQLAlchemy中使用事务

Flask-SQLAlchemy默认使用自动提交模式，当我们调用db.session.commit()时，会提交当前事务。如果发生错误，我们可以调用db.session.rollback()来回滚事务：

try:
    # 创建新用户
    new_user = User(username='test', email='test@example.com', password='password')
    db.session.add(new_user)
    # 提交事务
    db.session.commit()
except Exception as e:
    # 发生错误时回滚事务
    db.session.rollback()
    print(f"错误：{e}")

9. 最佳实践

9.1 数据库配置

1. 使用环境变量：将数据库连接信息存储在环境变量中，而不是硬编码在代码里
2. 分环境配置：为开发、测试和生产环境使用不同的数据库配置
3. 使用连接池：合理配置连接池大小，提高数据库访问性能

9.2 模型设计

1. 合理设计表结构：根据业务需求设计合理的表结构和关系
2. 使用适当的数据类型：为每个字段选择合适的数据类型
3. 添加索引：为经常用于查询的字段添加索引，提高查询性能
4. 避免过度设计：不要创建不必要的表和字段

9.3 查询优化

1. 减少查询次数：使用join操作减少N+1查询问题
2. 使用延迟加载：只在需要时加载关联数据
3. 限制查询结果：使用limit和offset限制查询结果数量
4. 使用原生SQL：对于复杂查询，可以使用原生SQL提高性能

9.4 安全性

1. 防止SQL注入：始终使用ORM提供的参数化查询，避免直接拼接SQL语句
2. 保护敏感数据：对密码等敏感数据进行加密存储
3. 限制数据库用户权限：为应用程序使用的数据库用户分配最小必要的权限
4. 定期备份：定期备份数据库，防止数据丢失

10. 完整示例：用户管理系统

10.1 应用结构

myapp/
├── app.py
├── models.py
├── templates/
│   ├── base.html
│   ├── users.html
│   └── add_user.html
└── config.py

10.2 代码实现

config.py：

class Config:
    SECRET_KEY = 'your-secret-key'
    SQLALCHEMY_DATABASE_URI = 'sqlite:///users.db'
    SQLALCHEMY_TRACK_MODIFICATIONS = False

models.py：

from flask_sqlalchemy import SQLAlchemy

db = 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)
    created_at = db.Column(db.DateTime, default=db.func.current_timestamp())
    
    def __repr__(self):
        return f'<User {self.username}>'

app.py：

from flask import Flask, render_template, request, redirect, url_for, flash
from models import db, User
from config import Config
from flask_migrate import Migrate

app = Flask(__name__)
app.config.from_object(Config)

# 初始化数据库
 db.init_app(app)

# 初始化迁移工具
migrate = Migrate(app, db)

# 创建表
with app.app_context():
    db.create_all()

@app.route('/')
def index():
    users = User.query.all()
    return render_template('users.html', users=users)

@app.route('/add_user', methods=['GET', 'POST'])
def add_user():
    if request.method == 'POST':
        username = request.form['username']
        email = request.form['email']
        
        # 检查用户名和邮箱是否已存在
        if User.query.filter_by(username=username).first():
            flash('用户名已存在', 'danger')
            return redirect(url_for('add_user'))
        
        if User.query.filter_by(email=email).first():
            flash('邮箱已存在', 'danger')
            return redirect(url_for('add_user'))
        
        # 创建新用户
        new_user = User(username=username, email=email)
        db.session.add(new_user)
        db.session.commit()
        
        flash('用户添加成功', 'success')
        return redirect(url_for('index'))
    
    return render_template('add_user.html')

@app.route('/delete_user/<int:user_id>')
def delete_user(user_id):
    user = User.query.get_or_404(user_id)
    db.session.delete(user)
    db.session.commit()
    
    flash('用户删除成功', 'success')
    return redirect(url_for('index'))

if __name__ == '__main__':
    app.run(debug=True)

templates/base.html：

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>{% block title %}用户管理系统{% endblock %}</title>
    <style>
        .flash {
            padding: 10px;
            margin: 10px 0;
            border-radius: 5px;
        }
        .success {
            background-color: #d4edda;
            color: #155724;
        }
        .danger {
            background-color: #f8d7da;
            color: #721c24;
        }
    </style>
</head>
<body>
    <div class="container">
        <h1>用户管理系统</h1>
        
        {% with messages = get_flashed_messages(with_categories=true) %}
            {% if messages %}
                {% for category, message in messages %}
                    <div class="flash {{ category }}">{{ message }}</div>
                {% endfor %}
            {% endif %}
        {% endwith %}
        
        {% block content %}
        {% endblock %}
    </div>
</body>
</html>

templates/users.html：

{% extends 'base.html' %}

{% block content %}
    <h2>用户列表</h2>
    <a href="{{ url_for('add_user') }}">添加用户</a>
    <table border="1" cellpadding="5" cellspacing="0">
        <tr>
            <th>ID</th>
            <th>用户名</th>
            <th>邮箱</th>
            <th>创建时间</th>
            <th>操作</th>
        </tr>
        {% for user in users %}
            <tr>
                <td>{{ user.id }}</td>
                <td>{{ user.username }}</td>
                <td>{{ user.email }}</td>
                <td>{{ user.created_at }}</td>
                <td>
                    <a href="{{ url_for('delete_user', user_id=user.id) }}">删除</a>
                </td>
            </tr>
        {% endfor %}
    </table>
{% endblock %}

templates/add_user.html：

{% extends 'base.html' %}

{% block content %}
    <h2>添加用户</h2>
    <form method="post">
        <div>
            <label for="username">用户名：</label>
            <input type="text" id="username" name="username" required>
        </div>
        <div>
            <label for="email">邮箱：</label>
            <input type="email" id="email" name="email" required>
        </div>
        <div>
            <input type="submit" value="添加">
        </div>
    </form>
    <a href="{{ url_for('index') }}">返回用户列表</a>
{% endblock %}

11. 总结

Flask数据库集成是Web应用开发的重要组成部分。通过Flask-SQLAlchemy扩展，我们可以轻松地使用SQLAlchemy ORM框架来操作数据库，使数据库操作更加简洁和安全。

在实际开发中，我们需要注意以下几点：

1. 合理设计数据库模型和关系
2. 使用事务管理保证数据一致性
3. 使用数据库迁移工具管理数据库模式变化
4. 优化查询性能，提高应用响应速度
5. 确保数据库操作的安全性

通过本集的学习，您已经掌握了Flask数据库集成的基本概念和使用方法，能够开发具有数据库功能的Web应用。