如何将原生SQL转为异步SQLAlchemy并实现动态条件查询

实现完全覆盖两个核心需求：

• 原生SQL逻辑1:1映射为异步SQLAlchemy 2.0语法
• search_title参数动态判空，参数为None时自动跳过标题模糊过滤，返回全量统计结果

注意：原SQL中写的FULL JOIN shops存在明显笔误，关联条件为manager.id=cars.manager_id且SELECT子句统计的是cars表数据，以下实现按关联cars表编写，若确实需要关联shops表替换对应模型即可。

前置ORM模型定义 #

from sqlalchemy import Column, String, DateTime, Integer, ForeignKey, func, case
from sqlalchemy.orm import DeclarativeBase
from datetime import datetime

class Base(DeclarativeBase):
    pass

class Manager(Base):
    __tablename__ = "manager"
    id = Column(Integer, primary_key=True)
    title = Column(String)
    status = Column(String)
    created_at = Column(DateTime, default=datetime.utcnow)

class Cars(Base):
    __tablename__ = "cars"
    id = Column(Integer, primary_key=True)
    manager_id = Column(Integer, ForeignKey("manager.id"))
    status = Column(String)

核心查询实现 #

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def get_manager_car_stats(
    db: AsyncSession,
    search_title: str | None,
    order_key: str,
    page_limit: int
):
    # 排序字段白名单校验，彻底避免SQL注入
    allowed_order_fields = {"title", "status", "created_at"}
    if order_key not in allowed_order_fields:
        raise ValueError("非法排序字段")

    # 构建基础查询框架
    query = (
        select(
            Manager.title,
            Manager.status,
            func.count(Cars.id).label("total_cars"),
            func.count(
                case((Cars.status == "active", 1), else_=None)
            ).label("active_cars"),
            Manager.created_at
        )
        .join(Cars, Manager.id == Cars.manager_id, full=True) # 对应FULL JOIN逻辑
        .group_by(Manager.created_at, Manager.status, Manager.title)
        .order_by(getattr(Manager, order_key))
        .limit(page_limit)
    )

    # 动态追加过滤条件：仅当search_title非None时生效
    if search_title is not None:
        query = query.where(Manager.title.ilike(f"%{search_title}%"))

    # 异步执行查询返回结果
    result = await db.execute(query)
    return result.all()

关键逻辑说明 #

• 动态过滤逻辑：先搭建基础查询结构，再判断search_title是否有值，无值时不追加任何标题过滤条件，直接返回全量分组统计结果
• 聚合逻辑对齐：COUNT(CASE WHEN cars.status='active' THEN 1 END)通过SQLAlchemy内置case构造实现，统计结果和原生SQL完全一致
• 安全加固：原f-string直接拼接SQL的写法存在注入风险，实现中查询参数由SQLAlchemy自动转义，排序字段提前做白名单校验，彻底规避注入问题
• 连接类型可调整：join方法的full=True参数对应原生FULL JOIN，可根据业务需要调整参数切换为内连接、左连接
• 返回结果为命名元组格式，可直接通过.title、.total_cars、.active_cars等属性访问字段值

内容的提问来源于stack exchange，提问作者Akyl Alamanov