在 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"])
但是,我无法访问验证器中的 request
或 app
对象,因此我无法正确解析 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
假设我在数据库中有一个简单的文件存储。 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"])
但是,我无法访问验证器中的 request
或 app
对象,因此我无法正确解析 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