（三）SQLAlchemy进行增删改查等操作

迈不过友情╰ 2022-09-07 12:29 270阅读 0赞

### 目录 ###

*   *  一：增加操作
     *  二：更新操作：
     *  三：删除操作：
     *  四： 简单查询操作
     *   *  4.1： Session.query进行查询：
         *   *  1： 条件查询：
             *  2： 查询单一数据返回：
             *  4： 条件组合：
             *  5： filter\_by进行查询：
             *  6： 连表查询：
         *  4.2： select()函数进行查询：
         *   *  1： 条件查询：
             *  2： 查询单一数据返回：
             *  3： 对列起别名：
             *  4: 条件组合：
             *  5： filter\_by进行查询：
             *  6： 连表查询：
     *  五： 复杂查询：
     *   *  1： 排序：
         *  2： 分组：
         *  3： 别名：
         *  4： 子查询：
         *   *  4.1: subquery
             *  4.2: 通用表表达式（CTE）：

*  在ORM中直接定义类对象就可以操作数据库，不再使用底层的Insert对象了。
 *  注意： query和exec中执行select语句都可以进行查询，两者的适用场景不同。
    
     *  对于复杂查询我们还是推荐使用select语句。
     *  对于简单的语句，我们推荐使用query语句。（写法简单）

## 一：增加操作 ##

*  1： 数据准备： 创建一个简单的User模型类
    
        from sqlalchemy.orm import registry
        from sqlalchemy import Column, Integer, VARCHAR
        
        from main import engine
        
        my_registry = registry()
        Base = my_registry.generate_base()
        
        
        class User(Base):
            __tablename__ = "users"
            id = Column(Integer, primary_key=True)
            name = Column(VARCHAR(255), nullable=False)
            age = Column(VARCHAR(255), nullable=False)
            sex = Column(Integer, nullable=True)
        
        
        my_registry.metadata.create_all(engine)
 *  2： 我们可以使用**类的对象来表示一行数据**，如果我们打印这个对象会发现，这个对象会补充默认的数据。
    
     *  本质： **模型类会映射出一个SQL语句**。
    
        from sqlalchemy import create_engine
        import settings
        from module import User
        
        engine = create_engine(settings.DB_URI, echo=True)
        
        user = User(name='梁山', age=23, sex=1)
        print(user.id, user.name, user.age, user.sex)
        # None 梁山 23 1
 *  3： 原来我们使用with上下文来管理资源，现在我们直接使用session来管理资源。
    
     *  基于引擎创建session: session = Session(engine)
     *  将对象添加进入session： session.add(user)
     *  查看所有已经添加到会话的对象(没有提交的)：session.new
     *  注意： **session会将对象添加到一个IdentitySet的字典中，这个字典会对所有添加的对象进行哈希处理，因此同一个session对象，不管你加入多少次user，最终只会提交到数据库一次**。
    
        engine = create_engine(settings.DB_URI, echo=True)
        
        user = User(name='梁山', age=23, sex=1)
        session = Session(engine)
        session.add(user)
        print(session.new)
        # IdentitySet([<module.User object at 0x7f8f59a87a90>])
    
        engine = create_engine(settings.DB_URI, echo=True)
        
        user = User(name='梁山', age=23, sex=1)
        session = Session(engine)
        # 添加两次，字典中只保存一次。
        session.add(user)
        session.add(user)
        print(session.new)
        # IdentitySet([<module.User object at 0x7f8f59a87a90>])
 *  4：理解Flush刷新：
    
     *  作用： 预提交
     *  特点： **在数据库中执行sql语句，返回结果，但是不会持久化到数据库。**
     *  对于插入来讲，**flush刷新会导致对象存在了主键值**。
    
        engine = create_engine(settings.DB_URI, echo=True)
        
        user = User(name='梁山', age=23, sex=1)
        session = Session(engine)
        session.add(user)
        session.flush()
    
        2021-08-23 11:20:52,657 INFO sqlalchemy.engine.Engine BEGIN (implicit)
        2021-08-23 11:20:52,658 INFO sqlalchemy.engine.Engine INSERT INTO users (name, age, sex) VALUES (%(name)s, %(age)s, %(sex)s) RETURNING users.id
        2021-08-23 11:20:52,658 INFO sqlalchemy.engine.Engine [generated in 0.00018s] {
              'name': '梁山', 'age': 23, 'sex': 1}
    
        user = User(name='梁山', age=23, sex=1)
        print(user.id)
        # None
        session = Session(engine)
        session.add(user)
        session.flush()
        print(user.id)
        # id
 *  5: 提交与关闭会话:
    
     *  作用： 持久化到数据库，关闭会话。
    
        engine = create_engine(settings.DB_URI, echo=True)
        
        user = User(name='梁山', age=23, sex=1)
        print(user.id)
        # None
        session = Session(engine)
        session.add(user)
        session.commit()
        session.close()
    
        2021-08-23 11:27:56,195 INFO sqlalchemy.engine.Engine INSERT INTO users (name, age, sex) VALUES (%(name)s, %(age)s, %(sex)s) RETURNING users.id
        2021-08-23 11:27:56,196 INFO sqlalchemy.engine.Engine [generated in 0.00025s] {
              'name': '梁山', 'age': 23, 'sex': 1}
        2021-08-23 11:27:56,254 INFO sqlalchemy.engine.Engine COMMIT

