本篇文章基于上一篇文章进行的优化编写:https://blog.csdn.net/zhouzhiwengang/article/details/112167171

Python 基于SQLAlchemy 实现对表的复杂查询操作

from orm import Users, Person, Hobby, Hosts, Session 
from sqlalchemy import and_, or_ 
from sqlalchemy.sql import func 
# 原生SQL查询 
from sqlalchemy.sql import text 
 
# 全局session 实例化 
session = Session() 
 
 
# 复杂条件查询一 
def condition_one(): 
    ret = session.query(Users).filter_by(name='alex').all() 
    print(ret) 
    ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all() 
    print(ret) 
    ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all() 
    print(ret) 
    ret = session.query(Users).filter(Users.id.in_([1, 3, 4])).all() 
    print(ret) 
    ret = session.query(Users).filter(~Users.id.in_([1, 3, 4])).all() 
    print(ret) 
    ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all() 
    print(ret) 
 
 
# 复杂条件查询二:and 或or 查询 
def condition_two(): 
    ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all() 
    print(ret) 
    ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all() 
    print(ret) 
    ret = session.query(Users).filter( 
        or_( 
            Users.id < 2, 
            and_(Users.name == 'eric', Users.id > 3), 
            Users.extra != "" 
        )).all() 
    print(ret) 
 
 
# 通配符查询 
def wildcard(): 
    ret = session.query(Users).filter(Users.name.like('e%')).all() 
    print(ret) 
    ret = session.query(Users).filter(~Users.name.like('e%')).all() 
    print(ret) 
 
 
# 限制查询 
def stint(): 
    ret = session.query(Users)[1:2] 
    print(ret) 
 
 
# 排序查询 
def sort(): 
    ret = session.query(Users).order_by(Users.name.desc()).all() 
    print(ret) 
    ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all() 
    print(ret) 
 
 
# 分组查询 
def group(): 
    ret = session.query(Users).group_by(Users.extra).all() 
    print(ret) 
    ret = session.query( 
        func.max(Users.id), 
        func.sum(Users.id), 
        func.min(Users.id)).group_by(Users.name).all() 
    print(ret) 
 
    ret = session.query( 
        func.max(Users.id), 
        func.sum(Users.id), 
        func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) > 2).all() 
    print(ret) 
 
 
# 连表查询 
def join(): 
    ret = session.query(Hobby, Person).filter(Person.hobby_id == Hobby.id).all() 
    print(ret) 
    ret = session.query(Person).join(Hobby).all() 
    print(ret) 
    ret = session.query(Person).join(Hobby, isouter=True).all() 
    print(ret) 
 
 
# 组合查询 
def combination(): 
    q1 = session.query(Users.name).filter(Users.id > 2) 
    q2 = session.query(Hosts.name).filter(Hosts.id < 2) 
    ret = q1.union(q2).all() 
    print(ret) 
 
    q1 = session.query(Users.name).filter(Users.id > 2) 
    q2 = session.query(Hosts.name).filter(Hosts.id < 2) 
    ret = q1.union_all(q2).all() 
    print(ret) 
 
 
# 全局session 关闭 
session.close() 
 
if __name__ == '__main__': 
    # condition_one() 
    # condition_two() 
    # wildcard() 
    # stint() 
    # sort() 
    # group() 
    # join() 
    combination()