高级关联查询方法

关联查询 (Joins)

from database import Models
# 内连接
result = Models.session.query(Models.App, Models.User).join(
    Models.User, Models.App.user_id == Models.User.id
).all()

# 左外连接
result = Models.session.query(Models.App, Models.User).outerjoin(
    Models.User, Models.App.user_id == Models.User.id
).all()

# 多表连接
result = Models.session.query(Models.App, Models.User, Models.Company).join(
    Models.User, Models.App.user_id == Models.User.id
).join(
    Models.Company, Models.User.company_id == Models.Company.id
).all()

子查询 (Subqueries)

from database import Models
from sqlalchemy import exists_

# 使用 exists
subq = Models.session.query(Models.User).filter(
    Models.User.id == Models.App.user_id
).exists()
app_db = Models.session.query(Models.App).filter(subq).all()

# 作为列的子查询
subq = Models.session.query(
    func.count(Models.App.id)
).filter(
    Models.App.user_id == Models.User.id
).label('app_count')
user_db = Models.session.query(Models.User, subq).all()

其他实用方法

from database import Models
# 获取标量值
count = Models.session.query(func.count(Models.App.id)).scalar()

# 检查是否存在
exists = Models.session.query(
    Models.session.query(Models.App).filter_by(name="测试应用1").exists()
).scalar()

# 获取字典形式结果
app_dicts = [app.to_dict() for app in Models.session.query(Models.App).all()]

# 批量获取ID
app_ids = [id for (id,) in Models.session.query(Models.App.id).all()]

链式查询 (Chaining)

from database import Models
query = Models.session.query(Models.App)
query = query.filter(Models.App.category == "工具")
query = query.order_by(Models.App.rating.desc())
query = query.limit(10)
results = query.all()

动态查询构建

from database import Models
def search_apps(name=None, category=None, min_rating=None):
    query = Models.session.query(Models.App)
    if name:
        query = query.filter(Models.App.name.like(f"%{name}%"))
    if category:
        query = query.filter(Models.App.category == category)
    if min_rating:
        query = query.filter(Models.App.rating >= min_rating)
    return query.all()

原生SQL查询

from database import Models
# 直接执行SQL
result = Models.session.execute("SELECT * FROM apps WHERE rating > :rating", {"rating": 4.0})
for row in result:
    print(row)