Supongamos que tengo un almacenamiento de archivos simple en la base de datos. El modelo SQLAlchemy se ve así:
class Blob(Base): id = Column(Integer, primary_key=True) blob = deferred(Column(LargeBinary().with_variant(LONGBLOB, "mysql")))El modelo Pydantic respectivo se ve así:
class BlobBase(sqlalchemy_to_pydantic(Blob, exclude=["blob"])): class Config: orm_mode = TrueTambién hay una ruta FastAPI destinada a obtener archivos individuales por ID:
@router.get("/blob/{blob_id}") async def get_blob(blob_id: str): [...]Y una ruta para obtener una lista de todos los archivos:
@router.get("/blobs", response_model=List[BlobBase]) async def get_blobs(): return [BlobBase.from_orm(x) for x in db.session.query(Blob).all()] Ahora, me gustaría incluir la URL get_blob resuelta en cada entrada en get_blobs . Ingenuamente, supongo que esto debería verse así:
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"]) Sin embargo, no tengo acceso a una request o un objeto de app en el validador, por lo que no puedo resolver correctamente la URL. NB: tengo acceso al objeto del router , que es un APIRouter para la sub-URL actual, pero get_blob en la aplicación real está en un APIRouter diferente, por lo que no puedo usarlo sin bloquear todo en un solo archivo.
¿Cuál es la forma correcta de resolver esto, es decir, incluir una URL resuelta en la salida del modelo?
Dado que ha definido from_orm en la configuración de su modelo, no debería tener que hacer la transformación manual usando from_orm(x) en su vista get_blobs ; devolver el resultado de la consulta por sí solo debería ser suficiente.
@router.get("/blobs", response_model=List[BlobBase]) async def get_blobs(): return db.session.query(Blob).all() También se sugiere usar una dependencia para resolver la base de datos para cada punto final asíncrono (hay un ejemplo en los documentos de db ) en lugar de tener una entrada de base de db global.
Dado que la URL inversa no es realmente una propiedad del esquema en sí, creo que agregaría un esquema compuesto y luego lo completaría en su vista:
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() ] Otra opción es usar la estrategia llamando manualmente a from_orm y luego tener un esquema que amplíe 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