Search code examples
pythonsqlalchemyorm

Is there a way to manually create an ORM object in a persistent state?


I want to create an persistent ORM object from the some data i already have and deferred columns for everything else.

Simplified example - i have some code that loads data from several tables with multiple queries:

def my_func(
    session: Session, country_code: str, user_email: str
) -> tuple[Country | None, User | None]:
    country = session.scalar(
        select(Country)
        .where(Country.code == country_code)
        .options(load_only(Country.id))
    )

    user = session.scalar(
        select(User)
        .where(User.email == user_email)
        .options(load_only(User.id, User.name))
    )

    return country, user

and i'd like to rewrite it with UNION ALL without changing interface:

def my_func(
    session: Session, country_code: str, user_email: str
) -> tuple[Country | None, User | None]:
    country_query = (
        select(1, Country.id, None)
        .where(Country.code == country_code)
    )
    user_query = (
        select(2, User.id, User.name)
        .where(User.email == user_email)
    )
    result = session.scalars(union_all(country_query, user_query)).all()

    country_data = next((i[1] for i in result if i[0] == 1), None)
    user_data = next((i[1] for i in result if i[0] == 2), None)

    # somehow create ORM objects with received data and all other columns as deferred
    country = ... if country_data else None
    user = ... if user_data else None
    return country, user

The calling code relies on objects being there, and i'd like to avoid changing it.


Solution

  • Based on @GordThompson's solution, I ended up doing something like this (in a more generic way though):

    def my_func(
        session: Session, country_code: str, user_email: str
    ) -> tuple[Country | None, User | None]:
        col_ids = func.unnest(cast([0, 1], ARRAY(BigInteger))).alias()
        q = (
            select(
                col_ids.column,
            ).select_from(col_ids)
        )
        q = q.outerjoin(
            User,
            and_(col_ids.column == 0, User.email == user_email)
        ).add_columns(User)
        q = q.outerjoin(
            Country,
            and_(col_ids.column == 1, Country.code == country_code)
        ).add_columns(Country)
        
        result = {key: next(m for m in models if m) for key, *models in session.execute(q) if any(models)}
        return result.get(1), result.get(0)
        
    

    The idea is to make a virtual root table with unnest() and then left join other tables to it. With many joined tables, the result set will contain a lot of mostly empty columns, but for retrieving a small amount of rows it's fine, I think. Postgresql seems to handle such query pretty well.