在 FastAPI 中,如何使用基于字段的反向 URL 查找来扩展模型的输出?

In FastAPI, how can I expand a model's output with a reverse URL lookup based on a field?

假设我在数据库中有一个简单的文件存储。 SQLAlchemy 模型如下所示:

class Blob(Base):
    id = Column(Integer, primary_key=True)
    blob = deferred(Column(LargeBinary().with_variant(LONGBLOB, "mysql")))

相应的 Pydantic 模型如下所示:

class BlobBase(sqlalchemy_to_pydantic(Blob, exclude=["blob"])):
    class Config:
        orm_mode = True

还有一个旨在通过 ID 获取单个文件的 FastAPI 路由:

@router.get("/blob/{blob_id}")
async def get_blob(blob_id: str):
   [...]

以及获取所有文件列表的路径:

@router.get("/blobs", response_model=List[BlobBase])
async def get_blobs():
    return [BlobBase.from_orm(x) for x in db.session.query(Blob).all()]

现在,我想在 get_blobs 的每个条目中包含已解决的 get_blob URL。天真地,我想这应该是这样的:

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"])

但是,我无法访问验证器中的 requestapp 对象,因此我无法正确解析 URL。注意:我确实可以访问 router 对象,它是当前子 URL 的 APIRouter,但实际应用程序中的 get_blob 在不同的 APIRouter,所以如果不将所有内容都塞进一个文件中,我就无法使用它。

解决这个问题的正确方法是什么,即在模型的输出中包含已解决的 URL?

由于您已经在模型的配置中定义了 from_orm,因此您不必在 get_blobs 视图中使用 from_orm(x) 进行手动转换 - 从返回结果查询本身就足够了。

@router.get("/blobs", response_model=List[BlobBase])
async def get_blobs():
    return db.session.query(Blob).all()

还建议使用依赖项为每个异步端点解析 db(FastAPI 文档中有一个示例),而不是使用 global-ish db 条目。

由于反向 URL 并不是模式本身的 属性,我想我会添加一个复合模式,然后在您的视图中填充它:

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()
    ]

另一种选择是使用手动调用 from_orm 的策略,然后使用扩展 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