Patching uploaded files for usage in FastAPI background tasks
Sebastian Thomas (PhD)
Often files uploaded to a server need to be processed in the background such that the client can get an immediate response. However, if the server is set up using FastAPI, this pattern leads to problems in recent versions.
Issue
Since FastAPI v0.106.0, uploaded files cannot be used in a background task since the file will be closed by FastAPI (or the underlying Starlette) before the background task is executed.
async def process_file(
background_tasks: BackgroundTasks,
file: Annotated[UploadFile, File()]
):
background_tasks.add_task(task, file) # Fails due to closed file.
A workaround would be to consume the file and process the raw bytes or any file wrapper such as io.BytesIO in task.
async def process_file(
background_tasks: BackgroundTasks,
file: Annotated[UploadFile, File()]
):
data = await file.read()
in_memory_file = BytesIO(data)
background_tasks.add_task(
task, # Needs to be adapted.
in_memory_file
)
Depending on your setup, this solution might be ok for you.
However, this workaround moves even large files completely into memory. If this doesn't fit your requirements, you can use a more sophisticated patch.
Wrapping UploadFile
We can overwrite the (asynchronous) close method of the uploaded file, so that nothing happens when FastAPI (or the underlying Starlette) invokes it. To retain the complete behavior of fastapi.UploadFile, we can declare a wrapper:
class UploadFilePatch(UploadFile):
"""A patch for fastapi.UploadFile.
Replaces the close method of the patched instance such that it's a
no-op when called by FastAPI/Starlette.
"""
def __init__(self, upload_file: UploadFile) -> None:
self._upload_file = upload_file
close_awaitable = upload_file.close
setattr(upload_file, "close", self._do_not_close)
setattr(self, "close", close_awaitable)
@staticmethod
async def _do_not_close() -> None:
pass
@property
def file(self) -> BinaryIO:
return self._upload_file.file
@file.setter
def file(self, value: BinaryIO) -> None:
self._upload_file.file = value
@property
def filename(self) -> str | None:
return self._upload_file.filename
@filename.setter
def filename(self, value: str | None) -> None:
self._upload_file.filename = value
@property
def size(self) -> int | None:
return self._upload_file.size
@size.setter
def size(self, value: int | None) -> None:
self._upload_file.size = value
@property
def headers(self) -> Headers:
return self._upload_file.headers
@headers.setter
def headers(self, value: Headers) -> None:
self._upload_file.headers = value
async def write(self, data: bytes) -> None:
await self._upload_file.write(data)
async def read(self, size: int = -1) -> bytes:
return await self._upload_file.read(size)
async def seek(self, offset: int) -> None:
await self._upload_file.seek(offset)
async def process_file(
background_tasks: BackgroundTasks,
file: Annotated[UploadFile, File()]
):
file = UploadFilePatch(file)
background_tasks.add_task(
task, # Can still accept an UploadFile.
file
)
Downside
With this solution, FastAPI does no longer take care that the uploaded file is closed, so you have to close the file manually in the background task.
Conclusion
By leveraging Python's dynamic features, we can patch uploaded files to ensure they are properly processed within FastAPI's background tasks.
