【flask】flush刷新机制、外键关联、relationship关系字段、join关联查询

心已赠人 2022-12-19 00:49 12阅读 0赞

#### 1.flush刷新 ####

当我们在`db.session.add()`数据的时候，实际上它并不会直接写进数据库，而是等待`db.session.commit()`命令，而另一个比较像的命令是`db.session.flush()`，它会先将会话的数据取出，底层执行SQL模拟生成数据，最后进行回滚

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
    
    # 开启代码自动提交功能 [不推荐开启]
    # app.config["SQLALCHEMY_COMMIT_ON_TEARDOWN"] = True
    
    db = SQLAlchemy(app)
    
    class Goods(db.Model):
        __tablename__ = "goods"
        id = db.Column(db.Integer, primary_key=True)
        name = db.Column(db.String(64))
        count = db.Column(db.Integer, default=0)
    
    @app.route('/add')
    def add():
        goods = Goods(name="奔驰C260L", count=99)
        db.session.add(goods)
        print(goods.id)
        # 手动刷新
        # 会将会话层的数据取出，底层执行sql语句模拟生成数据，但是最后开启的事务的回滚
        db.session.flush()
        # 刷新后
        print("刷新后", goods.id)
    
        # commit提交的底层也会执行flush()操作
        db.session.commit()
    
        return "add success"
    
    if __name__ == '__main__':
        app.run(debug=True)

#### 2.外键关联 ####

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)
    
    class User(db.Model):
        __tablename__ = "table_user"
        id = db.Column(db.Integer, primary_key=True)
        name = db.Column(db.String(64), unique=True)
    
    class Address(db.Model):
        __tablename__ = "table_address"
        id = db.Column(db.Integer, primary_key=True)
        detail = db.Column(db.String(64), unique=True)
        user_id = db.Column(db.Integer)  # db.ForeignKey 设置外键约束，保证数据的完整性和一致性
        
        def __repr__(self):
            return "地址：{}".format(self.detail)
    
    @app.route('/add')
    def add():
        # 需求：外键关联字段，添加数据 [一起成功或者一起失败]
        # 保证数据的完整和一致性，将用户数据和地址数据一并提交
        user = User(name="pan")
        db.session.add(user)
        # TODO:此时user.id还是空数据，需要手动刷新
        db.session.flush()
        addr1 = Address(detail="广西省贺州市", user_id=user.id)
        addr2 = Address(detail="广东省广州市", user_id=user.id)
        db.session.add_all([addr1, addr2])
        db.session.commit()
        return "add success"
    
    @app.route('/query')
    def query():
        user = User.query.filter(User.name == "pan").first()
        addr_list = Address.query.filter(Address.user_id == user.id).all()
        print(addr_list)
        return "query success"
    
    if __name__ == '__main__':
        # db.drop_all()
        # db.create_all()
        app.run(debug=True)

#### 3.relationship关系字段 ####

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)
    
    class User(db.Model):
        __tablename__ = "table_user"
        id = db.Column(db.Integer, primary_key=True)
        name = db.Column(db.String(64), unique=True)
        # 参数1：对应字段的模型类名称
        # 参数2：backref="user" 反向引用字段
        addrs = db.relationship("Address", backref="user")
    
    
    class Address(db.Model):
        __tablename__ = "table_address"
        id = db.Column(db.Integer, primary_key=True)
        detail = db.Column(db.String(64), unique=True)
        # 外键字段
        user_id = db.Column(db.Integer, db.ForeignKey("table_user.id"))
        # user_id = db.Column(db.Integer, db.ForeignKey(User.id))
    
        # user = db.relationship("User") # 如果不在User类指定backref="user"，可以使用relationship()
    
    
    @app.route('/add')
    def add():
        user = User(name="xiaoming")
        db.session.add(user)
        # TODO:此时user.id还是空数据，需要手动刷新
        db.session.flush()
        addr1 = Address(detail="深圳人民公园", user_id=user.id)
        addr2 = Address(detail="深圳北", user_id=user.id)
        db.session.add_all([addr1, addr2])
        db.session.commit()
        return "add success"
    
    @app.route('/query')
    def query():
        user = User.query.filter(User.name == "xiaoming").first()  # type:User
        print(user.addrs)
        addr = Address.query.filter(Address.detail == "深圳北").first()
        print(addr.user)
        return "query success"
    
    
    if __name__ == '__main__':
        # db.drop_all()
        # db.create_all()
        app.run(debug=True)

**注意：** 关系字段在数据库中并不存在，仅仅是在面向对象的层面方便查询的字段

#### 4.join关联查询 ####

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__ = "table_user"
        id = db.Column(db.Integer, primary_key=True)
        name = db.Column(db.String(64), unique=True)
    
    class Address(db.Model):
        __tablename__ = "table_address"
        id = db.Column(db.Integer, primary_key=True)
        detail = db.Column(db.String(64), unique=True)
        user_id = db.Column(db.Integer, db.ForeignKey("table_user.id"))
    
    @app.route('/query')
    def query():
        # join多表关联查询
        # 查询xiaoming的id和地址名称
        # 注意：查询结果是一个列表嵌套元组，应用:db.session.query
        # query(查询的字段)
        # join(从表类名称, 联合条件)
        # filter(查询条件)
        ret = db.session.query(User.id, Address.detail).join(Address, User.id == Address.user_id).filter(User.name == "xiaoming").all()
    
        for item in ret:
            print(item.id)
            print(item.detail)
    
        return "query success"
    
    if __name__ == '__main__':
        app.run(debug=True)