在 ORM 查询中包含可选字段

Question

我正在尝试执行动态查询，但其中三个字段是动态的并且不是必需的。在 SQL Server 中，查询是：

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;

列名 roles.name、roles.active 和 ORDER BY last_modified 是动态的，不是必需的。

目前使用 SQLAlchemy 我已经完成了这个

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

如何实现动态字段？

Answer 1

您可以构建要包含的字段列表，然后将该 *list 传递给查询，例如，

from sqlalchemy import Boolean, Column, Integer, select, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()


class Roles(Base):
    __tablename__ = "roles"
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    abbreviation = Column(String(5))
    active = Column(Boolean)


# basic query
columns_to_include = [Roles.id, Roles.abbreviation]
query = select(*columns_to_include)
print(query)
"""
SELECT roles.id, roles.abbreviation 
FROM roles
"""

# add optional field
columns_to_include.append(Roles.name)
query = select(*columns_to_include)
print(query)
"""
SELECT roles.id, roles.abbreviation, roles.name 
FROM roles
"""

你可以做同样的事情.order_by()：

# basic order_by
columns_to_order_by = [Roles.id]
query = select(*columns_to_include).order_by(*columns_to_order_by)
print(query)
"""
SELECT roles.id, roles.abbreviation, roles.name 
FROM roles ORDER BY roles.id
"""

# add another field for select() and order_by()
columns_to_include.append(Roles.active)
columns_to_order_by.append(Roles.active)
query = select(*columns_to_include).order_by(*columns_to_order_by)
print(query)
"""
SELECT roles.id, roles.abbreviation, roles.name, roles.active 
FROM roles ORDER BY roles.id, roles.active
"""