python - 如何从 FastAPI 中的 UploadFile 获取文件路径？

Question

基本上，我正在尝试创建一个端点以将文件上传到 Amazon S3。

async def upload_files(filepath: str, upload_file_list: List[UploadFile] = File(...)):
    for upload_file in upload_file_list:
        abs_file_path = "/manual/path/works" + upload_file.path
        # Replace above line to get absolute file path from UploadFile
        response = s3_client.upload_file(abs_file_path,bucket_name,
                                                   os.path.join(dest_path, upload_file.filename))

以上是我将多个文件上传到 S3 存储桶的代码。s3_client.upload_file() 接受要上传的文件的绝对文件路径。当我手动输入完整路径时它正在工作。

然而，这没有用:

response = s3_client.upload_file(upload_file.filename, bucket_name,
                                                   os.path.join(dest_path, upload_file.filename))

有没有办法在 FastAPI 中获取这个绝对路径？或者，在不复制或写入文件的情况下使用 temp_path 的任何替代方案？

如果没有，那么使用 boto3 使用 FastAPI 将文件上传到 S3 的替代方法是什么？

Answer 1

UploadFile使用 Python 的 SpooledTemporaryFile ，这是一个“存储在内存中的文件”，“一关闭就销毁”。您可以读取文件内容(即使用 contents = file.file.read() 或对于 async read/write 看看 this answer )，以及然后将这些字节上传到您的服务器(如果允许)，或将上传文件的内容复制到 NamedTemporaryFile ，正如解释的那样here .与 SpooledTemporaryFile 不同，NamedTemporaryFile“保证在文件系统中有一个可见的名称”，“可用于打开文件”。该名称可以从 name 属性(即 temp.name)中检索。示例:

@app.post("/upload")
def upload(file: UploadFile = File(...)):
    temp = NamedTemporaryFile(delete=False)
    try:
        try:
            contents = file.file.read()
            with temp as f:
                f.write(contents);
        except Exception:
            return {"message": "There was an error uploading the file"}
        finally:
            file.file.close()
            
        # Here, upload the file to your S3 service using `temp.name`
        
    except Exception:
        return {"message": "There was an error processing the file"}
    finally:
        #temp.close()  # the `with` statement above takes care of closing the file
        os.remove(temp.name)  # Delete temp file

更新

此外，可以使用 file 属性访问实际的 Python 文件。根据 documentation :

file: A SpooledTemporaryFile (a file-like object). This is the actualPython file that you can pass directly to other functions or librariesthat expect a "file-like" object.

因此，您也可以尝试使用 upload_fileobj 函数并传递 upload_file.file:

 response = s3_client.upload_fileobj(upload_file.file, bucket_name, os.path.join(dest_path, upload_file.filename))

或者，使用 SpooledTemporaryFile 的 _file 属性传递一个类似文件的对象, 它返回 io.BytesIO或 io.TextIOWrapper对象(取决于指定的是二进制模式还是文本模式)。

 response = s3_client.upload_fileobj(upload_file.file._file, bucket_name, os.path.join(dest_path, upload_file.filename))

更新2

您甚至可以将字节保存在内存缓冲区中 BytesIO ，使用它将内容上传到 S3 存储桶，最后关闭它(“调用 close() 方法时缓冲区被丢弃。”)。请记住在完成写入 BytesIO 流后调用 seek(0) 方法将光标重置回文件的开头。

contents = file.file.read()
temp_file = io.BytesIO()
temp_file.write(contents)
temp_file.seek(0)
s3_client.upload_fileobj(temp_file, bucket_name, os.path.join(dest_path, upload_file.filename))
temp_file.close()