Flask数据库迁移实战指南
数据库迁移操作
1. 安装必要的依赖
pip install flask flask-sqlalchemy flask-migrate pymysql cryptography2. 创建项目结构
# myapp/ # ├── app.py # 主应用文件 # ├── models.py # 数据模型定义 # └── config.py # 配置文件
3. config.py (数据库配置)
class Config:# MySQL 数据库连接配置 - 请替换为你的实际数据库信息SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://username:password@localhost:3306/flask_db'SQLALCHEMY_TRACK_MODIFICATIONS = False # 避免警告信息MYSQL_DEFAULT_CHARSET = 'utf8mb4' # 支持完整UnicodeMYSQL_DEFAULT_COLLATION = 'utf8mb4_unicode_ci' # 排序规则4. app.py (主应用文件)
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
class Config:# MySQL 数据库连接配置 - 请替换为你的实际数据库信息SQLALCHEMY_DATABASE_URI = 'mysql://root:123456@127.0.0.1:3306/youdb'SQLALCHEMY_TRACK_MODIFICATIONS = False # 避免警告信息
app = Flask(__name__)
app.config.from_object(Config) # 加载配置
# 初始化数据库扩展
db = SQLAlchemy(app)
migrate = Migrate(app, db) # 初始化迁移扩展
# 注意：导入模型必须在 db 初始化后
from models import User, Post5. models.py (数据模型定义)
from app import db
from datetime import datetime
class User(db.Model):id = db.Column(db.Integer, primary_key=True)username = db.Column(db.String(64), unique=True, nullable=False)email = db.Column(db.String(120), unique=True, nullable=False)password_hash = db.Column(db.String(128), nullable=False)created_at = db.Column(db.DateTime, default=datetime.utcnow)# 一对多关系: 一个用户有多篇文章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)content = db.Column(db.Text, nullable=False)created_at = db.Column(db.DateTime, default=datetime.utcnow)# 外键关联到用户表user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)def __repr__(self):return f'<Post {self.title}>'6. 初始化迁移环境
在终端里面运行
set FLASK_APP=app.py
flask db init #进行初始化
# 结果: 在当前工作目录中会创建 migrations 目录7. 创建初始迁移脚本 (检测模型变化)
flask db migrate -m "第一次操作"
# 结果:
# - 在 migrations/versions 生成迁移脚本文件 (如: 1a2b3c4d5e6f_initial_migration.py)
# - 检查生成的脚本文件是否准确反映了模型定义8. 应用迁移到 MySQL 数据库
# 步骤 5: 应用迁移到 MySQL 数据库
flask db upgrade
# 结果:
# 1. 创建 youdb 数据库(若不存在)
# 2. 创建 User 和 Post 表
# 3. 在 alembic_version 表中记录迁移版本9. 修改模型 (添加新字段)
# 文件: models.py (更新模型)
class User(db.Model):# ... 原有字段 ...# 新增字段is_admin = db.Column(db.Boolean, default=False) # 管理标识last_login = db.Column(db.DateTime) # 最后登录时间10. 生成新的迁移脚本
flask db migrate -m "第二次操作，添加了新的user字段"
# 生成的迁移脚本示例 (migrations/versions/xxx_add_fields.py):
"""
def upgrade():# ### commands auto generated by Alembic - please adjust! ###op.add_column('user', sa.Column('is_admin', sa.Boolean(), nullable=True))op.add_column('user', sa.Column('last_login', sa.DateTime(), nullable=True))# ### end Alembic commands #### 自定义操作: 设置默认值op.execute("UPDATE user SET is_admin = 0") # MySQL使用0表示False
"""
# 注意事项:
# 1. 检查生成的脚本是否准确
# 2. 布尔类型在MySQL中实际是TINYINT(1)
# 3. 可能需要手动添加默认值逻辑11. 应用新的迁移
flask db upgrade
# 结果:
# User 表新增 is_admin 和 last_login 列12. 回滚迁移操作
# 步骤 9: 回滚迁移 (如果需要)
# 查看迁移历史
flask db history
# 示例输出:
# 1a2b3c4d5e6f -> 7g8h9i0j1k2l (head), Add is_admin and last_login
# <base> -> 1a2b3c4d5e6f, Initial migration
# 回滚到上一个版本
flask db downgrade 1a2b3c4d5e6f # 使用版本ID
# 结果:
# 1. 删除 is_admin 和 last_login 列
# 2. 数据库回到之前的版本13. 生产环境部署迁移
# 最佳实践:
# 1. 保留所有迁移脚本文件
# 2. 部署新代码前执行:
flask db upgrade
# 3. 回滚操作:
flask db downgrade <target-revision>14. 常见问题解决方案
# 问题 1: 迁移失败 (外键约束)
# 解决方案: 在迁移脚本中添加依赖关系
down_revision = 'previous_version_id' # 必须正确设置前一个版本ID
# 问题 2: MySQL 字符集问题
# 解决方案: 在数据库URI中指定字符集
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://user:pass@localhost/db?charset=utf8mb4'
# 问题 3: 长字段索引问题
# MySQL 对于长字符串创建索引有限制，解决方案:
username = db.Column(db.String(191), unique=True) # 最大191字符可被索引