Correct way to map Python classes to Trino tables

[benrifkind]

I think this is related to using SQLAlchemy but I could not get my table mappings to work properly. My thought is that it should be possible to map a Python DataClass on top of a Trino (Iceberg) table.

So for example, suppose I have these two classes

from dataclasses import dataclass

@dataclass
class User:
    id: int
    name: str


@dataclass
class Purchase:
    user: User
    item_id: int
    cost: float

The idea is that the Purchases class corresponds to an Iceberg table whose schema looks like

 CREATE TABLE iceberg.scratch.purchases (
    user ROW(id integer, name varchar),
    item_id integer,
    cost real
 )

I would like to be able to insert an instance of the Purchase class into the Iceberg purchases table. And to be able to fetch a row from the Iceberg table as an instance of the Purchase class. Of course this functionality is not meant to be specific for just these classes. It should work for any Python object as long as the table schema matches the Python object schema.

I was able to write this custom functionality myself but I can't help thinking that this must already exist. I tried using SQLAlchemy but I ran into issues since Iceberg does not support a primary key and SQLAlchemy requires it. Even when I work around this (by creating the table separately from SQLAlchemy), I am unsure how to define a SQLAlchemy table that has a column that is itself a class. The user column must be defined as a tuple and returns as a tuple.

So for example, if I define this SQLAlchemy table

class Purchase(DeclarativeBase):
    __tablename__ = "purchases"

    item_id: Mapped[Optional[int]] = mapped_column(Integer, primary_key=True)
    user: User = Column(ROW(attr_types=[("id", Integer), ("name", String)]))
    cost: float = Column(FLOAT)

The following attempt to insert data fails with:

(trino.exceptions.NotSupportedError) Query parameter of type '<class 'main.User'>' is not supported.

from sqlalchemy.orm import Session

with Session(create_iceberg_engine(schema="scratch")) as session:
    purchases = [
        Purchase(item_id=1, user=User(name="foo", value=1), cost=1.5),
        Purchase(item_id=2, user=User(name="bar", value=2), cost=2.2),
        ]
    session.add_all(purchases)

And this works but outputs a tuple

with Session(create_iceberg_engine(schema="scratch")) as session:
    purchases = [
        Purchase(item_id=1, user=(1, "foo"), cost=1.5),
        Purchase(item_id=2, user=(2, "bar"), cost=2.2),
        ]
    session.add_all(purchases)
    results = session.query(Purchase).all()
    for result in results:
        print(result.user)

The outputs are

(1, 'foo')
(2, 'bar')

If I define a new type function as below and define the SQLAlchemy table using it, I get what I want, but this seems like a complicated way to get what I want. I also could not get it to work if I want a column to be a list of objects which in my mind, in Trino is an array of ROW's

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.types import TypeDecorator


def dataclass_type(data_class: type) -> type[TypeDecorator]:
    attr_types = [(k, mapping[v]) for k, v in data_class.__annotations__.items()]
    fields = [k for k in data_class.__annotations__.keys()]

    class DataClassType(TypeDecorator):
        impl = ROW(attr_types=attr_types)
        cache_ok = True

        def process_bind_param(self, value, dialect):
            return tuple(getattr(value, k) for k in fields)

        def process_result_value(self, value, dialect):
            return data_class(*value)

    return DataClassType

class Purchase(DeclarativeBase):
    __tablename__ = "purchases"

    item_id: Mapped[Optional[int]] = mapped_column(Integer, primary_key=True)
    # I define the type using the dataclass_type function defined above
    user: User = Column(dataclass_type(User))
    cost: float = Column(FLOAT)

I hope my question makes sense. I'm just looking for the simplest way to map Python objects to Trino tables. And likely using the ORM doesn't make sense for my use case since my tables are not relational. Thanks for any help!