flask使用sqlalchemy

flask-sqlalchemy集成方案

借助第三方模块flask-sqlalchemy，可以将sqlalchemy快速的集成到flask项目中。

pip install flask_sqlalchemy


from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()  # 产生一个SQLAlchemy对象

# 将db注册到app中
db.init_app(app)

# 视图函数中使用会话session，直接导入db即是线程安全的
db.session

# models.py中继承的declarative_base对象，被单例到db.Model中
class 表模型(db.Model):...

# 各种字段类型被集成到db中
db.Column(db.String(80), unique=True, nullable=False)
db.Column(db.Integer, db.ForeignKey("team.id"))

# 引擎的配置也可以直接在app中直接加，那么我们就不用自己创建引擎了，上述的session等也就基于此引擎生成
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://[email protected]:3306/ddd?charset=utf8"
SQLALCHEMY_POOL_SIZE = 5
SQLALCHEMY_POOL_TIMEOUT = 30
SQLALCHEMY_POOL_RECYCLE = -1
# 追踪对象的修改并且发送信号
SQLALCHEMY_TRACK_MODIFICATIONS = False

flask-migrate迁移表

我们通过sqlalchemy模块的declarative_base对象.metadata.create_all(engine)等代码方法，才能将表同步到数据库，并且也不支持修改表的字段，flask-migrate模块支持我们通过脚本命令的形式来同步表设计。

而这个模块依赖 flask-script 模块，以下是一种版本兼容的版本设置

flask2.2.2、flask-script2.0.3、flask-migrate==2.7.0

# 安装好相关依赖后，需要在启动文件中：
from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
manager = Manager(app)
Migrate(app, db)
manager.add_command('db', MigrateCommand) # 使用db相关指令迁移表

manager.run() # 使用python manage.py runserver 启动项目

迁移指令：

python manage.py db init   # 生成迁移文件夹，只需要建立项目时执行一次即可
python manage.py db migrate   # 记录表模型的变化
python manage.py db upgrade  # 将迁移记录同步到数据库

结合以上的flask项目

我们如果想整一个按命令启动的，可以迁移数据库的flask项目，可以参考以下的文件写法来拓展：

### manage.py
from flask import Flask
from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
from models import db

app = Flask(__name__)
app.config.from_object('settings.DevSettings')  # 初始数据库引擎前先导入配置
db.init_app(app)  # 按app将db初始化，创建引擎

manager = Manager(app)
Migrate(app, db)
manager.add_command('db', MigrateCommand)

if __name__ == '__main__':
    manager.run()

    
### models.py
from flask_sqlalchemy import SQLAlchemy
import datetime

db = SQLAlchemy()  # 产生一个db管理表模型


class User(db.Model):
    # 第四步：写字段
    id = db.Column(db.Integer, primary_key=True)  # 生成一列，类型是Integer，主键
    name = db.Column(db.String(32), index=True, nullable=False)  # name列varchar32，索引，不可为空
    email = db.Column(db.String(32), unique=True)
    # datetime.datetime.now不能加括号，加了括号，以后永远是当前时间
    ctime = db.Column(db.DateTime, default=datetime.datetime.now)
    # extra = Column(Text, nullable=True)

    # 第五步：写表名 如果不写以类名为表名
    __tablename__ = 'users'  # 数据库表名称

    # 第六步：建立联合索引，联合唯一
    __table_args__ = (
        db.UniqueConstraint('id', 'name', name='uix_id_name'),  # 联合唯一
        db.Index('ix_id_name', 'name', 'email'),  # 索引
    )


class Book(db.Model):
    __tablename__ = 'books'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(32))


### settings.py
class SettingBase:
    SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:[email protected]:3306/ddd?charset=utf8"
    SQLALCHEMY_POOL_SIZE = 5
    SQLALCHEMY_POOL_TIMEOUT = 30
    SQLALCHEMY_POOL_RECYCLE = -1
    # 追踪对象的修改并且发送信号
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    Debug = True


class DevSettings(SettingBase):
    Debug = False