Getting nested (joined) tables to display in the OpenAPI interface provided by FastAPI and SQLModel

Question

I'm having trouble understanding how to display the children data in a one-to-many relationship using FastAPI and SQLModel. I'm using Python 3.10.3, FastAPI version 0.78.0 and SQLModel version 0.0.6. Here's a simplified version of the parent/child database models:

from datetime import datetime
from email.policy import default
from sqlalchemy import UniqueConstraint
from sqlmodel import Field, SQLModel, Relationship

class CustomerBase(SQLModel):
    __table_args__ = (UniqueConstraint("email"),)

    first_name: str
    last_name: str
    email: str
    active: bool | None = True

class Customer(CustomerBase, table=True):
    id: int | None =Field(primary_key=True, default=None)

class CustomerCreate(CustomerBase):
    pass

class CustomerRead(CustomerBase):
    id: int

class CustomerReadWithCalls(CustomerRead):
    calls: list["CallRead"] = []

class CallBase(SQLModel):
    duration: int
    cost_per_minute: int | None = None
    customer_id: int | None = Field(default=None, foreign_key="customer.id")
    created: datetime = Field(nullable=False, default=datetime.now().date())

class Call(CallBase, table=True):
    id: int | None = Field(primary_key=True)

class CallCreate(CallBase):
    pass

class CallRead(CallBase):
    id: int

class CallReadWithCustomer(CallRead):
    customer: CustomerRead | None

Here is the API Route:

from fastapi import APIRouter, HTTPException, Depends, Query
from rbi_app.crud.customer import (
    get_customers,
    get_customer,
)
from rbi_app.models import (
    CustomerRead,
    CustomerReadWithCalls,
)
from rbi_app.database import Session, get_session

router = APIRouter()

@router.get("/customers/", status_code=200, response_model=list[CustomerRead])
def read_customers(
    email: str = "",
    offset: int = 0,
    limit: int = Query(default=100, lte=100),
    db: Session = Depends(get_session)
):
    return get_customers(db, email, offset=offset, limit=limit)

@router.get("/customers/{customer_id}", status_code=200, response_model=CustomerReadWithCalls)
def read_customer(id: int, db: Session = Depends(get_session)):
    customer = get_customer(db, id)
    if customer is None:
        raise HTTPException(status_code=404, detail=f"Customer not found for {id=}")
    return customer

And here are the queries to the database the API Route endpoints make:

from sqlmodel import select
from rbi_app.database import Session
from rbi_app.models import (
    Customer,
    CustomerCreate,
)
# from rbi_app.schemas.customer import CustomerCreate
    
def get_customer(db: Session, id: int):
    return db.get(Customer, id)
    
def get_customers(db: Session, email: str = "", offset: int = 0, limit: int = 100):
    if email:
        return db.exec(select(Customer).where(Customer.email == email)).first()
    return db.exec(select(Customer).offset(offset).limit(limit).order_by(Customer.id)).all()

When I navigate to a route to get all a customer my query runs and I get a customer, but there is no "calls" list attribute in the customer. The OpenAPI display shows a "calls" attribute, but it's empty.

What am I doing wrong?

Answer 1

The issue here seems to be that you did not define the relationship on the Customer model (or the Call module). Since you query the database with the Customer model and it has no calls attribute, none of that data is present in the object returned by your get_customer function.

Even though your route defines the CustomerReadWithCalls as a response model, upon calling it, the object of that class can only ever be instantiated with the data returned by your route handler function, which is your Customer instance in this case. Since that does not even have the calls attribute (let alone the data), the CustomerReadWithCalls object is essentially created with the default value that you defined for the calls field -- the empty list.

Adding

    calls: list["Call"] = Relationship(back_populates="customer")

to your Customer model should be enough.

(But as a side note, for me the route documentation only works properly, when I explicitly update the references on the CustomerReadWithCalls model after the CallRead definition.)

Here is a full working example.

models.py

from datetime import datetime

from sqlalchemy import UniqueConstraint
from sqlmodel import Field, Relationship, SQLModel


class CustomerBase(SQLModel):
    __table_args__ = (UniqueConstraint("email"),)

    first_name: str
    last_name: str
    email: str
    active: bool | None = True


class Customer(CustomerBase, table=True):
    id: int | None = Field(primary_key=True, default=None)

    calls: list["Call"] = Relationship(back_populates="customer")


class CustomerCreate(CustomerBase):
    pass


class CustomerRead(CustomerBase):
    id: int


class CustomerReadWithCalls(CustomerRead):
    calls: list["CallRead"] = []


class CallBase(SQLModel):
    duration: int
    cost_per_minute: int | None = None
    customer_id: int | None = Field(default=None, foreign_key="customer.id")
    created: datetime = Field(nullable=False, default=datetime.now().date())


class Call(CallBase, table=True):
    id: int | None = Field(primary_key=True, default=None)

    customer: Customer | None = Relationship(back_populates="calls")


class CallCreate(CallBase):
    pass


class CallRead(CallBase):
    id: int


# After the definition of `CallRead`, update the forward reference to it:
CustomerReadWithCalls.update_forward_refs()


class CallReadWithCustomer(CallRead):
    customer: CustomerRead | None

routes.py

from fastapi import FastAPI, HTTPException, Depends
from sqlmodel import Session, SQLModel, create_engine

from .models import CustomerReadWithCalls, Customer, Call


api = FastAPI()

sqlite_file_name = 'database.db'
sqlite_url = f'sqlite:///{sqlite_file_name}'
engine = create_engine(sqlite_url, echo=True)


@api.on_event('startup')
def initialize_db():
    SQLModel.metadata.drop_all(engine)
    SQLModel.metadata.create_all(engine)

    # For testing:
    with Session(engine) as session:
        customer = Customer(first_name="Foo", last_name="Bar", email="foo@bar.com")
        call1 = Call(duration=123)
        call2 = Call(duration=456)
        customer.calls.extend([call1, call2])
        session.add(customer)
        session.commit()


def get_session() -> Session:
    session = Session(engine)
    try:
        yield session
    finally:
        session.close()


def get_customer(db: Session, id: int):
    return db.get(Customer, id)


@api.get("/customers/{customer_id}", status_code=200, response_model=CustomerReadWithCalls)
def read_customer(customer_id: int, db: Session = Depends(get_session)):
    customer = get_customer(db, customer_id)
    if customer is None:
        raise HTTPException(status_code=404, detail=f"Customer not found for {customer_id=}")
    return customer

Starting the API server and sending GET to http://127.0.0.1:8000/customers/1 gives me

{
  "first_name": "Foo",
  "last_name": "Bar",
  "email": "foo@bar.com",
  "active": true,
  "id": 1,
  "calls": [
    {
      "duration": 123,
      "cost_per_minute": null,
      "customer_id": 1,
      "created": "2022-08-16T00:00:00",
      "id": 1
    },
    {
      "duration": 456,
      "cost_per_minute": null,
      "customer_id": 1,
      "created": "2022-08-16T00:00:00",
      "id": 2
    }
  ]
}

Hope this helps.