如何在 Python 中将（有些复杂的）Postgresql 语句转换为 SQLAlchemy ORM？

Posted 2023-02-16

【中文标题】如何在 Python 中将（有些复杂的）Postgresql 语句转换为 SQLAlchemy ORM？

【英文标题】：How to convert a (somewhat complex) Postgresql statement to SQLAlchemy ORM in Python?

【发布时间】：2021-06-26 10:22:13

【问题描述】：

我接到了一项我觉得有点难以处理的任务，我正在寻找一些指导，包括如何解决我的问题以及在哪里可以找到有关如何解决类似问题的类似信息。

所以，我的任务是通过将相对复杂的查询转换为 SQLAlchemy 代码来使用 SQLAlchemy ORM 查询数据库。所以不要做一个 connection.execute("select a,b,c as sea from A,B,C where...") 编辑：我有一个到 db_session 对象的连接，我应该做类似的事情 db_session.query(B.email,B.tel,A.createdC.a_name.label("C")).join(B, end_(B.id=A.B_id...).filter(wf_id=db_session.query(wf.id.filter(id='12345'))等

下面的代码部分引用了我必须转换的 postgres 查询。 我在以下方面遇到了特别的麻烦： cast(cfields as json)->>'name'

left join C on C.id=A.C_id

这是代码。如果您能帮助我构建代码，尤其是指导我找到一个资源，让初学者了解更多关于 SQLAlchemy ORM 的信息，我将不胜感激！

select B.email,B.tel,A.created,C.a_name as C,result,cfields ,
       cast(cfields as json)->>'name' as gname
from A
         join B on B.id=A.B_id
         left join C on C.id=A.C_id
where wf_id=(select id from wf where uuid=`12345`)
and c4_id=(select id from c4 where uuid=`54321`)
  and A.result='complete'
order by A.created asc;

更新1： 到目前为止，我已经到了这一点：

db_session.query(B.email, B.tel, A.created, C.a_name.label("C"),
                 A.result, A.cfields, A.cfields['name'].label("gname")).join(
    A, and_(A.B_id == B.id).join(
        C, C.id == A.C_id, isouter=True).filter(
        A.wf_id == (db_session.query(wf.id).filter(wf.uuid == '12345'))
    ).order_by(A.created)
)

我目前的问题是 A.cfields['name'].label("gname") 不能用作 cast(cfields as json)->>'name' as gname 的转换 关于如何使用它的任何想法？

非常感谢您的宝贵时间， 比尔

【问题讨论】：

两件事。后记号在 Postgresql 中不起作用。但更重要的是，你为什么要这样做？为什么要将一个体面且有效的、声明性且易于阅读的 SQL 脚本转换为专有命令式调用的神秘序列？

啊，谢谢你的反引号，忘记了！至于 ORM，有人告诉我它更安全，并且与当前代码库更好地集成。习惯使用 SQLA ORM 对我来说也是一个很好的练习，因为我刚刚加入该项目，需要熟悉代码库。

【参考方案1】：

好的，经过大量的反复试验、谷歌搜索和阅读文档，我找到了解决方案，我将在此处发布它以防其他人需要它。

因此，将我查询的属性转换为 JSON 并获取特定值的答案是首先从 sqlalchemy 导入 cast 和 JSON，然后在尝试通过给它一个键来获取值之前对其进行转换。

所以，这就是我所做的：

from sqlalchemy import cast, JSON

db_session.query(B.email, B.tel, A.created, C.a_name.label("C"),
                 A.result, A.cfields, cast(A.cfields,JSON)['name'].label("gname")).join(
    A, and_(A.B_id == B.id).join(
        C, C.id == A.C_id, isouter=True).filter(
        A.wf_id == (db_session.query(wf.id).filter(wf.uuid == '12345'))
    ).order_by(A.created)
)