SQLAlchemy查询

结果查询

上节课使用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()) # 差剩下的所有