SQLAlchemy ORM查询中如何实现可选动态字段功能

需要在SQL Server环境下基于SQLAlchemy实现动态ORM查询，其中roles.name模糊匹配、roles.active状态过滤、排序规则均为非必填动态参数，对应原生SQL语句如下：

SELECT 
roles.id,
roles.name,
roles.abbreviation,
roles.active,
(CASE WHEN roles.updated_at IS NULL
    THEN roles.created_at
    ELSE roles.updated_at
    END) as last_Modified,
(CASE WHEN user_roles_count.number_of_users IS NULL
    THEN 0
    ELSE user_roles_count.number_of_users
    END) as number_of_users
FROM roles
    LEFT JOIN (SELECT user_roles.role_id, COUNT(user_roles.user_id) as number_of_users
        FROM user_roles GROUP BY user_roles.role_id
    ) as user_roles_count ON roles.id = user_roles_count.role_id ORDER BY roles.id ASC, last_Modified DESC;

初始编写的基础版本查询未适配动态可选字段逻辑，代码如下：

def get_roles(session, offset, limit):
    status_1 = f"""(CASE WHEN roles.updated_at IS NULL
        THEN roles.created_at
        ELSE roles.updated_at
        END) as last_Modified"""
    count_Query = session.query(UserRoles.role_id, func.count(
        UserRoles.user_id).label("number_of_users")).group_by(UserRoles.role_id).subquery()
    status_2 = f"""(CASE WHEN {count_Query.c.role_id} IS NULL
    THEN 0
    ELSE {count_Query.c.role_id}
    END) as number_of_users"""
    statement_result = session.query(
        Roles.id,
        Roles.name,
        Roles.abbreviation,
        Roles.active,
        text(status_1),
        text (status_2)
    ).join(count_Query, count_Query.c.role_id == Roles.id,
    isouter=True).order_by(asc(Roles.id)).slice(offset, limit).all()
    columns = ["id", "name", "abbreviation",
                "active", "updatedAt", "numberOfUsers"]
    get_products = struct_response(statement_result, columns)
    return get_products

排查确认原代码存在分页参数offset、limit的逻辑错误，修复后基于SQLAlchemy的链式调用特性完成动态逻辑适配：先构造不含动态条件的基础查询，再通过独立if判断校验入参，存在对应可选字段时追加对应的过滤、排序条件，最后统一执行分页查询。
最终可用代码如下：

def get_roles(session, offset, limit, event):
    status_1 = f"""(CASE WHEN roles.updated_at IS NULL
        THEN roles.created_at
        ELSE roles.updated_at
        END) as last_Modified"""

    count_Query = session.query(UserRoles.role_id, func.count(
        UserRoles.user_id).label("number_of_users")).group_by(UserRoles.role_id).subquery()

    status_2 = f"""(CASE WHEN {count_Query.c.role_id} IS NULL
    THEN 0
    ELSE {count_Query.c.role_id}
    END) as number_of_users"""

    # 构造基础查询
    query = session.query(
        Roles.id,
        Roles.name,
        Roles.abbreviation,
        Roles.active,
        text(status_1),
        text(status_2)
    ).join(count_Query, count_Query.c.role_id == Roles.id,
        isouter=True).order_by(asc(Roles.id))

    # 动态追加名称模糊搜索条件
    if "name" in event:
        name = event["name"]
        search = "%{}%".format(name)
        query = query.filter(Roles.name.like(search))
    # 动态追加状态过滤条件
    if "status" in event:
        query = query.filter(Roles.active == event["status"])
    # 动态追加排序规则
    if "newest_first" in event:
        if event["newest_first"] == True:
            query = query.order_by(asc(Roles.created_at))
        else:
            query = query.order_by(desc(Roles.created_at))

    # 执行分页查询
    query_result = query.slice(offset, limit).all()

    columns = ["id", "name", "abbreviation",
            "active", "updatedAt", "numberOfUsers"]
    get_roles = struct_response(query_result, columns)

    return get_roles

内容的提问来源于stack exchange，提问作者Juan David Polo