Search code examples
apisqlalchemyormfastapisqlmodel

Fetch data from 3 different tables with SQLAlchemy or SQLModel


I want to fetch data from 3 tables with SQLAlchemy or SQLModel. For example lets say that my tables are the following:

class A(SQLModel, table=true):
    id: int
    title: str

class B(SQLModel, table=true):
    id: int
    a_id: foreign_key("a.id")
    name: str
 
class C(SQLModel, table=true):
    id: int
    b_id: foreign_key("b.id")
    text: str

The response that I want to have is the following:

[
    {
        "id": 1,
        "title": "This is A table",
        "b": [
            {
                "id": 1,
                "name": "This is B table",
                "c":[
                   {
                        "id":1,
                        "text": "My text from c" 
                   }               
                ]
            }
        ]
    }
]

I am trying with selectinload but doesn't work

query = (
        select(A)
        .where(A.id == a_id)
        .options(
            selectinload(A.b).joinedload(
                B.c
            )
        )
    )

    try:
        response = (await session.exec(query)).one()
    except NoResultFound:
        raise HTTPException(status_code=404, detail="Data not found")

Thank you in advance


Solution

  • I have tried these models with newer version of sqlalchemy (not sqlmodel) it works as shown in this example (sqlalchemy=1.4.39)

    class A(Base):
        __tablename__ = 'a'
    
        id = Column(Integer, primary_key=True)
        title = Column(String(length=100))
    
        b = relationship("B", backref="a")
    
    class B(Base):
       __tablename__ = 'b'
    
       id = Column(Integer, primary_key=True)
       name = Column(String(length=100))
    
       a_id = Column(ForeignKey('a.id', ondelete='CASCADE'))
    
       c = relationship("C", backref="b")
    
    class C(Base):
       __tablename__ = 'c'
    
       id = Column(Integer, primary_key=True)
       text = Column(String(length=100))
    
       b_id = Column(ForeignKey('b.id', ondelete='CASCADE'))
    

    Query:

    stmt = select(A).options(selectinload(A.b).selectinload(B.c))
    resp = await session.execute(query)
    return resp.scalars().all() 
    

    (Prepare pydantic model related to this query) schema.py

    class CResp(BaseModel):
        id: int
        text: str
    
    class BResp(BaseModel):
        id: int
        name: str
        c: List[CResp]
    
    class AResp(BaseModel):
        id: int
        title: str
        b: List[BResp]
    

    views.py

    @app.get("/", response_model=List[AResp])
    async def index(session=Depends(get_session)):
        query = select(A).options(selectinload(A.b).selectinload(B.c))
        resp = await session.execute(query)
        return resp.scalars().all() 
    

    If you want to use this query with sqlmodel, you can try some other versions of sqlalchemy as mentioned here (SQLModel error: "AttributeError: 'Team' object has no attribute 'heroes'"), sqlmodel has some problems with SQLalchemy 1.4.36+ versions