\"fix: remove list brackets from case() for SQLAlchemy 2.0 compatibility\"

This commit is contained in:
DXC
2026-03-12 08:56:08 +08:00
parent 7e403a87cd
commit 3ecdb121d2

View File

@ -313,22 +313,26 @@ class MaterialBaseService:
enable_warning_sort = filters.get('enableWarningSort', False) enable_warning_sort = filters.get('enableWarningSort', False)
if enable_warning_sort: if enable_warning_sort:
print("====== [DEBUG] 成功进入预警强排逻辑 ======") print("====== [DEBUG] 成功进入预警强排逻辑 (SQLA 2.0) ======")
# 强制统一数据类型 # 强制统一数据类型
red_val = cast(MaterialWarningSetting.red_threshold, Numeric) red_val = cast(MaterialWarningSetting.red_threshold, Numeric)
yellow_val = cast(MaterialWarningSetting.yellow_threshold, Numeric) yellow_val = cast(MaterialWarningSetting.yellow_threshold, Numeric)
inv_val = cast(total_inv, Numeric) inv_val = cast(total_inv, Numeric)
warning_level = case([ # 注意:移除了 case 内部的 []
warning_level = case(
(and_(MaterialWarningSetting.is_enabled.is_(True), inv_val <= red_val), 2), (and_(MaterialWarningSetting.is_enabled.is_(True), inv_val <= red_val), 2),
(and_(MaterialWarningSetting.is_enabled.is_(True), inv_val <= yellow_val), 1) (and_(MaterialWarningSetting.is_enabled.is_(True), inv_val <= yellow_val), 1),
], else_=0).label('sort_level') else_=0
red_shortage = case([ ).label('sort_level')
(and_(MaterialWarningSetting.is_enabled.is_(True), inv_val <= red_val), red_val - inv_val) red_shortage = case(
], else_=0).label('sort_red') (and_(MaterialWarningSetting.is_enabled.is_(True), inv_val <= red_val), red_val - inv_val),
yellow_distance = case([ else_=0
(and_(MaterialWarningSetting.is_enabled.is_(True), inv_val > red_val, inv_val <= yellow_val), inv_val - red_val) ).label('sort_red')
], else_=999999).label('sort_yellow') yellow_distance = case(
(and_(MaterialWarningSetting.is_enabled.is_(True), inv_val > red_val, inv_val <= yellow_val), inv_val - red_val),
else_=999999
).label('sort_yellow')
query = query.add_columns(warning_level, red_shortage, yellow_distance) query = query.add_columns(warning_level, red_shortage, yellow_distance)
query = query.order_by(None).order_by( query = query.order_by(None).order_by(
text("sort_level DESC"), text("sort_level DESC"),