SQLModel: Many-to-many relationship data dropped when paginating

[thebleucheese]

This is similar to #294

However, when using SQLModel the relationship data is dropped. I'm always getting null for tags in my FastAPI response.

The queries seem fine. If I remove pagination, a simple unpaginated List of CustomerViews contains the tags. Maybe I'm doing something wrong with the model setup?

Excuse typos / weirdness. I had to pull this out of a large project and pare it down to make it readable.

# Relevant schema classes ----------------------------------------------

class CustomerViewToTagBase(SQLModel, TimeStampMixin):
    customer_view: int = Field(foreign_key="customer_view.id", primary_key=True)
    tag: int = Field(foreign_key="tag.id", primary_key=True)

class CustomerViewToTag(CustomerViewToTagBase, table=True):
    __tablename__ = "customer_view_to_tag"

class TagBase(SQLModel):
    name: str

class Tag(TagBase, table=True):
    __tablename__ = "tag"
    id: Optional[int] = Field(default=None, primary_key=True)
    customer_views: List["CustomerView"] = Relationship(back_populates="tags", link_model=CustomerViewToTag)

class CustomerViewBase(SQLModel):
    __tablename__ = "customer_view"
    name: Optional[str] = Field(default=None)

class CustomerView(CustomerViewBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    tags: List["Tag"] = Relationship(back_populates="customer_views", link_model=CustomerViewToTag)

# For FastAPI
class CustomerViewWithTag(CustomerViewBase):
    id: int
    tags: Optional[List[Tag]]

    # I have no idea if this is necessary, I added it based on the linked issue above #294 
    class Config:
        orm_mode = True


# Router ------------------------------------------------------------------------

@router.get("/withtags", response_model=Page[CustomerViewWithTag], status_code=200)
async def get_customer_views_with_tags(
        db_session: AsyncSession = Depends(get_db_session),
        params: Params = Depends(),
        filter: Optional[str] = None
    ):
    async with db_session as db:
        customerviews_query = select(CustomerView).order_by(CustomerView.id).options(selectinload(CustomerView.tags))

        if filter is not None:
            customerviews_query = customerviews_query.filter(
                    CustomerView.name.ilike('%' + filter + '%')
            )

        # Switching the response_model to a List and the following code makes the relationship work properly
        # res = await db.execute(customerviews_query)
        # return res.scalars().all()
        return await paginate(session=db, query=customerviews_query, params=params)

[uriyyo]

Hi @thebleucheese,

I will take a closer look

[thebleucheese]

Hi @thebleucheese,

I will take a closer look

Thanks!

I'll try to dig into it some more this weekend to see if I can find a fix unless you beat me to it.

PS - I love the project, it's helped a lot with my first foray into using FastAPI

[uriyyo]

Hi @thebleucheese,

It will be great if you can provide a minimal working example that I can locally. Currently, I can't help with code that you provided(

[thebleucheese]

Hey, here's an example. You can see the difference between the List and Page endpoints.

pyproject.toml

[tool.poetry]
name = "fastapi_sqlmodel_pagination_dbg"
version = "0.1.0"
description = ""
authors = ["thebleucheese <thebleucheese@wedgesalad>"]
license = "MIT"

[tool.poetry.dependencies]
python = "^3.8"
fastapi = "^0.78.0"
# NOTE: known SQLModel bug in latest SQLAlchemy with relationships, hardcode older dep ver
SQLAlchemy = "1.4.34"
sqlmodel = "^0.0.6"
# no idea what I'm doing with these extras but this seems to work
fastapi-pagination = {extras = ["SQLAlchemy", "SQLModel", "Async_SQLModel", "Async_SQLAlchemy"], version = "^0.9.3"}
uvicorn = "^0.17.6"

[tool.poetry.dev-dependencies]

[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

main.py

import os
from typing import List, Optional

from fastapi import FastAPI, Depends
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select
from sqlalchemy.orm import selectinload

from fastapi_pagination import Page, Params
from fastapi_pagination.ext.sqlmodel import paginate


class CustomerViewToTagBase(SQLModel):
    customer_view: int = Field(foreign_key="customer_view.id", primary_key=True)
    tag: int = Field(foreign_key="tag.id", primary_key=True)

class CustomerViewToTag(CustomerViewToTagBase, table=True):
    __tablename__ = "customer_view_to_tag"

class TagBase(SQLModel):
    name: str

class Tag(TagBase, table=True):
    __tablename__ = "tag"
    id: Optional[int] = Field(default=None, primary_key=True)
    customer_views: List["CustomerView"] = Relationship(back_populates="tags", link_model=CustomerViewToTag)

class CustomerViewBase(SQLModel):
    __tablename__ = "customer_view"
    name: Optional[str] = Field(default=None)
    description: Optional[str] = Field(default=None)

class CustomerView(CustomerViewBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    tags: List["Tag"] = Relationship(back_populates="customer_views", link_model=CustomerViewToTag)

# For FastAPI
class CustomerViewWithTag(CustomerViewBase):
    id: int
    tags: Optional[List[Tag]]


if os.path.exists("database.db"):
  os.remove("database.db")
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url, echo=True)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def create_data():
    with Session(engine) as session:
        view_recruiter = CustomerView(name="Recruiter", description="Interviewer view")
        view_interviewer = CustomerView(name="Interviewer", description="Recruiter view")
        view_manager = CustomerView(name="Manager", description="Manager view")

        tag_resume = Tag(name="Resume",
            customer_views=[view_recruiter, view_interviewer],
        )
        tag_salary = Tag(name="Salary",
            customer_views=[view_recruiter],
        )

        session.add(view_recruiter)
        session.add(view_interviewer)
        session.add(view_manager)
        session.commit()

        session.refresh(view_recruiter)
        session.refresh(view_interviewer)
        session.refresh(view_manager)

        print("Recruiter:", view_recruiter)
        print("Recruiter tags:", view_recruiter.tags)
        print("Interviewer:", view_interviewer)
        print("Interviewer tags:", view_interviewer.tags)
        print("Manager:", view_manager)
        print("Manager tags:", view_manager.tags)



app = FastAPI()

@app.on_event("startup")
def on_startup():
    create_db_and_tables()
    create_data()


@app.get("/views/")
def read_views():
    with Session(engine) as session:
        views = session.exec(select(CustomerView)).all()
        return views

@app.get("/views_with_tags_list/", response_model=List[CustomerViewWithTag], status_code=200)
def read_views_with_tags_as_list():
    with Session(engine) as session:
        views = session.exec(select(CustomerView).order_by(CustomerView.id).options(selectinload(CustomerView.tags))).all()
        return views

@app.get("/views_with_tags_paginated/", response_model=Page[CustomerViewWithTag], status_code=200)
def read_views_with_tags_paginated(
         params: Params = Depends()
    ):
    with Session(engine) as session:
        customerviews_query = select(CustomerView).order_by(CustomerView.id).options(selectinload(CustomerView.tags))
        result = paginate(session=session, query=customerviews_query, params=params)
        return result

run with uvicorn main:app

Haven't had a chance to get a local dev version of fastapi-pagination to do a deeper debug but I'll try that when I'm able to get back to this project.

[thebleucheese]

@uriyyo

There's a really simple work-around to get this working. If you convert the pagination result to a dict before returning from the route, it works:

@app.get("/views_with_tags_paginated/", response_model=Page[CustomerViewWithTag], status_code=200)
def read_views_with_tags_paginated(
         params: Params = Depends()
    ):
    with Session(engine) as session:
        customerviews_query = select(CustomerView).order_by(CustomerView.id).options(selectinload(CustomerView.tags))
        result = paginate(session=session, query=customerviews_query, params=params)
        return dict(result)

I have a feeling this is a FastAPI issue, but I'm not familiar enough with FastAPI's internals to make a definitive call. The problem appears to be related to FastAPI's jsonable_encoder function. I've described some testing and debugging below.

It could be related to this or some of the other similar issues:
tiangolo/fastapi@d79c13d

...

Debugging Method (For posterity, in case anyone cares)
I debugged this by locally checking out the fastapi-pagination project and adding it as a local dev dependency so I could step-through or modify fastapi-pagination easily and test changes. I verified that nothing in the Page object creation was acting strangely.

I then moved over to my sample code, where I converted result to a dict and printed debug statements in the FastAPI router. At this point I was stumped and didn't find any issues, so I went looking through FastAPI and SQLModel issues and saw references to jsonable_encoder when working with nested objects. The problems didn't exactly match what I was seeing but there's a note at the bottom of this page: https://fastapi.tiangolo.com/tutorial/encoder/?h=jsonabl which states that jsonable_encoder is used internally.

I figured this function may be used to generate FastAPI json output so I tested by including from fastapi.encoders import jsonable_encoder and calling jsonable_encoder on result. I then print outputted the json encoded result object and noticed it was dropping my tags - exactly the problem I saw when directly returning the paginate function's Page object with nested SQLModel relationship objects.

[uriyyo]

@thebleucheese Thanks for such a detailed investigation. I will close this ticket. Please, reopen it in case you still have some issues.