Flask框架中SQLAlchemy的集成与数据迁移实践
在Flask中集成SQLAlchemy
SQLAlchemy作为Python生态中最成熟的ORM框架，可以通过flask-sqlalchemy扩展包无缝接入Flask应用。该扩展封装了SQLAlchemy的核心功能，提供了Flask风格的API调用方式。
环境准备与基础配置
首先通过pip安装依赖包：
pip install flask-sqlalchemy
接下来在应用初始化文件中完成配置：
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
# 配置数据库连接参数
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///blog.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 初始化SQLAlchemy实例
database = SQLAlchemy(app)
# 在模型定义中继承database.Model
class Post(database.Model):
id = database.Column(database.Integer, primary_key=True)
title = database.Column(database.String(100), nullable=False)
content = database.Column(database.Text)
created_at = database.Column(database.DateTime, default=datetime.datetime.now)
__tablename__ = 'posts'
常用操作接口
通过database对象可以访问会话管理和模型定义：
database.session- 获取数据库会话，用于增删改查操作database.Model- 所有模型类的基类database.Column- 定义数据表字段
使用Migrate进行数据库迁移
直接使用SQLAlchemy的create_all方法只能创建新表，无法处理表结构变更。flask-migrate扩展提供了命令行迁移工具，支持版本管理和增量更新。
依赖版本推荐：Flask 2.2.2、flask-script 2.0.6、flask-migrate 4.0.5
pip install flask-script flask-migrate
在应用入口文件中配置迁移命令：
from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
# 创建管理器实例
manager = Manager(app)
# 初始化迁移功能
Migrate(app, database)
manager.add_command('db', MigrateCommand)
if __name__ == '__main__':
manager.run()
迁移命令说明
执行以下命令完成数据库同步：
python manage.py db init # 初始化迁移环境
python manage.py db migrate # 检测模型变更并生成迁移脚本
python manage.py db upgrade # 执行迁移脚本更新数据库结构
完整项目示例
以下是一个包含完整配置的示例项目结构：
项目启动文件（run.py）
from flask import Flask
from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
from app.models import database
app = Flask(__name__)
app.config.from_object('config.ProductionConfig')
database.init_app(app)
manager = Manager(app)
Migrate(app, database)
manager.add_command('migrate', MigrateCommand)
if __name__ == '__main__':
manager.run()
数据模型定义（app/models.py）
from flask_sqlalchemy import SQLAlchemy
import datetime
database = SQLAlchemy()
class Author(database.Model):
"""作者信息表"""
id = database.Column(database.Integer, primary_key=True)
username = database.Column(database.String(32), unique=True, nullable=False)
email = database.Column(database.String(64))
joined_date = database.Column(database.DateTime, default=datetime.datetime.now)
__tablename__ = 'authors'
# 定义复合约束
__table_args__ = (
database.UniqueConstraint('username', 'email', name='uix_name_email'),
database.Index('ix_username_email', 'username', 'email'),
)
class Article(database.Model):
"""文章表"""
id = database.Column(database.Integer, primary_key=True)
author_id = database.Column(database.Integer, database.ForeignKey('authors.id'))
title = database.Column(database.String(128), nullable=False)
body = database.Column(database.Text)
publish_time = database.Column(database.DateTime)
__tablename__ = 'articles'
配置文件（config.py）
class BaseConfig:
"""基础配置"""
SQLALCHEMY_DATABASE_URI = 'postgresql://admin:password@localhost:5432/blogdb'
SQLALCHEMY_POOL_SIZE = 10
SQLALCHEMY_POOL_TIMEOUT = 20
SQLALCHEMY_TRACK_MODIFICATIONS = False
class DevelopmentConfig(BaseConfig):
"""开发环境配置"""
DEBUG = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///dev.db'
class ProductionConfig(BaseConfig):
"""生产环境配置"""
DEBUG = False
SQLALCHEMY_POOL_SIZE = 20
使用示例
# 添加数据
new_author = Author(username='writer', email='writer@example.com')
database.session.add(new_author)
database.session.commit()
# 查询数据
authors = Author.query.filter_by(username='writer').all()
# 关联查询
articles = Article.query.join(Author).filter(Author.username == 'writer').all()
通过以上配置，即可在Flask应用中实现完整的数据库ORM操作和版本迁移管理。
