Python3 集成SQLAlchemy ORM 框架：实现对表的复杂查询操作

朱雀 2023-10-10 12:15 83阅读 0赞

本篇文章基于上一篇文章进行的优化编写:[https://blog.csdn.net/zhouzhiwengang/article/details/112167171][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()

[https_blog.csdn.net_zhouzhiwengang_article_details_112167171]: https://blog.csdn.net/zhouzhiwengang/article/details/112167171