在 sqlalchemy 中使用 case 时出现 SQL 语法错误

Question

我在 sqlAlchemy 中使用 case 如下所示：

abc = "%abc%"
def = "%def%"
proj1 = "%project1%"
proj2 = "%project2%"
case_condition = case([
        (text('FPro.status = "ON" and Ab.name.like (''' + abc + ''') and F.project.like (''' + proj1 ''')'''), 'value1'),
        (text('FPro.status = "ON" and Ab.name.like (''' + abc + ''') and F.project.like (''' + proj2 + ''')'''), 'value2'),
        (text('FPro.status = "OFF" and Ab.name.like (''' + def + ''') and F.project.like (''' + abc + ''')'''), 'value3')]).label (deriver_vals)

query = db.session.query(F)\
        .join(FPro, F.id == FPro.f_id)\
        .join(Ab, Ab.id == F.ab_id).with_entities(FPro.f_id, case_condition,
                                                                    F.f_name, F.proj,
                                                                    FPro.status).subquery()

main_query = db.session.query(Tags).join(query, Tags.tag == query.c.derived_vals).\
        with_entities(query.c.f_id.label('f_id'), query.c.derived_vals.label('derived_vals'), Tags.id.label('tag_id')).all()

以上代码生成如下sql语句：

SELECT anon_1.f_id AS f_id, anon_1.derived_vals AS derived_vals, fw_tags.id AS tag_id 
FROM fw_tags INNER JOIN (SELECT fpro.f_id AS f_id, CASE WHEN FPro.status = "ON" and Ab.name.like (%%abc%%) and F.project.like (%%proj1%%) THEN %(param_1)s WHEN FPro.status = "ON" and Ab.name.like ( %%abc%%) and Flow.project.like ( %%proj2%%) THEN %(param_2)s WHEN FPro.status = "ON" and Ab.name.like (%%def%%) and F.project.like (%%proj1%%) THEN %(param_3)s END AS derived_vals, F.f_name , F.proj AS project,  
FROM F INNER JOIN FPro ON f.id = Fpro.f_id INNER JOIN Ab ON Ab.id = F.ab_id) AS anon_1 ON fw_tags.tag = anon_1.derived_vals]

这正是我想要的查询，但在执行包含上述代码的脚本时出现以下错误：

sqlalchemy.exc.ProgrammingError: (pymysql.err.ProgrammingError) (1064, "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near '(%abc%) and F.project.like (%proj1%) THEN 'value1' WHEN FPro' at line 2")

我猜测错误是附加了 '%'，但我不确定，所以任何关于为什么会发生此错误或可以采取哪些措施来防止添加 % 的帮助将不胜感激。

有没有办法在不使用 text 或 literal() 的情况下做到这一点？

Answer 1

在标准 SQL 中，"ON" 是一个分隔/引用的标识符，而不是（文本）文字，用于标识列、表或其他一些架构对象。请改用单引号：'ON'。一些 DBMS 的模式允许对文字使用双引号，甚至尝试从上下文推断含义，但养成这种习惯可能不是一个好主意。

在 Python 中，'FPro.status ...''' 是 2 个字符串文字 'FPro.status ...' 和 '' 的串联，而不是带有转义单引号的单个文字。

请不要将值连接或以其他方式手动格式化 SQL 查询，您已经发现这很容易出错。从生成的 SQL 中可以明显看出，这些值是按原样连接的，没有正确引用，因此会产生不正确的语句。将值传递给（原始）SQL 查询的正确方法是使用占位符，或者对于 SQLAlchemy，使用 SQL 表达式语言。

使用占位符：

abc = "%abc%"
def_ = "%def%"
proj1 = "%project1%"
proj2 = "%project2%"
# Using **raw SQL** fragments
case_condition = case([
    (text("FPro.status = 'ON' AND Ab.name LIKE :abc AND F.project LIKE :proj1"), 'value1'),
    (text("FPro.status = 'ON' AND Ab.name LIKE :abc AND F.project LIKE :proj2"), 'value2'),
    (text("FPro.status = 'OFF' AND Ab.name LIKE :def_ AND F.project LIKE :abc"), 'value3')
])

query = db.session.query(F)\
    .join(FPro, F.id == FPro.f_id)\
    .join(Ab, Ab.id == F.ab_id)\
    .with_entities(
        FPro.f_id,
        case_condition.label('derived_vals'),
        F.f_name,
        F.proj,
        FPro.status)\
    .subquery()

main_query = db.session.query(Tags)\
    .join(query, Tags.tag == query.c.derived_vals)\
    .with_entities(
        query.c.f_id.label('f_id'),
        query.c.derived_vals.label('derived_vals'),
        Tags.id.label('tag_id'))\
    .params(abc=abc, def_=def_, proj1=proj1, proj2=proj2)\
    .all()

使用表达语言：

from sqlalchemy import and_

abc = "%abc%"
def_ = "%def%"
proj1 = "%project1%"
proj2 = "%project2%"
# Using SQLAlchemy SQL Expression Language DSL **in Python**
case_condition = case([
    (and_(FPro.status == 'ON', Ab.name.like(abc), F.project.like(proj1)), 'value1'),
    (and_(FPro.status == 'ON', Ab.name.like(abc), F.project.like(proj2)), 'value2'),
    (and_(FPro.status == 'OFF', Ab.name.like(def_), F.project.like(abc)), 'value3')
])

query = db.session.query(F)\
    .join(FPro, F.id == FPro.f_id)\
    .join(Ab, Ab.id == F.ab_id)\
    .with_entities(
        FPro.f_id,
        case_condition.label('derived_vals'),
        F.f_name,
        F.proj,
        FPro.status)\
    .subquery()

main_query = db.session.query(Tags)\
    .join(query, Tags.tag == query.c.derived_vals)\
    .with_entities(
        query.c.f_id.label('f_id'),
        query.c.derived_vals.label('derived_vals'),
        Tags.id.label('tag_id'))\
    .all()