在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.name。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

如何实现动态字段?

解决方案发现检查输入数据，我发现分页(偏移量和限制)是错误的，所以我进行了修复，代码按预期工作。感谢所有针对此问题的建议解决方案，并在事件

中添加了可选字段验证作为单个验证。

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 = query.slice(offset, limit).all()
columns = ["id", "name", "abbreviation",
"active", "updatedAt", "numberOfUsers"]
get_roles = struct_response(query, columns)
return get_roles

Answer 1

你可以建立一个你想要包含的字段列表，然后将这个*列表传递给查询，例如

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
"""