SQLAlchemy -高级查询

SQLAlchemy -高级查询,第1张

SQLAlchemy -高级查询

查询

  1. # -*- coding: utf-8 -*-  

  2. from sqlalchemy.orm import sessionmaker  

  3. from SQLAlchemy.create import engine,User  

  4. Session = sessionmaker(engine)  

  5. db_session = Session()  

  6.             
     
  7. query = db_session.query(User).filter(User.name.like('%2%')).order_by(User.id)  

  8.             
     
  9. # ------> 查询 all list  
  10. print(query.all())  

  11.             
     
  12. # ------> first() 第一个  

  13. print(query.first())  

  14.             
     
  15. # ------> one()  如果这个结果集少于或者多于一条数据, 结论有且只有一条数据的时候才会正常的返回,否则抛出异常  

  16. print(query.one())  

  17.             
     
  18. # ------> one_or_none() 在结果集中没有数据的时候也不会抛出异常  

  19.             
     
  20. # ------> scalar()  底层调用one()方法,并且如果one()方法没有抛出异常,会返回查询到的第一列的数据  

  21. query2 = db_session.query(User).filter(User.name=="2").scalar()  

  22. print(query2)  

  23.             
     
  24. # ------> text  

  25. from sqlalchemy import text  

  26. query3 = db_session.query(User).filter(text("id<3")).order_by(text('id'))  

  27. print(query3.all())  

  28.             
     
  29. # text 带变量方式 用 :传入变量,用params接收  

  30. query4 = db_session.query(User).filter(text("id<:value")).params(value=4)  

  31. print(query4.all())  

  32.             
     
  33. # from_statement 原生sql语句  

  34. query5 = db_session.query(User).from_statement(text("select * from user where id>:value")).params(value=2).all()  

  35. print(query5)  

  36.  
     
  37. # ------> and_ or_ 普通的表达式在这里面不好使  
  38. from sqlalchemy.sql import and_, or_    
  39. query6 = db_session.query(User).filter(and_(User.id==1, User.name=="ewqe")).first()  
  40. print(query6.id)  
  41. # 结果 1  
  42.   
     
  43. query7 = db_session.query(User).filter(or_(User.id==1, User.name=="rre")).all()  
  44. print([i.id for i in query7])  
  45. # 结果 [1, 3]  
  46.  
  47. # ------> between 大于多少小于多少  
  48. query8 = db_session.query(User).filter(User.id.between(1, 3)).all()  
  49. print(query8)  
  50.   
     
  51. # ------> in_  在里面对应还有not in等  
  52. query9 = db_session.query(User).filter(User.id.in_([1])).all()  
  53. print(query9)  
  54.  
     
  55. # ------> synchronize_session
  56. # synchronize_session=False 在原有值的基础上增加和删除 099 str  
  57. db_session.query(User).filter(User.id > 0).update({User.name: User.name + '099'}, synchronize_session=False)  
  58. db_session.commit()  
  59.   
     
  60. # synchronize_session=evaluate 在原有值的基础上增加和减少 11, 必须是整数类型  
  61. db_session.query(User).filter(User.id > 0).update({"name": User.name + 1}, synchronize_session="evaluate")  
  62. db_session.commit()  
  63.   
     
  64. # 如果查询条件里有in_,需要在delete()中加如下参数: fetch  删除的更快  
  65. query10 = db_session.query(User).filter(User.id.in_([1, 2])).delete(synchronize_session='fetch')  
  66. print(query10) # 返回找到的数量  
  67. db_session.commit() 
  68.  

           
 

计数(Count)

  1. 有时候你想明确的计数,比如要统计users表中有多少个不同的姓名,  

  2. 那么简单粗暴的采用以上count是不行的,因为姓名有可能会重复,  

  3. 但是处于两条不同的数据上,如果在原生数据库中,可以使用distinct关键字,  

  4. 那么在SQLAlchemy中,可以通过func.count()方法来实现  

  5. from sqlalchemy import func  

  6. query1 = db_session.query(func.count(User.name)).first() #  (4,)  

  7.           
     
  8. # 如果想实现select count(*) from users,可以通过以下方式来实现:  

  9. query2 = db_session.query(func.count("*")).select_from(User).scalar() # 4  

  10.           
     
  11. # 如果指定了要查找的表的字段,可以省略select_from()方法:  

  12. query3 = db_session.query(func.count(User.id)).scalar() # 4  

   
 

排序

  1. order_by:可以指定根据这个表中的某个字段进行排序,如果在前面加了一个-,代表的是降序排序    

  2. 正向排序和反向排序:默认情况是从小到大,从前到后排序的,如果想要反向排序,可以调用排序的字段的desc方法。


      

一.
relationship的order_by参数 

teachers = relationship('Teacher',secondary=association_table, back_populates='classes', order_by=Teacher.name)  

二. 在模型定义中,添加以下代码:  

  1.     __mapper_args__ = {  

  2.      "order_by": name  

  3.    }  

   
 

limit、offset和切片

  1. limit:可以限制每次查询的时候只查询几条数据。


      

  2. offset:可以限制查找数据的时候过滤掉前面多少条。


      

  3. 切片:可以对Query对象使用切片 *** 作,来获取想要的数据。


      

  4.       
     
  5. all = db_session.query(Teacher).limit(2).all()  

  6. print([i.id for i in all ])  

  7. # 结果 [1, 2]  

  8.       
     
  9. all2 = db_session.query(Teacher).limit(2).offset(1).all()  

  10. print([i.id for i in all2])  

  11. # 结果 [2, 3]  

   
 

