如何将含CASE WHEN的SQL查询转换为SQLAlchemy代码？

首先需要从SQLAlchemy导入case工具函数，用来构造CASE WHEN逻辑：

from sqlalchemy import case

完整转换后的代码 #

# 构造SQLAlchemy查询
query = db.session.query(
    table1.id,
    table1.year,
    # 对应第一个CASE WHEN：判断processed是否为空
    case(
        [(table1.processed.is_(None), 'no')],
        else_='yes'
    ).label('Processed'),
    # 对应第二个CASE WHEN：处理Author字段
    case(
        [(table1.email.is_(None), 'oryginal')],
        else_=table1.email
    ).label('Author'),
    # 对应第三个多分支CASE WHEN：判断文件类型
    case(
        [(table1.txt.isnot(None), 'txt')],
        [(table1.csv.isnot(None), 'csv')],
        else_='no file'
    ).label('Type')
).filter(table1.pd_id == id)

# 转换为DataFrame（和原逻辑一致）
df = pd.read_sql(query.statement, db_engine)

各部分对应原SQL的说明 #

1. 第一个CASE逻辑
   对应原SQL：CASE WHEN processed IS NULL THEN 'no' else 'yes' END [Processed]
   SQLAlchemy中用case()接收条件-结果元组，is_(None)等价于SQL的IS NULL，label()用来设置列别名。

2. 第二个CASE逻辑
   对应原SQL：CASE WHEN email IS NULL THEN 'oryginal' else email END [Author]
   这里else_直接传入table1.email字段，等价于SQL中返回原字段值。

3. 多分支CASE逻辑
   对应原SQL：CASE WHEN txt IS NOT NULL THEN 'txt' WHEN csv IS NOT NULL THEN 'csv' ELSE 'no file' END [Type]
   在case()中传入多个条件-结果元组，SQLAlchemy会按顺序执行判断，匹配到第一个满足的条件就返回对应结果，最后用else_设置默认值。

额外说明 #

• 用filter(table1.pd_id == id)替代原SQL的字符串拼接，避免SQL注入风险，同时更符合SQLAlchemy的ORM规范。
• 最终通过query.statement获取查询语句，传入pd.read_sql()即可生成DataFrame，和原代码的输出完全一致。

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