1. 程式人生 > 其它 >Python3 整合SQLAlchemy ORM 框架:實現對錶的複雜查詢操作

Python3 整合SQLAlchemy ORM 框架:實現對錶的複雜查詢操作

技術標籤:python 學習筆記

本篇文章基於上一篇文章進行的優化編寫: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()