懒加载

  1. 在一对多,或者多对多的时候,如果想要获取多的这一部分的数据的时候,往往能通过一个属性就可以全部获取了。


      

  2. 比如有一个作者,想要或者这个作者的所有文章,那么可以通过user.articles就可以获取所有的。


      

  3. 但有时候我们不想获取所有的数据,比如只想获取这个作者今天发表的文章,  

  4. 那么这时候我们可以给relationship传递一个lazy='dynamic',  

  5. 以后通过user.articles获取到的就不是一个列表,而是一个AppendQuery对象了。


      

  6. 这样就可以对这个对象再进行一层过滤和排序等 *** 作  

   
 

group_by

比如我想根据名字来分组, 统计每个名字分组里面有多少人  

  1. from sqlalchemy import func  

  2. # 我想根据名字来分组, 统计每个名字分组里面有多少人  

  3. all = db_session.query(Teacher.name, func.count(Teacher.id)).group_by(Teacher.name).all()  

   
 

having

  1. having是对查找结果进一步过滤。


    比如只想要看未成年人的数量,那么可以首先对年龄进行分组统计人数,然后再对分组进行having过滤  

  2.       
     
  3. result = session.query(User.age,func.count(User.id)).group_by(User.age).having(User.age >= 18).all()  

   
 

join方法

表结构

  1. class Address(Base):  

  2.     __tablename__ = 'address'  

  3.     id = Column(Integer, primary_key=True)  

  4.     name = Column(String(32))  

  5.     user_id = Column(Integer,ForeignKey("user.id"))  # 外键关系 Fk  

  6.     
     
  7.     
     
  8. class User(Base):  

  9.     __tablename__ = 'user'  

  10.     id = Column(Integer, primary_key=True)  

  11.     name = Column(String(32))

 
 

  1. ---> 普通查询  

  2. all = db_session.query(User, Address).filter(User.id == Address.user_id).all()  

  3. for u, a in all:  

  4.     print(u.name, a.name)  

  5.         
     
  6. 输出结果  

  7. peach 地址1  

  8. peach 地址2  

  9. peach 地址3  

  10.     
     
  11. ---> join查询  

  12. all = db_session.query(User, Address).join(Address).all()  

  13. for u, a in all:  

  14.     print(u.name, a.name)  

  15.         
     
  16. 输出结果  

  17. peach 地址1  

  18. peach 地址2  

  19. peach 地址3  

  20.     
     
  21. ---> outerjoin查询 left join是以左边为准 right 是以右边为准  

  22. all = db_session.query(User, Address).outerjoin(Address).all()  

  23. for u in all:  

  24.     print(u)  

  25.     
     
  26. 输出结果  

  27. (<__main__.User object at 0x7fb9e3160518><__main__.Address object at 0x7fb9e3160588>)  

  28. (<__main__.User object at 0x7fb9e3160518><__main__.Address object at 0x7fb9e31605f8>)  

  29. (<__main__.User object at 0x7fb9e3160518><__main__.Address object at 0x7fb9e3160668>)  

  30. (<__main__.User object at 0x7fb9e31606d8>, None)  

 
 

别名

  1. # 当多表查询的时候,有时候同一个表要用到多次,这时候用别名就可以方便的解决命名冲突的问题了  

  2. from sqlalchemy.orm import aliased  

  3. adalias1 = aliased(Address)  

  4. adalias2 = aliased(Address)  

  5.     
     
  6. for username, ad1name, ad2name in db_session.query(User.name, adalias1.name, adalias2.name).join(adalias1).all():  

  7.     print(username, ad1name, ad2name)  

 
 

子查询

  1. # 构造子查询  

  2. # 首先查询用户有多少地址  

  3. from sqlalchemy.sql import func  

  4. addr = db_session.query(Address.user_id.label('user_id'), func.count("*").label('address_count')).group_by(Address.user_id).all()  

  5. # 结果  

  6. 个地址  

  7. .label('user_id') 是取的别名  

 
 

  1. # 将子查询放到父查询中  

  2. addrs = db_session.query(Address.user_id.label('user_id'), func.count("*").label('address_count')).group_by(Address.user_id).subquery() # 子查询语句  

  3. for u,count in db_session.query(User,addrs.c.address_count).outerjoin(addrs,User.id==addrs.c.user_id).order_by(User.id):  

  4.     print(u.name,count)  

  5.     
     
  6. # 结果  

  7. # peach 3  

  8. # peach2 None  

  9.     
     
  10. # 一个查询如果想要变为子查询,则是通过subquery()方法实现  

  11. # 变成子查询后,通过子查询.c属性来访问查询出来的列  

  12. # 以上方式只能查询某个对象的具体字段  

 
 

以上方式只能查询某个对象的具体字段,如果要查找整个实体,则需要通过aliased方法

 
 

 
 

  

欢迎分享,转载请注明来源:内存溢出

原文地址: http://outofmemory.cn/zaji/587811.html

(0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
上一篇 2022-04-12
下一篇 2022-04-12

发表评论

登录后才能评论

评论列表(0条)

保存