python-3.x - SQLAlchemy:在 SELECT 子查询中引用标签

Question

我想弄清楚如何在 SQLAlchemy 中复制以下查询

SELECT c.company_id AS company_id,
  (SELECT policy_id FROM associative_table at WHERE at.company_id = c.company_id) AS policy_id_ref,
  (SELECT `default` FROM policy p WHERE p.policy_id = policy_id_ref) AS `default`,
FROM company c;

请注意，这是我实际处理的一个简单的基本示例。实际模式支持数据和关系版本控制，这需要子查询包含额外的条件、排序和限制，这使得它们成为连接是不切实际的(如果不是不可能的话)。

问题的症结在于第二个子查询如何依赖 policy_id_ref —— 从第一个子查询获得的值。在 SQLAlchemy 中，这实际上是我现在拥有的:

ct = aliased(classes.company)
at = aliased(classes.associative_table)
pt = aliased(classes.policy)

policy_id_ref = session.query(at.policy_id).\                       
    filter(at.company_id == ct.company_id).\
    label('policy_id_ref')             

policy_default = session.query(pt.default).\
    filter(pt.id == 'policy_id_ref').\              
    label('default')

query = session.query(ct.company_id,policy_id_ref,policy_default)

“company”表的拉取与检索“policy_id_ref”列的第一个子查询一样工作正常。问题是必须引用该“policy_id_ref”列的第二个子查询。我不知道如何编写它的过滤器，使其在结果查询中按字面意思呈现“policy_id_ref”，以匹配第一个子查询的标签。

建议？

提前致谢

Answer 1

您可以将查询写成

select(
   Companies.company_id,
   AssociativeTable.policy_id.label('policy_id_ref'),
   Policy.default.label('policy_default'), 
).select_from(
   Companies,
).join(
   AssociativeTable,
   AssociativeTable.company_id == Companies.company_id,
).join(
   Policy,
   AssociativeTable.policy_id == Policy.id  
)

但如果您需要从子查询中引用标签 => 使用 literal_column

from sqlalchemy import func, select, literal_column

session.query(
    func.array_agg(
        literal_column('batch_info'),
        JSONB
    ).label('history')
).select_from(
    select(
        func.jsonb_build_object(
            'batch_id', AccountingQueueBatch.id,
            'batch_label', AccountingQueueBatch.label,
        ).label('batch_info')
    ).select_from(
        AccountingQueueBatch,
    )
)