Get field from postgres record type in SQLAlchemy

I have a query that returns a record, and I need to access a field on this record.

I can use this with postgres (items.properties is a jsonb column):

SELECT (p).key FROM items, jsonb_each(items.properties) p;

How can I make this query in sqlalchemy?

I have this so far:

properties_data = func.jsonb_each(ItemModel.properties).alias("p")
q = db.query(properties_data).select_from(ItemModel)

This produces the following, but I cannot find a way to access key field on p:

SELECT p.jsonb_each_1 AS p_jsonb_each_1 
FROM stac.items, jsonb_each(stac.items.properties) AS p

I should mention I don't want to return values in python and access key there. I'm going to use this query as a subquery in another query.

Solution

What you want is "table valued functions", it shows an example for json_each which can be trivially modified for jsonb_each.

Here is a self contained code snippet running in SQLAlchemy 2 that selects only the key from the result.

The lines of interest are as follows, note that you have to pass joins_implicitly=True to avoid a warning related to cartesian product as suggested by docs

    properties_data = func.jsonb_each(ItemModel.properties).table_valued('key', joins_implicitly=True)
    statement = select(properties_data.c.key).select_from(ItemModel)

from sqlalchemy import create_engine, select, func
from sqlalchemy.orm import Mapped, DeclarativeBase, Session, mapped_column
from sqlalchemy.dialects.postgresql import JSONB


class Base(DeclarativeBase):
    pass


class ItemModel(Base):
    __tablename__ = 'items'
    id: Mapped[int] = mapped_column(primary_key=True)
    properties: Mapped[dict[str, str]] = mapped_column(JSONB)

engine = create_engine('connection string')
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(ItemModel(properties={'foo': '1', 'bar': '2'}))
    session.add(ItemModel(properties={'bax': '3', 'qux': '4'}))
    session.commit()

with Session(engine) as session:
    properties_data = func.jsonb_each(ItemModel.properties).table_valued('key', joins_implicitly=True)
    statement = select(properties_data.c.key).select_from(ItemModel)
    for i in session.scalars(statement):
        print(i)

Output

bar
foo
bax
qux