SQLAlchemy查询

SQLAlchemy查询

结果查询：

1 from databases.wechat import User 2 from config import session 3 4

5 def search():

6 result = session.query(User).all() # 查询所有 7 result = result[0] # 索引取值

8 print(result.username) # 对象属性查询 9 session.query(User).first() # 查询第⼀条10

11 session.query(User).filter(User.username == 'bob').all() # 按条件查询所有12 13

14 if __name__ == '__main__':15 search()

all() :

查询所有

返回⼀个列表对象first()

查询第⼀个符合条件的对象返回⼀个对象索引取值

相当于列表取值

返回⼀个列表内的值(对象)条件查询：

⽤fillter⽅法来增加查询条件属性查询：

直接该属性的对象对其进⾏普通的类属性的调⽤即可 条件查询

1 from databases.wechat import User 2 from config import session 3 4

5 def search():

6 # query接收⼀个查询范围，fillter增加查询条件的约束

7 result = session.query(User.username).filter(User.username=='bob').all() # [('bob',)] 8 result = session.query(User.username).filter_by(username='bob').all() # [('bob',)] 9 \"\"\"

10 fillter和filter_by

11 fillter可以进⾏⽐较运算(==, >, < ...)来对条件进⾏灵活的运⽤, 不同的条件⽤'，'(逗号)分割12 fillter_by只能指定参数传参来获取查询结果13 \"\"\"14 15 16

17 if __name__ == '__main__':18 search()

query接收⼀个查询范围多个范围⽤逗号隔开，fillter增加查询条件的约束fillter和filter_by

fillter可以进⾏⽐较运算(==, >, < ...)来对条件进⾏灵活的运⽤, 不同的条件⽤'，'(逗号)分割fillter_by只能指定参数传参来获取查询结果

模糊查询

1 from databases.wechat import User 2 from config import session 3 4

5 def search():

6 # like⾥⾯传⼊⼀个字符串，不确定的位置⽤%代替即可

7 result = session.query(User.username).filter(User.username.like('b%')).all() # [('bob',)] 8 # notlike取like的取反结果

9 result = session.query(User.username).filter(User.username.notlike('b%')).all()10 # is_ 相当于 ==

11 result = session.query(User.username).filter(User.username.is_(None)).all()12 result = session.query(User.username).filter(User.username == None).all()13 # isnot 相当于 !=

14 result = session.query(User.username).filter(User.username.isnot(None)).all()15 result = session.query(User.username).filter(User.username != None).all()16 # in_传⼊⼀个可迭代对象,对前⾯的username进⾏约束， notin_ 和in_取反

17 result = session.query(User.username).filter(User.username.in_(['bob', 'ivy1'])).all()18 result = session.query(User.username).filter(User.username.notin_(['bob', 'ivy1'])).all()

19 # limit 数量查询, limit⾥传⼊⼀个整型来约束查看的数量, 当limit⾥⾯的参数⼤于实例表中的数量时，会返回所有的查询结果20 result = session.query(User.username).limit(6).all()

21 # offset 偏移量查询,offset中传⼊⼀个整型，从表中的该位置开始查询，offset可以和limit混⽤来进⾏22 result = session.query(User.username).offset(1).all()

23 result = session.query(User.username).offset(1).limit(6).all()24 # slice 切⽚查询,遵循左闭右开原则，可以和offset、limit混⽤

25 result = session.query(User.username).slice(1, 3).offset(2).limit(6).all()

26 # one 获取查询对象的⼀条，且查询的结果有且仅有⼀条，但查询结果多了的时候会报错27 result = session.query(User.username).filter_by(username='bob').one()28 29 30 31

32 if __name__ == '__main__':33 search()

like⾥⾯传⼊⼀个字符串，不确定的位置⽤%代替即可notlike取like的取反结果is_ 相当于 ==

isnot 相当于 !=

in_传⼊⼀个可迭代对象,对前⾯的username进⾏约束， notin_ 和in_取反

limit 数量查询, limit⾥传⼊⼀个整型来约束查看的数量, 当limit⾥⾯的参数⼤于实例表中的数量时，会返回所有的查询结果offset 偏移量查询,offset中传⼊⼀个整型，从表中的该位置开始查询，offset可以和limit混⽤来进⾏slice 切⽚查询,遵循左闭右开原则，可以和offset、limit混⽤

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

1 from databases.wechat import User 2 from config import session 3 from sqlalchemy import desc 4

5 def search(): 6 # 升序排列

7 result = session.query(User.username, User.id).order_by(User.id).all() 8 # 降序排列

9 result = session.query(User.username, User.id).order_by(desc(User.id)).all()10 # 结合filter查询

11 result = session.query(User.username, User.id).order_by(User.id).filter(User.username != 'bob').all()12 result = session.query(User.username, User.id).filter(User.username != 'bob').order_by(User.id).all()13 14 15 16

17 if __name__ == '__main__':18 search()

排序查询，排序查询可结合filter、limit、slice等综合运⽤　聚合函数

1 from databases.wechat import User 2 from databases.config import session 3 from sqlalchemy import func, extract 4 5

6 def search(): 7 # count

8 result = session.query(User.password, func.count(User.id)).group_by(User.password).all() 9 # sum

10 result = session.query(User.password, func.sum(User.id)).group_by(User.password).all()11 # max

12 result = session.query(User.password, func.max(User.id)).group_by(User.password).all()13 # min

14 result = session.query(User.password, func.min(User.id)).group_by(User.password).all()15 # having

16 result = session.query(User.password, func.count(User.id)).group_by(User.password).having(func.count(User.id) > 1).all()17 # label extract

18 result = session.query(

19 extract('minute',User.create_time).label('minute'),20 func.count(User.id)21 ).group_by('minute')

22 # 想当于-->SELECT EXTRACT(minute FROM user.create_time) AS minute, count(user.id) AS count_1 FROM user GROUP BY minute23

24 if __name__ == '__main__':25 search()

多表查询

1 from databases.config import Base

2 from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey 3 from datetime import datetime 4

5 class User(Base):

6 __tablename__ = 'user'

7 id = Column(Integer, primary_key=True, autoincrement=True) 8 username = Column(String(20)) 9 password = Column(String(20))

10 create_time = Column(DateTime, default=datetime.now())11 is_login = Column(Boolean, default=False, nullable=False)12 13

14 class UserDetails(Base):

15 __tablename__ = 'userdetails'

16 id = Column(Integer, primary_key=True, autoincrement=True)17 id_card = Column(Integer, nullable=True, unique=True)18 last_login = Column(DateTime)

19 login_num = Column(Integer, default=0)

20 user_id = Column(Integer, ForeignKey('user.id')) # user.id 表名+属性名21 22 23

24 if __name__ == '__main__':25 Base.metadata.create_all()

新建表

1 from databases.wechat import User, UserDetails 2 from databases.config import session 3 4

5 def search(): 6 # 笛卡尔连接

7 result = session.query(User, UserDetails)

8 # 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 us 9 # 加filter查询

10 result = session.query(User, UserDetails).filter(UserDetails.id==User.id).all()

11 result = session.query(User.username, UserDetails.id_card).join(UserDetails, UserDetails.id==User.id).filter(UserDetails.id==User.id)

12 # 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.id13 14 15 16 17

18 if __name__ == '__main__':19 search()

原⽣sql查询

1 from databases.config import session

2 3

4 def search():

5 sql = 'select * from user ' 6 result = session.execute(sql) 7 result.fetchone() 8 result.fetchmany() 9 result.fetchone()10 11 12 13

14 if __name__ == '__main__':15 search()