Error fastapi.exceptions.ResponseValidationError and other relationship and back_population problems

[diegocgaona]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

# I had to divide in two files because of an error related to "clients" table already in metadata.
# It is a minimal, because the models are more complex and I think I would get more errors related to TypeError: issubclass() arg 1 must be a class because of the order of the related tables.

# test.py 
from sqlmodel import Field, SQLModel, Relationship, Column, create_engine, Session, select
from sqlalchemy import DateTime, func
from datetime import datetime
from datetime import date
from typing import List, Any, Dict, Optional


###############################################
# MODELS:
# Related models:
class Jobs(SQLModel, table=True):
    id: Optional[int] = Field(primary_key=True, index=True)
    owner_id: int = Field(nullable=False, foreign_key="usermodel.id")

    expected_pay_date: date
    pay_date: Optional[date]
    start_date: date
    nfe_date: Optional[date]
    value: float
    description: str
    costs: float = Field(default=0)
    is_nfe_issued: bool = Field(default=False)
    estimated_hours: Optional[float]
    nfe_url: str = Field(max_length=250)
    boleto_url: str = Field(max_length=250)
    pix_url: str = Field(max_length=250)
    report_url: str = Field(max_length=250)
    created_at: datetime = Field(default_factory=datetime.utcnow)

    pay_account_id: int = Field(foreign_key="paymentaccount.id")
    pay_type_id: int = Field(foreign_key="paymenttypes.id")
    status_id: int = Field(foreign_key="jobstatus.id")
    client_id: int = Field(foreign_key="clients.id")

    # types: List["JobTypes"] = Relationship(back_populates="jobs",
    #                                        link_model=JobJobTypesLink)  # Not really appear in the table
    client: "Clients" = Relationship(back_populates="jobs")  # Not really appear in the table

    # pay_type: "PaymentTypes" = Relationship(back_populates="jobs")
    # pay_account: "PaymentAccount" = Relationship(back_populates="jobs")

    class Config:
        orm_mode = True
        read_with_orm_mode = True


class ClientNotes(SQLModel, table=True):
    id: Optional[int] = Field(primary_key=True, index=True)
    owner_id: int = Field(nullable=False, foreign_key="usermodel.id")

    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: Optional[datetime] = Field(sa_column=Column(DateTime(), onupdate=func.now()))
    note: str

    client_id: int = Field(foreign_key="clients.id")
    client: "Clients" = Relationship(back_populates="notes")


class ClientObjective(SQLModel, table=True):
    id: Optional[int] = Field(primary_key=True, index=True)
    name: str = Field(unique=True, nullable=False, max_length=60)
    clients: List["Clients"] = Relationship(back_populates="objective")


class ClientTypes(SQLModel, table=True):
    id: Optional[int] = Field(primary_key=True, index=True)
    name: str = Field(unique=True, nullable=False, max_length=60)
    clients: List["Clients"] = Relationship(back_populates="client_type")


class ClientsBase(SQLModel):
    name: str = Field(unique=True, nullable=False, max_length=60)
    description: Optional[str] = None
    owner_id: int = Field(nullable=False, foreign_key="usermodel.id")

    status: bool = Field(default=True)
    cnpj: Optional[str] = Field(max_length=19)
    email: str = Field(max_length=60, nullable=False)
    emails_finance: Optional[str] = Field(max_length=180)
    emails_reports: Optional[str] = Field(max_length=180)
    hubspot_url: Optional[str] = Field(max_length=120)
    proposal_url: Optional[str] = Field(max_length=120)
    industry: Optional[str] = Field(max_length=60)
    discount_period: Optional[int] = None
    discount_perc: Optional[int] = None

    client_type_id: int = Field(default=None, foreign_key="clienttypes.id")
    objective_id: int = Field(default=None, foreign_key="clientobjective.id")


class Clients(ClientsBase, table=True):
    id: Optional[int] = Field(primary_key=True, index=True)

    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: Optional[datetime] = Field(sa_column=Column(DateTime(), onupdate=func.now()))

    jobs: List["Jobs"] = Relationship(back_populates="client", sa_relationship_kwargs={
        "lazy": 'joined'
    })  # Not really appear in the table
    notes: List["ClientNotes"] = Relationship(back_populates="client", sa_relationship_kwargs={
        "lazy": 'joined'
    })  # Not really appear in the table
    objective: "ClientObjective" = Relationship(back_populates="clients", sa_relationship_kwargs={
        "lazy": 'joined'
    })  # Not really appear in the table
    client_type: "ClientTypes" = Relationship(back_populates="clients", sa_relationship_kwargs={
        "lazy": 'joined'
    })  # Not really appear in the table

    class Config:
        orm_mode = True
        read_with_orm_mode = True


class ClientsRead(ClientsBase):
    id: int
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None

    jobs: Optional[List["Jobs"]] = []
    notes: Optional[List["ClientNotes"]] = []
    objective: Optional["ClientObjective"] = None
    client_type: Optional["ClientTypes"] = None


class ClientsCreate(ClientsBase):
    pass


# DB CONFIG
connect_args = {"check_same_thread": False}
engine = create_engine("sqlite:///db.sqlite3", echo=True, connect_args=connect_args)
SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session






####################################
# main.py

import uvicorn
from fastapi import FastAPI, APIRouter, Depends, Query, Response
from fastapi.middleware.cors import CORSMiddleware
from typing import List, Any, Dict, Optional
from sqlmodel import Session, select
from test import Clients, ClientsRead, get_session
from datetime import date
###############################################
# FAST API

router = APIRouter(
    prefix="/clients",
    tags=["Clients"],
    responses={404: {"description": "Not Found"}},
)


