SQLAlchemy查询

SQLAlchemy查询

---

结果查询：

• from databases.wechat import User
  from config import session
  
  
  def search():
      result = session.query(User).all()  # 查询所有
      result = result[0]  # 索引取值
      print(result.username)  # 对象属性查询
      session.query(User).first()  # 查询第一条
  
      session.query(User).filter(User.username == 'bob').all()  # 按条件查询所有
  
  
  if __name__ == '__main__':
      search()

• all() :
  • 查询所有
  • 返回一个列表对象
• first()
  • 查询第一个符合条件的对象
  • 返回一个对象
• 索引取值
  • 相当于列表取值
  • 返回一个列表内的值(对象)
• 条件查询：
  • 用fillter方法来增加查询条件
• 属性查询：
  • 直接该属性的对象对其进行普通的类属性的调用即可

条件查询

• from databases.wechat import User
  from config import session
  
  
  def search():
      # query接收一个查询范围，fillter增加查询条件的约束
      result = session.query(User.username).filter(User.username=='bob').all()  # [('bob',)]
      result = session.query(User.username).filter_by(username='bob').all()  # [('bob',)]
      """
      fillter和filter_by
      fillter可以进行比较运算(==, >, < ...)来对条件进行灵活的运用, 不同的条件用'，'(逗号)分割
      fillter_by只能指定参数传参来获取查询结果
      """
  
  
  
  if __name__ == '__main__':
      search()

  query接收一个查询范围多个范围用逗号隔开，fillter增加查询条件的约束
  fillter和filter_by
  fillter可以进行比较运算(==, >, < ...)来对条件进行灵活的运用, 不同的条件用'，'(逗号)分割
  fillter_by只能指定参数传参来获取查询结果

模糊查询

• from databases.wechat import User
  from config import session
  
  
  def search():
      # like里面传入一个字符串，不确定的位置用%代替即可
      result = session.query(User.username).filter(User.username.like('b%')).all()  # [('bob',)]
      # notlike取like的取反结果
      result = session.query(User.username).filter(User.username.notlike('b%')).all()
      # is_ 相当于 ==
      result = session.query(User.username).filter(User.username.is_(None)).all()
      result = session.query(User.username).filter(User.username == None).all()
      # isnot 相当于 !=
      result = session.query(User.username).filter(User.username.isnot(None)).all()
      result = session.query(User.username).filter(User.username != None).all()
      # in_传入一个可迭代对象,对前面的username进行约束， notin_ 和in_取反
      result = session.query(User.username).filter(User.username.in_(['bob', 'ivy1'])).all()
      result = session.query(User.username).filter(User.username.notin_(['bob', 'ivy1'])).all()
      # limit 限制数量查询, limit里传入一个整型来约束查看的数量, 当limit里面的参数大于实例表中的数量时，会返回所有的查询结果
      result = session.query(User.username).limit(6).all()
      # offset 偏移量查询,offset中传入一个整型，从表中的该位置开始查询，offset可以和limit混用来进行限制
      result = session.query(User.username).offset(1).all()
      result = session.query(User.username).offset(1).limit(6).all()
      # slice 切片查询,遵循左闭右开原则，可以和offset、limit混用
      result = session.query(User.username).slice(1, 3).offset(2).limit(6).all()
      # one 获取查询对象的一条，且查询的结果有且仅有一条，但查询结果多了的时候会报错
      result = session.query(User.username).filter_by(username='bob').one()
  
  
  
  
  if __name__ == '__main__':
      search()

• like里面传入一个字符串，不确定的位置用%代替即可

• notlike取like的取反结果

• is_ 相当于 ==

• isnot 相当于 !=

• in_传入一个可迭代对象,对前面的username进行约束， notin_ 和in_取反

• limit 限制数量查询, limit里传入一个整型来约束查看的数量, 当limit里面的参数大于实例表中的数量时，会返回所有的查询结果

• offset 偏移量查询,offset中传入一个整型，从表中的该位置开始查询，offset可以和limit混用来进行限制

