【flask】SQLAlchemy 的增删改查（CRUD）

逃离我推掉我的手 2022-12-19 00:41 2阅读 0赞

#### 1.增加数据 ####

增加数据的步骤很简单，只需要把模型类对象`db.session.add(模型类对象)`即可，不过如果没有开启自动提交的话，需要手动提交`db.session.commit()`

某些情况，我们需要获取未插入数据库的数据主键id，可以使用flush()

goods1 = Goods(name='小米10', count=1000)
    db.session.add(goods1)
    db.session.flush()
    print(goods1.id)
    db.session.commit()

**注意：** SQLAlchemy会自动创建一个隐式事务（implicit），需要手动提交，但是事务失败，它会自动回滚

#### 2.查询数据 ####

##### 2.1 准备数据 #####

我们先定义好用户模型类，并且添加数据

##### 2.2 查询执行器 #####

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>all()</td> 
   <td>返回一个元素为查询结果的列表</td> 
  </tr> 
  <tr> 
   <td>count()</td> 
   <td>返回查询结果的数量</td> 
  </tr> 
  <tr> 
   <td>first()</td> 
   <td>返回第一个结果或None</td> 
  </tr> 
  <tr> 
   <td>first_or_404()</td> 
   <td>返回第一个结果或404</td> 
  </tr> 
  <tr> 
   <td>get(主键)</td> 
   <td>返回主键对应的对象或None</td> 
  </tr> 
  <tr> 
   <td>get_or_404(主键)</td> 
   <td>返回主键对应的对象或None</td> 
  </tr> 
  <tr> 
   <td>paginate(页码, 每页条数)</td> 
   <td>返回一个分页查询结果</td> 
  </tr> 
 </tbody> 
</table>

##### 2.3 查询过滤器 #####

<table> 
 <thead> 
  <tr> 
   <th>过滤器</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>filter_by(字段名=值)</td> 
   <td>等值查询</td> 
  </tr> 
  <tr> 
   <td>filter(函数引用/比较运算)</td> 
   <td>函数或比较查询</td> 
  </tr> 
  <tr> 
   <td>limit(限定条数)</td> 
   <td>限定返回结果数量</td> 
  </tr> 
  <tr> 
   <td>offset(偏移条数)</td> 
   <td>对查询结果进行偏移</td> 
  </tr> 
  <tr> 
   <td>order_by(排序字段)</td> 
   <td>对查询结果进行排序</td> 
  </tr> 
  <tr> 
   <td>options()</td> 
   <td>针对原查询限定查询的字段</td> 
  </tr> 
 </tbody> 
</table>

提示：以上查询得到的都是一个BaseQuery对象

##### 2.4 查询实例 #####

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__ = "users"  # 表名 默认使用类名的小写
       ...
    @app.route("/add")
    def add():
        ...
        return "add success"
    
    @app.route("/query")
    def query():
        # 简单查询
        User.query.all()  # 查询所有用户数据
        User.query.count()  # 查询有多少个用户
        User.query.first()  # 查询第1个用户
    
        # 根据id查询 返回模型对象/None
        User.query.get(5)
        User.query.filter_by(id=5).all()
        User.query.filter(User.id == 5).first()
    
        # 查询名字以某个字符开始/结尾/包含的所有用户
        User.query.filter(User.name.endswith("g")).all()
        User.query.filter(User.name.startswith("w")).all()
        User.query.filter(User.name.contains("n")).all()
        User.query.filter(User.name.like("w%n%g")).all()  # 模糊查询
    
        # 查询名字不等于wang的所有用户
        from sqlalchemy import not_, and_, or_
        User.query.filter(not_(User.name == "wang")).all()
        User.query.filter(User.name != "wang").all()
    
        User.query.filter(User.id.in_([1, 5, 6])).all()  # 查询id在某个范围的用户
        # 排序查询
        User.query.order_by(User.age, User.id.desc()).limit(5).all()  # 所有用户先按年龄从小到大, 再按id从大到小排序, 取前5个
        # 分页查询
        pn = User.query.paginate(3, 3)
        # pn.pages 总页数 pn.page 当前页码 pn.items 当前页的数据 pn.total 总条数
    
        # 聚合查询
        # 查询每个年龄的人数 select age, count(name) from t_user group by age 分组聚合
        from sqlalchemy import func
        data = db.session.query(User.age, func.count(User.id).label("count")).group_by(User.age).all()
        for item in data:
            # print(item[0], item[1])
            print(item.age, item.count)
    
        # 只查询所有人的姓名和邮箱 优化查询 User.query.all() # 相当于select *
        from sqlalchemy.orm import load_only
        data = User.query.options(load_only(User.name, User.email)).all()
        # 另一种写法： data = session.query(User).with_entities(User.id).filter(xxx).all()
        for item in data:
            print(item.name, item.email)
    
        data = db.session.query(User.name, User.email).all()
        for item in data:
            print(item.name, item.email)
    
        return "query success"
    
    if __name__ == '__main__':
        db.drop_all()  # 删除所有表，为方便测试，实际勿用
        db.create_all()  # 创建所有表
        app.run(debug=True)

#### 3.更新数据 ####

#### 4.删除数据 ####