代码之家 › 专栏 › 技术社区 › Nikolai Prokoschenko

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

starlette pydantic fastapi

0

Nikolai Prokoschenko · 技术社区 · 4 年前

假设数据库中有一个简单的文件存储。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

还有一个FastAPI路由,用于按ID获取单个文件:

@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_blob 中每个条目的URL get_blobs .天真的,我想应该是这样的:

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

但是,我没有访问 request 或者 app 对象,因此我无法正确解析URL。注:我确实有权访问 router 对象,这是一个 APIRouter 对于当前子URL,但是 弄脏 在实际应用中是以不同的方式进行的 APIRouter ,所以我不能使用它而不将所有内容塞进一个文件中。

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

0 回复 | 直到 4 年前

1

0

MatsLindh 4 年前

既然你已经定义了 from_orm 在模型的配置中,不必使用 from_orm(x) 在你的 get_blobs 视图-仅返回查询结果就足够了。

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

还建议使用依赖项来解决问题 db 对于每个异步端点(FastAPI文档中有一个示例),而不是全局 分贝 进入

由于反向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()
    ]

另一个选择是使用手动调用的策略 来自 ,然后有一个扩展的模式 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