## 二：更新操作： ##

*  1： 普通的更新操作：
    
        engine = create_engine(settings.DB_URI, echo=True)
        
        # 1: 实例化session
        session = Session(engine)
        # 2:进行更新操作
        session.execute(update(User).where(User.name == "梁山").values(name="王老五"))
        # 3：提交与关闭
        session.commit()
        session.close()
 *  2：基于查询更新：
    
     *  先查询出这个对象，然后修改这个对象的属性，此时会话层数据和持久层数据发生不一致了。
     *  当session.commit()的时候，会话层数据持久化到持久层，就保持了一致。
    
        from sqlalchemy import create_engine
        from sqlalchemy.orm import Session
        import settings
        from module import User
        
        engine = create_engine(settings.DB_URI, echo=True)
        
        # 1: 实例化session
        session = Session(engine)
        # 2:先进行查询操作
        user = session.query(User).filter(User.name == "王老五").first()
        user.name = "梁山"
        # 3：提交与关闭
        session.commit()
        session.close()

## 三：删除操作： ##

*  1： 普通的删除操作：
    
        # 1: 实例化session
        session = Session(engine)
        # 2:先进行查询操作
        session.execute(delete(User).where(User.name == "梁山"))
        # 3：提交与关闭
        session.commit()
        session.close()
 *  2： 基于查询进行删除：
    
        # 1: 实例化session
        session = Session(engine)
        # 2:先进行查询操作
        user = session.query(User).filter(User.name == "梁山").first()
        session.delete(user)
        # 3：提交与关闭
        session.commit()
        session.close()

## 四： 简单查询操作 ##

### 4.1： Session.query进行查询： ###

#### 1： 条件查询： ####

*  注意：与select不同的是，使用query直接得到对象列表，遍历这个列表直接就是对象。

from sqlalchemy import create_engine
    from sqlalchemy.orm import Session
    import settings
    from module import User
    
    engine = create_engine(settings.DB_URI, echo=True)
    
    # 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    users = session.query(User).filter(User.name == "梁山").all()
    print(type(users))
    # <class 'list'>
    for user in users:
        print(user)
        # <module.User object at 0x7f14b893af28>
    session.close()

*  返回部分数据：

# 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user = session.query(User.name).first()
    print(type(user))
    # <class 'sqlalchemy.engine.row.Row'>
    print(user.name)
    # 梁山
    session.close()

#### 2： 查询单一数据返回： ####

*  query().first()

engine = create_engine(settings.DB_URI, echo=True)
    
    # 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user = session.query(User).filter(User.name == "梁山").first()
    print(user)
    # <module.User object at 0x7ff000f19198>
    session.close()