# Get many items
@router.get("/", response_model=List[ClientsRead])
async def read_clients(
        *,
        response: Response,
        session: Session = Depends(get_session),
        _start: Optional[int] = 0, _end: Optional[int] = 0,
        _order: Optional[str] = Query(
            None, description="ASC or DESC (Ascending or Descending"),
        _sort: Optional[str] = Query(
            None, description="The field used to order"),
        # user: User = Depends(current_active_user),
        id: Optional[List[int]] = Query(
            None, description="List of ids of the clients"),
        type_of_client: Optional[str] = Query(
            None, description="The type of recurrence: Monthly, Freelance"),
        email: Optional[str] = Query(
            None, description="Main email address of the client"),
        cnpj: Optional[str] = Query(
            None, description="CNPJ of the client"),
        name: Optional[str] = Query(
            None, description="Name of the client"),
        objective: Optional[str] = Query(
            None, description="The main objective of the client: Leads, Sales, Others..."),
        creation_date_start: Optional[date] = Query(
            None, description="Initial interval of creation date, format: YYY-MM-DD"),
        creation_date_end: Optional[date] = Query(
            None, description="End interval of creation date, format: YYY-MM-DD"),
        industry_type: Optional[str] = Query(
            None, description="The main industry of the client"),
        status: Optional[bool] = Query(
            None, description="Is the client active or inactive?"),
):
    # NEED TO ADAPT FROM TORTOISE ORM TO SQLMODEL (How? It accepts strings for fields?)
    filters: Dict[Any, Any] = {"id__in": id,
                               "client_type__name__icontains": type_of_client,
                               "email__icontains": email,
                               "cnpj__icontains": cnpj,
                               "name__icontains": name,
                               "objective__name__icontains": objective,
                               "created_at__gte": creation_date_start,
                               "created_at__lte": creation_date_end,
                               "industry__icontains": industry_type,
                               "status": status}
    used_filters = {key: value for key, value in filters.items() if value is not None}

    order_dict: Dict[str] = {"ASC": "", "DESC": "-"}
    if _order and _sort:
        order_str: str = f"{order_dict.get(_order)}{_sort}".lower()
    else:
        order_str: str = "id"

    if _end:
        client_list = session.exec(select(Clients).offset(_start).limit(_end - _start).order_by(order_str)).all()
        print(client_list)
    else:
        client_list = session.exec(select(Clients).offset(_start)
                                   .order_by(order_str)
                                   # .where(Clients.owner_id == "5a0454b3-4725-4fce-adb2-f04109a03699")
                                   ).first()

        print(client_list.dict)
        print(client_list.json())
        print(client_list.jobs)

    return client_list


app = FastAPI()

app.include_router(router)

# For CORS support, the allowed origins
origins = [
    "http://localhost",
    "http://localhost:8080",
    "http://localhost:3000",
    "http://localhost:3000/",
    "http://localhost:8000",
]
# CORS options
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["X-Total-Count"],
)

if __name__ == "__main__":
    uvicorn.run("main:app", host="localhost", port=8000, log_level="info", reload=True)

Description

First, if I try to use my code with the @router.get("/", response_model=List[ClientsRead]), I get an error:

File "/home/diego/python/financeiro/venv/lib/python3.11/site-packages/fastapi/routing.py", line 291, in app
    content = await serialize_response(
              ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/diego/python/financeiro/venv/lib/python3.11/site-packages/fastapi/routing.py", line 154, in serialize_response
    raise ResponseValidationError(
fastapi.exceptions.ResponseValidationError

I don't have any idea what is the field or problem with my schemas, I tried to follow the way the tutorial show, but I think I missed something because of this error.

If I try @router.get("/") (without response_model) my response in the /docs not show the related fields notes or jobs, (and I had records related, I see using sqlite browser) the other fields are fine.

In my print(client_list.dict) or print(client_list.jobs) I can see the related jobs for example. But in my print(client_list.json()), the
jobs field do not appears.

What I'm doing wrong? Why my response_model does not work and why I can't see the related data in my response/json?
I tried to found solutions, but I couldn't find a really similar issue, and I tried some things I found, but without success...

Another question (less important, is a doubt) I was using Tortoise ORM and now I'm trying to migrate to SQLModel and I was using the filters as above, they came as parameters in my url from my front end and I was using a dict and strings to make the filters. It is possible to do something this way with SQLModels? I can filter using strings? O another way to filter using the query parameters that are not None or that appears in my request?

My best regards, Diego.

Operating System

Linux

Operating System Details

Ubuntu 20.4 under WSL on Windows 11

SQLModel Version

0.08

Python Version

3.11.4

Additional Context

Other packages version:
SQLAlchemy: 1.4.41
FastAPI: 0.101.0
Pydantic: 1.10.12
Uvicorn: 0.23.2
Starlette: 0.27.0

[diegocgaona]

I still didn't found a solution... somebody have any idea?

[JonLev]

Hi @diegocgaona,
I faced the same issue, and actually the error is kinda of wrong I think: it reports you an error in the response model but I think fundamentally the model is not complete... yet !
From my side, it's due to the fact that my app is async, so the loading of relationship is probably async too. Therefore, the examples of the doc can't work because it miss an await at some point. There is a little insight in this article: https://dev.to/arunanshub/the-async-side-of-sqlmodel-relationships-part-2-4ebc

I fixed the error by eager loading what I needed in the query itself before returning the object. So adding to your query sqlAlchemy options with joinedload even if you defined your lazy options in relationships.

[diegocgaona]

Thanks!! I will see to it, but I think I will wait to do a production implementation of the library until it is complete.