Let's assume I have a simple file storage in the database. The SQLAlchemy model looks like this:
class Blob(Base):
id = Column(Integer, primary_key=True)
blob = deferred(Column(LargeBinary().with_variant(LONGBLOB, "mysql")))
The respective Pydantic model looks like this:
class BlobBase(sqlalchemy_to_pydantic(Blob, exclude=["blob"])):
class Config:
orm_mode = True
There is also a FastAPI route intented for fetching single files by ID:
@router.get("/blob/{blob_id}")
async def get_blob(blob_id: str):
[...]
And a route for fetching a list of all files:
@router.get("/blobs", response_model=List[BlobBase])
async def get_blobs():
return [BlobBase.from_orm(x) for x in db.session.query(Blob).all()]
Now, I would like to include the resolved get_blob URL in each entry in get_blobs. Naively, I suppose this should look like this:
class BlobBase(sqlalchemy_to_pydantic(Blob, exclude=["blob"])):
class Config:
orm_mode = True
url: Optional[str]
@validator("url")
def make_url(cls, v, values):
return request.url_for("get_blob", blob_id=values["id"])
However, I do not have access to a request or an app object in the validator, so that I cannot properly resolve the URL. NB: I do have access to the router object, which is an APIRouter for the current sub-URL, but get_blob in the actual application is in a different APIRouter, so I can't use that without jamming everything in a single file.
What is the correct way to solve this, i.e. include a resolved URL in model's output?
Since you have defined from_orm in the config for your model, you shouldn't have to do the manual transformation using from_orm(x) in your get_blobs view - returning the result from the query by itself should be enough.
@router.get("/blobs", response_model=List[BlobBase])
async def get_blobs():
return db.session.query(Blob).all()
It's also suggested to use a dependency to resolve db for each async endpoint (there's an example in the FastAPI docs) instead of having a global-ish db entry.
Since the reverse URL isn't really a property of the schema itself, I think I'd go with adding a composite schema and then populate it in your view:
class BlobWithUrl(BaseModel):
blob: BaseBlob # Consider using Blob as the name instead - base indicates that it should only be inherited
url: str
@router.get("/blobs", response_model=List[BlobWithUrl])
async def get_blobs(request: Request):
return [
{'url': url_for(...), 'blob': blob}
for blob in db.session.query(Blob).all()
]
Another option is to use the strategy with manually calling from_orm, and then having a schema that extends BlobBase:
class BlobWithUrl(BaseBlob):
url: Optional[str]
@router.get("/blobs", response_model=List[BlobWithUrl])
async def get_blobs(request: Request):
blobs = []
for retrieved_blob in db.session.query(Blob).all():
blob = BlobWithUrl.from_orm(retrieved_blob)
blob.url = url_for(...)
blobs.append(blob)
return blobs