结果查询
上节课使用query从数据库中查询到了结果，但是query返回的对象是直接可用的吗？
首先导入模块
from connect import session from user_modules import User
query返回对象
rs = session.query(User) print(rs) for i in rs: print(i) #---------------------------------- #根据返回结果来看， rs 是一个 Query 对象，打印出来可以看到转化的 SQL rs = session.query(User).filter(User.username=='塔卡')
#all 是返回所有符合条件的数据 rs = session.query(User).all() #查询所有数据,返回一个列表 print(rs) print(rs[0].username) #属性访问
#first 是返回所有符合条件数据的第一条数据 rs = session.query(User).first() #查询第一条数据
#[0] 和 first 类似，但是如果没有符合条件的数据则会报错 session.query(User).filter(User.username=='budong')[0]
#这里，在 query 中查询对象的某个属性值 ( 对应为查询表中某个字段的值 )，返回的结果不再是一个 Query 对象，而是一个列表 rs = session.query(User.username).filter(User.username == '塔卡').all()
#同理，但是 first 返回结果是一个元组 rs = session.query(User.username).filter(User.username == '塔卡').first()
#[0] 和 first 类似，但是如果没有符合条件的数据则会报错 session.query(User.username).filter(User.username=='budong')[0] #getattr(rs[0], 'username'), rs[0].username这两种方式可以取到具体的数据值 rs = session.query(User).filter(User.username=='budong').all() print(hasattr(rs,'username')) print(getattr(rs,'username','litao')) rs = session.query(User)[0:2] #索引取值 print(rs) # rs2 = session.query(User).filter(User.username=='塔卡').all() # print(rs,type(rs))
条件查询一
过滤函数
filter 是一个过滤函数，过滤条件都可以书写在此函数中，不同的条件之间用 逗号 分隔
filter_by 也是一个过滤函数，但是功能要弱一些
filter 和 filter_by 的区别
二者都是 过滤函数，但是使用有如下差别：
1. filter 中需要添加 类对象，filter_by不需要
2. filter_by 中只能添加等于的条件，不能添加 不等于、大于小于等条件，filter没有这个限制
rs = session.query(User.username).filter(User.username == '塔卡') rs2 = session.query(User.username).filter_by(username = '塔卡') print(rs)
like 和 notlike
like 是模糊查询，和数据库中的 like 用法一样
notlike 和 like 作用相反
rs = session.query(User).filter(User.username.like('%塔%')).all() rs = session.query(User).filter(User.username.notlike('%塔%')).all()
in_ 和 notin_
in_ 和 notin_ 是范围查找，参数为列表
rs = session.query(User.username).filter(User.username.in_(['塔卡','小泼'])).all() rs = session.query(User.username).filter(User.username.notin_(['塔卡','小泼'])).all()
is_ 和 isnot
is_ 和 isnot 精确查找
rs = session.query(User).filter(User.username.is_(None)).all() rs = session.query(User).filter(User.username.isnot(None)).all() #判断为空还可以使用： session.query(User.id).filter(User.username==None).all()
查询结果数:限制查询
all、limit、offset、slice、one
rs =session.query(User.username).all() print(rs) rs =session.query(User.username).limit(2).all() #限制数量查询 rs =session.query(User.username).offset(2).all() #偏移量 rs =session.query(User.username).slice(1,4).all() #切片 #不管怎样写one只能查一条数据，如果有多条重复数据，会报错 #sqlalchemy.orm.exc.MultipleResultsFound: Multiple rows were found for one() rs = session.query(User.username).filter(User.username == '李涛').one() print(rs)
排序：倒叙先进行导入desc
from sqlalchemy import desc rs = session.query(User.username,User.id).order_by(User.id).all() #升序排列 rs = session.query(User.username,User.id).order_by(desc(User.id)).all() #综合使用 rs = session.query(User.username,User.id).order_by(desc(User.id)).filter(User.username == '李涛').all() print(rs)
聚合函数
func.count
使用函数时，需要导入 func， group_by 和 order_by 一样，是可以直接使用的，不需要导入
having 也可以直接使用，使用方法也和 SQL 中使用类似
func.sum、func.max、func.min
extract
extract 提取对象中的数据，这里提取分钟，并把提取出来的结果用 label 命名别名，之后就可以使用 group_by 来分组
count 里面同样可以使用 *
or_
or_ 是或者的意思，和数据库中的 or 一样
# 分组查询与聚合函数一起使用 from sqlalchemy import func,extract,or_ # rs = session.query(User.password,func.count(User.id)).group_by(User.password).all() # rs = session.query(User.password,func.count(User.id)).group_by(User.password).having(func.count(User.id) >1).all() # print(rs) # rs = session.query(User.password,func.sum(User.id)).group_by(User.password).all() # rs = session.query(User.password,func.max(User.id)).group_by(User.password).all() # rs = session.query(User.password,func.min(User.id)).group_by(User.password).all() # print(rs) #extract # rs = session.query(extract('minute',User.create_time).label('minute'),func.count(User.id)).group_by('minute').all() #or_ rs = session.query(User.username).filter(or_(User.username.isnot(None),User.password == '1234')).all() print(rs)
3 多表查询
在user_modules.py中增加一张表UserDetails表
class UserDetails(Base): __tablename__='user_details' id = Column(Integer, primary_key=True, autoincrement=True) id_card = Column(Integer,nullable=True,unique=True) lost_login = Column(DateTime) login_num = Column(Integer,default=0) user_id = Column(Integer,ForeignKey('user.id')) def __repr__(self): return '<UserDetails(id=%s,id_card=%s,last_login=%s,login_num=%s,user_id=%s)>'%( self.id, self.id_card, self.lost_login, self.login_num, self.user_id )
在query_test.py文件中导入新增加的表
from user_modules import User,UserDetails
笛卡尔连接：
rs =session.query(User.username,UserDetails.lost_login).filter(UserDetails.user_id==User.id).all()
去掉.all()，查看原生SQL如下：
SELECT user.id AS user_id, user.username AS user_username, user.password AS user_password,
user.create_time AS user_create_time, user._locked AS user__locked, user_details.id AS user_details_id,
user_details.id_card AS user_details_id_card, user_details.lost_login AS user_details_lost_login,
user_details.login_num AS user_details_login_num, user_details.user_id AS user_details_user_id
FROM user, user_details
WHERE user_details.user_id = user.id
inner join：内连接
rs = session.query(User.username,UserDetails.lost_login).
join(UserDetails,UserDetails.user_id == User.id)
查看原生SQL如下：
SELECT user.username AS user_username, user_details.lost_login AS user_details_lost_login
FROM user INNER JOIN user_details ON user_details.user_id = user.id
###使用笛卡尔积的filter条件与使用inner join运行的结果一致。但运行的原生sql不一致。
left join：外连接，可以将两张表对调
rs = session.query(User.username,UserDetails.lost_login). outerjoin(UserDetails,UserDetails.user_id==User.id).all() # print(rs)
运行SQL如下：
SELECT user.username AS user_username, user_details.lost_login AS user_details_lost_login
FROM user LEFT OUTER JOIN user_details ON user_details.user_id = user.id
运行结果如下：
#两张表对调
rs = session.query(UserDetails.lost_login,User.username). outerjoin(User,UserDetails.user_id==User.id).all() # print(rs)
运行SQL如下：
SELECT user_details.lost_login AS user_details_lost_login, user.username AS user_username
FROM user_details LEFT OUTER JOIN user ON user_details.user_id = user.id
运行结果如下：
union：
联合查询，有自动去重的功能，对应的还有 union_all
q1 = session.query(User.id) q2 = session.query(UserDetails.user_id) # print(q1.union(q2).all())
子表查询
声明子表
sql_0 = session.query(UserDetails.lost_login).subquery() #声明子表
#使用 # print(session.query(User,sql_0.c.lost_login))
运行结果如下：
SELECT user.id AS user_id, user.username AS user_username, user.password AS user_password,
user.create_time AS user_create_time, user._locked AS user__locked, anon_1.lost_login AS anon_1_lost_login
FROM user, (SELECT user_details.lost_login AS lost_login
FROM user_details) AS anon_1
原生SQL查询：
sql_1 = ''' select * from `user` ''' #查询 row = session.execute(sql_1) #循环取值
# for i in row: # print(i)
#取值 print(row) #fetch会记录游标运行了上面的for以后下面就查不到数据 print(row.fetchone()) #查询一条 print(row.fetchmany()) #在多查询一条 print(row.fetchall()) # 差剩下的所有