SQLAlchemy查询构造

Question

我正在使用一个由Postgres支持的Key模型，这是一个用于保存API键的通用表:

class Key(Model):
__tablename__ = "keys"
id = Column(Integer, primarykey=True)
user_id = Column(Integer, ForeignKey("users.id"))
brokerage_id = Column(Integer, ForeignKey("brokerages.id"))
account_id = Column(Integer, ForeignKey("accounts.id"))
key = Column(String(128))
value = Column(String(128))

在下面的示例中，user 2有三个键。三者均与brokerage 2和account 2相关联。它由id 4到6表示。对于这个站点，用户有一个身份验证令牌和两个查询id。

id  user_id brokerage_id    account_id      key         value
--------------------------------------------------------------------
4   2       2               2               token       999999999999
5   2       2               2               query_id    888888      
6   2       2               2               query_id    777777      
7   1       2               3               token       444444444444

我正在尝试构建一个查询，以便我的结果将被建模如下:

[(user_id, brokerage_id, account_id, token, [query_id_1, query_id_2, ...]), ...]

所以对于上面的例子，它看起来像这样

[(2, 2, 2, 999999999999, [888888, 777777]), (1, 2, 3, 444444444444, [])]

我有以下查询选择令牌和query_ids

tokens = db.session.query(
Key.user_id, Key.brokerage_id, Key.account_id, Key.value
).filter(Key.key=='token').all()
query_ids = db.session.query(
Key.user_id, Key.brokerage_id, Key.account_id, Key.value
).filter(Key.key=='query_id').all()

我试过以各种方式使用subquery，但不能完全得到我需要的输出。我如何构建一个查询，以一种与上面的元组列表对齐的方式返回结果?

结果

通过@rfkortekaas

添加最终的工作查询

from sqlalchemy.orm import aliased
from sqlalchemy import func, and_
from project.models import Key
from project.extensions import db
key_token = aliased(Key)
q = db.session.query(
key_token.user_id,
key_token.brokerage_id,
key_token.account_id,
key_token.value.label('token'),
func.array_agg(Key.value).label('query_ids')
).join(
Key,
and_(
key_token.user_id == Key.user_id,
key_token.brokerage_id == Key.brokerage_id,
key_token.account_id == Key.account_id,
Key.key == 'query_id'
)
).filter(
key_token.key == 'token'
).group_by(
key_token.user_id,
key_token.brokerage_id,
key_token.account_id,
key_token.value
)
results = q.all()

Answer 1

您可以使用PostgreSQL中的array_agg函数来创建一个结果数组:

from sqlalchemy.orm import aliased
key_token = aliased(Key)
stmt = select(key_token.user_id,
key_token.brokerage_id,
key_token.account_id,
key_token.value.label('token'),
func.array_agg(Key.value).label('query_ids')
).join(Key,
and_(key_token.user_id == Key.user_id,
key_token.brokerage_id == Key.brokerage_id,
key_token.account_id == Key.account_id,
Key.key == 'query_id'))
.where(key_token.key == 'token')
.group_by(key_token.user_id,
key_token.brokerage_id,
key_token.account_id,
key_token.value)
keys = session.execute(stmt).all()
for row in keys:
print(row)

结果:

<表类>user_idbrokerage_idaccount_id标记query_idstbody><<tr>123"44">[' 4 ']221' 33">[' 6 ']222"99">[' 8 ', ' 7]