基本上,我正在尝试创建一个端点以将文件上传到 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 吗?
UploadFile
使用Python的SpooledTemporaryFile
,它是“存储在内存中的文件”,“一关闭就销毁”。您可以读取文件内容(即使用 contents = file.file.read()
或 async
读/写查看 此答案),然后将这些字节上传到您的服务器(如果允许),或复制内容将上传的文件放入 NamedTemporaryFile
,如此处所述。与 SpooledTemporaryFile
不同,NamedTemporaryFile
“保证在文件系统中具有可见名称”,“可用于打开文件”。该名称可以从 name
属性(即 temp.name
)中检索。示例:
from fastapi import HTTPException
@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:
raise HTTPException(status_code=500, detail='Error on uploading the file')
finally:
file.file.close()
# Here, upload the file to your S3 service using `temp.name`
s3_client.upload_file(temp.name, 'local', 'myfile.txt')
except Exception:
raise HTTPException(status_code=500, detail='Something went wrong')
finally:
#temp.close() # the `with` statement above takes care of closing the file
os.remove(temp.name) # Delete temp file
此外,还可以使用
.file
属性访问实际的 Python 文件。根据文档:
:一个file
(一个类文件对象)。这是实际情况 可以直接传递给其他函数或库的 Python 文件 期望一个“类似文件”的对象。SpooledTemporaryFile
因此,您也可以尝试使用
upload_fileobj
函数并传递 upload_file.file
:
response = s3_client.upload_fileobj(upload_file.file, bucket_name, os.path.join(dest_path, upload_file.filename))
或者,使用
._file
的
SpooledTemporaryFile
属性传递类似文件的对象,该属性返回 io.BytesIO
或 io.TextIOWrapper
对象(取决于指定的是二进制还是文本模式) .
response = s3_client.upload_fileobj(upload_file.file._file, bucket_name, os.path.join(dest_path, upload_file.filename))
BytesIO
),使用它将内容上传到 S3 存储桶,最后关闭它(“当 close()
时,缓冲区将被丢弃”方法被称为。”)。写入 seek(0)
流后,请记得调用 BytesIO
方法将光标重置回文件开头。
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()