Use FastAPI dependencies when rendering response models

[olivier-thatch]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" 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 FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from fastapi import FastAPI
from pydantic import BaseModel

from .database import Base
from .deps import get_db

# DB models

class ItemDB(Base):
    __tablename__ = "items"

    id: int = Column(Integer, primary_key=True, index=True)
    title: str = Column(String, nullable=False)
    
    def times_ordered(self, db: Session) -> int:
        return db.query(func.count(OrderDB.id)).filter(Order.item_id == self.id).scalar()


class OrderDB(Base):
    __tablename__ = "orders"

    id: int = Column(Integer, primary_key=True, index=True)
    item_id: int = Column(Integer, ForeignKey("items.id"))


# Pydantic model

class Item(BaseModel):
    title: str
    times_ordered: int
 

app = FastAPI()

@app.get("/items/{item_id}", response_model=Item)
async def get_item(db: Session = Depends(get_db), item_id: int) -> Item:
    item_db = db.query(ItemDB).filter(ItemDB.id == item_id).first()
    # assume item_db exists
    item = Item(
        title=item_db.title,
        times_ordered=item_db.times_ordered(db),
    )
    return item

Description

(Assuming the DB is already populated with 1 item with id 1, and 3 orders pointing to the item.)

• Send a request to /items/1
• The API replies with a payload similar to {"title": "Item title", "times_ordered": 3}

Because the ItemDB.times_ordered() instance method needs the db dependency, I was not able to simply return the ItemDB model and let FastAPI map the DB model to the Pydantic model. Instead, I had to manually instantiate the Item Pydantic schema in the route handler.

Wanted Solution

It would be great if there was some way I could tell FastAPI how to instantiate the Pydantic model from the DB model while using FastAPI's dependency injection.

Below is an example of what it could look like. I don't know how FastAPI's DI actually works so this example may not necessary be realistic -- maybe FastAPI would need more information in the route decorator or something. But the gist of it is that the route handler can still return the DB model and not have to worry about mapping the DB model to the Pydantic model.

Wanted Code

from fastapi import FastAPI
from pydantic import BaseModel

from .database import Base
from .deps import get_db

# DB models

class ItemDB(Base):
    __tablename__ = "items"

    id: int = Column(Integer, primary_key=True, index=True)
    title: str = Column(String, nullable=False)
    
    def times_ordered(self, db: Session = Depends(get_db)) -> int:
        return db.query(func.count(OrderDB.id)).filter(Order.item_id == self.id).scalar()


class OrderDB(Base):
    __tablename__ = "orders"

    id: int = Column(Integer, primary_key=True, index=True)
    item_id: int = Column(Integer, ForeignKey("items.id"))


# Pydantic model

class Item(BaseModel):
    title: str
    times_ordered: int
 

app = FastAPI()

@app.get("/items/{item_id}", response_model=Item)
async def get_item(db: Session = Depends(get_db), item_id: int) -> ItemDB:
    item_db = db.query(ItemDB).filter(ItemDB.id == item_id).first()
    # assume item_db exists
    return item_db

Alternatives

The alternative would be to simply do what's in the first example code. For clarity, the mapping of DB model to Pydantic model could be moved into a helper method on the Pydantic model class, but the route handler still has to call the helper method and have its return type set to the Pydantic model rather than the DB model.

Operating System

Linux

Operating System Details

No response

FastAPI Version

0.78.0

Python Version

3.10.4

Additional Context

This example is a bit contrived -- if you had this setup in practice you would also define an orders relationship on the ItemDB DB model class, then you could change times_ordered to be a property that returns len(orders) and Pydantic would be able to fill the times_ordered field with the property's return value. However this would be suboptimal, because it would load all orders from the DB rather than using COUNT(order.id).

More importantly, this would only work for this specific use case. In practice, you may have more complex SQL queries that would not map so cleanly to SQLAlchemy's relationships, or you may need to use other dependencies entirely. For instance, I have a model that has an s3_key column, and I need to inject an S3 client to compute a presigned URL and return the URL in the Pydantic model.