#### 4： 条件组合： ####

engine = create_engine(settings.DB_URI, echo=True)
    
    # 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user = session.query(User.name).filter(and_(User.name == "梁山", User.age == "23")).first()
    print(user.name)
    
    session.close()

#### 5： filter\_by进行查询： ####

*  注意： filter\_by子句中使用的是属性赋值的方式。

# 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user = session.query(User).filter_by(name="梁山", age="23").first()
    print(user.name)
    # 梁山
    session.close()

#### 6： 连表查询： ####

*  使用join进行连表查询： 注意join中写后面的那个表。

# 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user_and_address = session.query(User, Address.address).filter(User.name == "梁山").join(Address).all()
    print(type(user_and_address))
    # <class 'list'>
    print(user_and_address)
    # [(<module.User object at 0x7fec195447f0>, '山东德州'), (<module.User object at 0x7fec195447f0>, '山东济南'), (<module.User object at 0x7fec195447f0>, '山东青岛')]
    
    session.close()

*  全连接：

engine = create_engine(settings.DB_URI, echo=True)
    
    # 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user_and_address = session.query(User, Address.address).join(Address, full=True)
    for ua in user_and_address:
        print(ua)
        # (<module.User object at 0x7fa854b8b588>, '山东德州')
        # (<module.User object at 0x7fa854b8b588>, '山东济南')
        # (<module.User object at 0x7fa854b8b588>, '山东青岛')
        # (<module.User object at 0x7fa854b8b5f8>, '北京西直门')
        # (<module.User object at 0x7fa854b8b5f8>, '北京东直门')
        # (<module.User object at 0x7fa854b8b668>, None)
    session.close()

*  外连接：

### 4.2： select()函数进行查询： ###

#### 1： 条件查询： ####

*  注意： select语句返回的是result对象， 遍历result对象得到的是row对象，这个对象是个元组，需要使用下标访问元组内容，才能得到对象。

# 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    users = session.execute(select(User).where(User.name == "梁山"))
    print(type(users))
    # <class 'sqlalchemy.engine.result.ChunkedIteratorResult'>
    for user in users:
        print(type(user))
        # <class 'sqlalchemy.engine.row.Row'>
        print(type(user[0]))
        # <class 'module.User'>
        print(user[0].name)
    
    session.close()

*  返回部分数据而不是全部数据：

#### 2： 查询单一数据返回： ####

*  注意： 是对execute执行的结果再使用first()调用取其中一个。
 *  注意：此时拿到的也是Row对象。

engine = create_engine(settings.DB_URI, echo=True)
    
    # 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user = session.execute(select(User).where(User.name == "梁山")).first()
    print(type(user))
    # <class 'sqlalchemy.engine.row.Row'>
    print(user[0])
    # <module.User object at 0x7f5673bc4da0>
    session.close()

#### 3： 对列起别名： ####

*  注意： 使用label对列进行改造和标识。

# 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user = session.execute(select(("姓名：" + User.name).label("username")).where(User.name == "梁山")).first()
    print(type(user))
    # <class 'sqlalchemy.engine.row.Row'>
    print(user.username)
    # 姓名：梁山
    session.close()

#### 4: 条件组合： ####

*  and\_ 与or\_的使用：
    
     *  注意： 虽然数据库存储的age是int类型但是我们编写代码也要用字符串类型。

# 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user = session.execute(select(User).where(and_(User.name == "梁山", User.age == "23"))).first()
    print(user[0].name)
    # 梁山
    session.close()

#### 5： filter\_by进行查询： ####

*  注意： filter\_by使用的方式是属性赋值的方式。

engine = create_engine(settings.DB_URI, echo=True)
    
    # 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user = session.execute(select(User).filter_by(name="梁山", age="23")).first()
    print(user[0].name)
    # 梁山
    session.close()

#### 6： 连表查询： ####

*  join\_from()
 *  join()

数据准备：