• slice 切片查询,遵循左闭右开原则，可以和offset、limit混用

• one 获取查询对象的一条，且查询的结果有且仅有一条，但查询结果多了的时候会报错

• from databases.wechat import User
  from config import session
  from sqlalchemy import desc
  
  def search():
      # 升序排列
      result = session.query(User.username, User.id).order_by(User.id).all()
      # 降序排列
      result = session.query(User.username, User.id).order_by(desc(User.id)).all()
      # 结合filter查询
      result = session.query(User.username, User.id).order_by(User.id).filter(User.username != 'bob').all()
      result = session.query(User.username, User.id).filter(User.username != 'bob').order_by(User.id).all()
  
  
  
  
  if __name__ == '__main__':
      search()

  排序查询，排序查询可结合filter、limit、slice等综合运用

　

聚合函数

• from databases.wechat import User
  from databases.config import session
  from sqlalchemy import func, extract
  
  
  def search():
      # count
      result = session.query(User.password, func.count(User.id)).group_by(User.password).all()
      # sum
      result = session.query(User.password, func.sum(User.id)).group_by(User.password).all()
      # max
      result = session.query(User.password, func.max(User.id)).group_by(User.password).all()
      # min
      result = session.query(User.password, func.min(User.id)).group_by(User.password).all()
      # having
      result = session.query(User.password, func.count(User.id)).group_by(User.password).having(func.count(User.id) > 1).all()
      # label extract
      result = session.query(
          extract('minute',User.create_time).label('minute'),
          func.count(User.id)
      ).group_by('minute')
      # 想当于-->SELECT EXTRACT(minute FROM user.create_time) AS minute, count(user.id) AS count_1 FROM user GROUP BY minute
  
  if __name__ == '__main__':
      search()

多表查询

• from databases.config import Base
  from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey
  from datetime import datetime
  
  class User(Base):
      __tablename__ = 'user'
      id = Column(Integer, primary_key=True, autoincrement=True)
      username = Column(String(20))
      password = Column(String(20))
      create_time = Column(DateTime, default=datetime.now())
      is_login = Column(Boolean, default=False, nullable=False)
  
  
  class UserDetails(Base):
      __tablename__ = 'userdetails'
      id = Column(Integer, primary_key=True, autoincrement=True)
      id_card = Column(Integer, nullable=True, unique=True)
      last_login = Column(DateTime)
      login_num = Column(Integer, default=0)
      user_id = Column(Integer, ForeignKey('user.id'))  # user.id 表名+属性名
  
  
  
  if __name__ == '__main__':
      Base.metadata.create_all()

   新建表

• from databases.wechat import User, UserDetails
  from databases.config import session
  
  
  def search():
      # 笛卡尔连接
      result = session.query(User, UserDetails)
      # SELECT user.id AS user_id, user.username AS user_username, user.password AS user_password, user.create_time AS user_create_time, user.is_login AS user_is_login, userdetails.id AS userdetails_id, userdetails.id_card AS userdetails_id_card, userdetails.last_login AS userdetails_last_login, userdetails.login_num AS userdetails_login_num, userdetails.user_id AS userdetails_user_id FROM user, userdetails
      # 加filter查询
      result = session.query(User, UserDetails).filter(UserDetails.id==User.id).all()
      result = session.query(User.username, UserDetails.id_card).join(UserDetails, UserDetails.id==User.id).filter(UserDetails.id==User.id)
      # SELECT user.username AS user_username, userdetails.id_card AS userdetails_id_card FROM user INNER JOIN userdetails ON userdetails.id = user.id WHERE userdetails.id = user.id
  
  
  
  
  
  if __name__ == '__main__':
      search()

原生sql查询

• from databases.config import session
  
  
  def search():
      sql = 'select * from user '
      result = session.execute(sql)
      result.fetchone()
      result.fetchmany()
      result.fetchone()
  
  
  
  
  if __name__ == '__main__':
      search()