Session事务回滚、flask-migrate数据迁移

ゝ一纸荒年。 2022-12-19 00:52 84阅读 0赞

#### 1.session ####

这里所说的session可不是和request一起的那个session，而是flask-sqlalchemy 对于 sqlalchemy本体 的 Session 进行了一定的封装，它会在首次数据操作时自动创建结束时自动销毁，整个过程使用的是同一个线程隔离的session

Session中可以包含多个事务，并且事务失败后会自动回滚。但是，同一个请求中, 想要在前一个事务失败的情况下创建新的事务, 必须先手动回滚事务 Session.rollback

from flask import Flask
    from flask_sqlalchemy import SQLAlchemy
    
    app = Flask(__name__)
    app.config["SQLALCHEMY_DATABASE_URI"] = "mysql+pymysql://root:mysql@192.168.243.157:3306/test39"
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
    app.config["SQLALCHEMY_ECHO"] = True
    
    db = SQLAlchemy(app)
    
    class User(db.Model):
        __tablename__ = "t_user"
        id = db.Column(db.Integer, primary_key=True)
        # unique=True 唯一约束
        name = db.Column(db.String(64), unique=True)
    
    @app.route('/add')
    def add():
        # 事务1：重复添加数据 --异常（为了测试，已经在数据库中插入该条数据）
        user = User(name="pan")
        db.session.add(user)
        try:
            db.session.commit()
        except Exception as e:
            print(e)
            # 手动回滚
            db.session.rollback()  # 如果此处不回滚，则会影响下面的事务，可注释这里进行演示
    
        # 事务2：添加正常的数据
        user1 = User(name="jordan")
        db.session.add(user1)
        db.session.commit()
        # 总结：当事务1失败后，需要手动回滚，才不会影响下面的事务创建和执行
        return "add success"
    
    if __name__ == '__main__':
        app.run(debug=True)

#### 2.数据迁移 ####

flask没有像Django那样集成了迁移功能，如果想要实现迁移，可以使用`flask-migrate`组件，需要另外安装

pip install flask-migrate

from flask import Flask
    from flask_sqlalchemy import SQLAlchemy
    
    app = Flask(__name__)
    
    app.config["SQLALCHEMY_DATABASE_URI"] = "mysql://root:mysql@192.168.203.132:3306/python39"
    app.config["SQLALCHEMY_ECHO"] = True
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
    
    db = SQLAlchemy(app)
    
    Migrate(app, db)
    
    
    class User(db.Model):
        __tablename__ = "user39"
    
        id = db.Column(db.Integer, primary_key=True)
        # unique=True 唯一约束
        name = db.Column(db.String(64), unique=True)
        age = db.Column(db.Integer)
        introduction = db.Column(db.String(64))
    
    
    
    # 方案1：表不存在
    # 指定指定文件
    export FLASK_APP="demo10_migrate"
    # 初始化版本
    flask db init
    # 执行迁移，生成迁移的版本
    flask db migrate
    # 版本升级
    flask db upgrade
    # 查看历史版本
    flask db history
    # 切换到任何版本
    flask db downgrade 版本号
    
    # 方案2：数据库中表存在修改字段
    export FLASK_APP="demo10_migrate"
    flask db migrate
    flask db upgrade
    flask db history
    flask db downgrade
    
    
    @app.route("/")
    def index():
        return "index page"
    
    
    if __name__ == '__main__':
        app.run(debug=True)