class User(Base):
        __tablename__ = "users"
        id = Column(Integer, primary_key=True)
        name = Column(VARCHAR(255), nullable=False)
        age = Column(VARCHAR(255), nullable=False)
        sex = Column(Integer, nullable=True)
    
    
    class Address(Base):
        __tablename__ = "address"
        id = Column(Integer, primary_key=True)
        address = Column(VARCHAR(255), nullable=True)
        user_id = Column(Integer, ForeignKey("users.id"))

*  使用join\_from进行连表查询： 查询姓名为梁山的人的所有信息和详细地址信息。

# 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user_and_address = session.execute(select(User, Address.address)
                                       .filter(User.name == "梁山")
                                       .join_from(User, Address)
                                       ).all()
    for ua in user_and_address:
        print(ua[0].name, ua[1])
        # 梁山 山东德州
        # 梁山 山东济南
        # 梁山 山东青岛

*  使用join进行连表查询：
    
     *  区别： join()里面只用写后面的那个表，而join\_from需要两个表都要写。
     *  **join\_from显示的指定那个表是左表，那个是右表。而join是指定右表，自动推测左表**。

# 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user_and_address = session.execute(select(User, Address.address)
                                       .filter(User.name == "梁山")
                                       .join(Address)
                                       ).all()
    for ua in user_and_address:
        print(ua[0].name, ua[1])
        # 梁山 山东德州
        # 梁山 山东济南
        # 梁山 山东青岛

*  内连接，外连接（左连接）， 全连接区别：
    
     *  内连接： 两个表的公共部分
     *  外连接（左连接）：左表保留和公共部分保留。
     *  全连接： 笛卡尔乘积
 *  join和join\_from都是内连接。
 *  外连接和全连接的使用：
    
     *  外连接

engine = create_engine(settings.DB_URI, echo=True)
    
    # 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user_and_address = session.execute(select(User, Address.address)
                                       .join(Address, isouter=True)
                                       ).all()
    for ua in user_and_address:
        print(ua[0].name, ua[1])
        # 梁山 山东德州
        # 梁山 山东济南
        # 梁山 山东青岛
        # 雪域 北京西直门
        # 雪域 北京东直门
        # 孟获 None
    
    session.close()

# 全连接：
    engine = create_engine(settings.DB_URI, echo=True)
    
    # 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    user_and_address = session.execute(select(User, Address.address)
                                       .join(Address, full=True)
                                       ).all()
    for ua in user_and_address:
        print(ua[0].name, ua[1])
    
    session.close()

## 五： 复杂查询： ##

### 1： 排序： ###

*  **列.asc()正序， 列.sesc()逆序**。
 *  **asc(列名)， desc(列名)**(推荐使用)
 *  select语句进行排序：

# 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    users = session.execute(select(User.name, User.age).order_by(User.age.desc()))
    for row in users:
        print(row[0], row[1])
        # 孟获 44
        # 雪域 24
        # 梁山 23
    session.close()

*  query进行排序： 当然如果多个排序字段，后面直接追加就可以了。

engine = create_engine(settings.DB_URI, echo=True)
    
    # 1: 实例化session
    session = Session(engine)
    # 2:先进行查询操作
    users = session.query(User.name, User.age).order_by(User.age.desc()).all()
    print(users)
    # [('孟获', '44'), ('雪域', '24'), ('梁山', '23')]
    users2 = session.query(User.name, User.age).order_by(User.age.asc()).all()
    print(users2)
    # [('梁山', '23'), ('雪域', '24'), ('孟获', '44')]
    session.close()

### 2： 分组： ###

