How to properly handle join results from SQLAlchemy with Pydantic/FastAPI

user_3510043 0
Viewed 3453

I would like a piece of advice on handling the result of a join operation performed in SQLAlchemy and do the serialization with Pydantic (in FastAPI).

If I am not mistaken, the result of the join on two table leads to a list of tuples of SQLAlchemy models. Here is a mock for it, like_a_join being my understanding for the result of a join query.

from pydantic import BaseModel
from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class A(BaseModel):
    a: int

    class Config:
        orm_mode = True


class B(BaseModel):
    b: int

    class Config:
        orm_mode = True


class Am(Base):
    __tablename__ = "A"
    a = Column(Integer, primary_key=True, nullable=False)


class Bm(Base):
    __tablename__ = "B"
    b = Column(Integer, primary_key=True, nullable=False)


def like_a_join():
    return [(Am(a=1), Bm(b=1))]

While it is possible to pass a model object to Pydantic with from_orm for simple query (which is done for example on the FastAPI Postgres cookie cutter, here, this is not obvious to me how to best handle the join/tuple case.

Would it make sense to create a class for handling the tuple like the following?

from typing import List
from pydantic import parse_obj_as
from dataclasses import dataclass

@dataclass
class Cm:
    a: Am
    b: Bm

class C(BaseModel):
    a: A
    b: B

    class Config:
        orm_mode = True

def like_a_join_with_class():
    return [Cm(a=Am(a=1), b=Bm(b=1))]

print(parse_obj_as(List[C], like_a_join_with_class()))

Would it be better to use a dictionary?

def like_a_join_with_dict():
    return [{"a": Am(a=1), "b": Bm(b=1)}]

The idea behind would be to include the query result in a FastAPI endpoint with serialization handled automatically.

Thanks in advance for your help.

pydantic fastapi sqlalchemy python
1 Answers

It is always good to have a pydantic model to handle and manipulate Your data, but it is especially practical when You want to return such join from an endpoint. In such case You can use a pydantic model to cover a path operation's response_model argument:

@router.get("/some_endpoint_path", response_model=SomePydanticModel)
def request_handler():
    ...

If this is Your case, then I would create a model that covers a whole query result (i.e. list of 2-tuple [A, B]). If ORM Am maps to pydantic A, and ORM Bm maps to pydantic B, then whole query result should successfully map to the following pydantic model:

import typing as t

class JoinResult(BaseModel):
    results: t.List[t.Tuple[A, B]]

    class Config:
        orm_mode = True

The exact form of sqlalchemy query depends on the sqlalchemy version You are using and the abstraction level You are choosing, but for modern 2.x style queries with async ORM session, it would look (more or less) like this:

statement = select(Am, Bm).join(<your_join_here>).where(<your_condition_here>)
result = await session.execute(statement)
scalar_results = result.scalars()

This scalar_results should be list-like object containing 2-tuples holding instances of sqlalchemy models Am and Bm. You should be able to parse it as follows:

jr = JoinResult(results=iter(scalar_results))

You can then return jr directly from Your path operation function.

Alternatively, if You really want to return query results directly, You can try the new project from tiangolo: SQLModel. This is a missing bridge between pydantic and sqlalchemy. It may bring You some ideas and solutions. But keep in mind that SQLModel is in very early stage of development, potentially with many bugs. Report them however, tiangolo will be happy :)

user_107437420
Related