[mcleantom]

I cant test it as I dont have your code but FastAPI allows you to have sub-dependencies, so you should be able to do something like:

class Item(BaseModel):
    title: str
    times_ordered: int

    @classmethod
    def get_from_db(cls, item_id: str, db = Depends(get_db)) -> Item:
        item_db = db.query(ItemDB).filter(ItemDB.id == item_id).first()
        return cls(title=item_db.title, times_ordered=item_db.times_ordered)

@app.get("/items/{item_id}", response_model=Item)
async def get_item(item = Depends(Item.get_from_db)) -> Item:
       return item

If you want automatic conversion from a database to a pydantic model, you could write a function like:

class DBToPydantic:
    def __init__(self, basemodel: BaseModel, database_model: Base):
        self.basemodel = basemodel
        self.database_model = database_model

    def __call__(self, item_id: int, db = Depends(get_db)):
        db_model = db.query(self.database_model).filter(self.database_model.id == item_id).first()
        return self.basemodel(**db_model)

@app.get("/items/{item_id}", response_model=Item)
async def get_item(item = Depends(DBToPydantic(Item, ItemDB)) -> Item:
    return item

[olivier-thatch]

Thanks! That's very helpful, and I think partially solves my ask. There are more complex use cases which I think would be harder to solve this way though. I'll experiment with your approach and report back.

[mcleantom]

@olivier-thatch No worries.

I had a go in my own code retrieving an item from a database, returning it directly and then using a response model and it worked fine for me. I am not sure why it didnt work for your code, were there any error messages?

[olivier-thatch]

No, your code works great! However, it changes the input to the route handler from an ItemDB instance to an Item instance. This works fine for a retrieve endpoint, but for e.g. an update endpoint it's a bit more problematic.

class ItemUpdate(BaseModel):
    title: str

@app.put("/items/{item_id}", response_model=Item)
async def update_item(db: Session = Depends(get_db), item_id: int, params: ItemUpdate) -> ItemDB:
    item_db = db.query(ItemDB).filter(ItemDB.id == item_id).first()
    # assume item_db exists
    item_db.title = params.title
    db.add(item_db)
    db.commit()
    db.refresh(item_db)
    return item_db

I can change item_id: str to item: Item as in your example, but I need the ItemDB instance to actually perform the DB update.

I guess I could store a reference to the ItemDB instance as a private proprety in the Item Pydantic model? Something like this:

class Item(BaseModel):
    _db_model: ItemDB = PrivateAttr()
    title: str
    times_ordered: int

    @classmethod
    def get_from_db(cls, item_id: str, db = Depends(get_db)) -> Item:
        item_db = db.query(ItemDB).filter(ItemDB.id == item_id).first()
        result = cls(title=item_db.title, times_ordered=item_db.times_ordered)
        result._db_model = item_db
        return result

@app.put("/items/{item_id}", response_model=Item)
async def update_item(db: Session = Depends(get_db), item: Item = Depends(Item.get_from_db), params: ItemUpdate) -> Item:
    item_db = item._db_model
    item_db.title = params.title
    db.add(item_db)
    db.commit()
    db.refresh(item_db)
    return item

Seems a bit hacky, but maybe it's not so bad...

EDIT: I tested it, the above doesn't actually work because the Item instance is not updated with the new values. I'll sleep on it and give it another go tomorrow, but open to suggestions :)

[mcleantom]

@olivier-thatch Can you use a validator in the pydantic model to retrieve the times_ordered item, like this:

from pydantic import validator
from inspect import ismethod

class Item(BaseModel):
    title: str
    times_ordered: int
    
    @validator('times_ordered')
    def validate_times_ordered(cls, v):
        if ismethod(v):
            if v.__func__ is ItemDB.times_ordered:
              db = get_db()
              return v(db)
        return v
 
app = FastAPI()

@app.get("/items/{item_id}", response_model=Item)
async def get_item(db: Session = Depends(get_db), item_id: int) -> Item:
    item_db = db.query(ItemDB).filter(ItemDB.id == item_id).first()
    return item_db