*  1： 使用select进行简单分组：
    
        engine = create_engine(settings.DB_URI, echo=True)
        
        # 1: 实例化session
        session = Session(engine)
        # 2:先进行查询操作
        users = session.execute(select(User.sex, func.count(User.id).label("数量")).group_by(User.sex))
        for row in users:
            print(row[0], row[1])
            # 0 1
            # 1 2
        
        session.close()
 *  2： 使用query进行简单分组：
    
        engine = create_engine(settings.DB_URI, echo=True)
        
        # 1: 实例化session
        session = Session(engine)
        # 2:先进行查询操作
        users = session.query(User.sex, func.count(User.id).label("性别")).group_by(User.sex).all()
        print(users)
        # [(0, 1), (1, 2)]
        
        session.close()
 *  3：分组后，对分组的数据进行条件过滤：
    
        engine = create_engine(settings.DB_URI, echo=True)
        
        # 1: 实例化session
        session = Session(engine)
        # 2:先进行查询操作
        users = session.query(User.sex, func.count(User.id).label("数量")).group_by(User.sex).having(
            func.count(User.id) > 1).all()
        
        for row in users:
            print(row[0], row[1])
            # 1 2
        
        session.close()
    
        engine = create_engine(settings.DB_URI, echo=True)
        
        # 1: 实例化session
        session = Session(engine)
        # 2:先进行查询操作
        users = session.execute(
            select(User.sex, func.count(User.id).label("数量")).group_by(User.sex).having(func.count(User.id) > 1))
        for row in users:
            print(row[0], row[1])
            # 1 2
 *  4： 分组完在组内排序 ：
    
        # 1: 实例化session
          session = Session(engine)
        # 2:先进行查询操作
          users = session.query(User.name, User.age).group_by("id").order_by(desc("age"))
        for row in users:
              print(row)
              # ('孟获', '44')
              # ('雪域', '24')
              # ('梁山', '23')
    
        # 1: 实例化session
        session = Session(engine)
        # 2:先进行查询操作
        users = session.execute(select(User.name, User.age).group_by("id").order_by(desc("age")))
        for row in users:
            print(row)
            # ('孟获', '44')
            # ('雪域', '24')
            # ('梁山', '23')

### 3： 别名： ###

*  场景： 子关联查询：
 *  需求： where子句中需要同一个表出现多次，为了区分需要给表起别名：
    
        engine = create_engine(settings.DB_URI, echo=True)
        
        # 1: 实例化session
        session = Session(engine)
        # 2:起别名
        user_alias_1 = aliased(User)
        user_alias_2 = aliased(User)
        
        users = session.query(user_alias_1.name, user_alias_2.name).where(user_alias_1.name == user_alias_2.name).all()
        
        
        for row in users:
            print(row)
            # ('梁山', '梁山')
            # ('雪域', '雪域')
            # ('孟获', '孟获')
    
        # 1: 实例化session
        session = Session(engine)
        # 2:起别名
        user_alias_1 = aliased(User)
        user_alias_2 = aliased(User)
        
        users = session.execute(select(user_alias_1.name, user_alias_2.name).where(user_alias_1.name == user_alias_2.name)).all()
        
        
        for row in users:
            print(row)
            # ('梁山', '梁山')
            # ('雪域', '雪域')
            # ('孟获', '孟获')
        
        session.close()

### 4： 子查询： ###

#### 4.1: subquery ####

*  1: 使用subquery生成一个子查询语句：
    
        # 1: 实例化session
        session = Session(engine)
        
        subq = select(User.name, User.age).where(User.age > "20")
        print(subq)
        """
        SELECT users.name, users.age 
        FROM users 
        WHERE users.age > :age_1
        """
        users = session.execute(subq)
        for row in users:
            print(row)
            # ('梁山', '23')
            # ('雪域', '24')
            # ('孟获', '44')
 *  2：使用子查询：
    
        engine = create_engine(settings.DB_URI, echo=True)
        
        # 1: 实例化session
        session = Session(engine)
        
        subq = select(User.name, User.age).where(User.age > "20")
        
        users = session.execute(select(User.name).where(User.name == subq.c.name))
        
        
        for row in users:
            print(row)
            # ('梁山',)
            # ('雪域',)
            # ('孟获',)
        session.close()

#### 4.2: 通用表表达式（CTE）： ####

*  **与subquery的区别是，CTE可以递归使用**。