FastAPI 上传到 S3答案

【问题标题】：FastAPI Upload to S3FastAPI 上传到 S3
【发布时间】：2021-03-09 07:09:02
【问题描述】：

这几天我一直在摸不着头脑。我仍然无法解决问题。基本上，我只是想将 CSV 文件放入 LocalStack S3 中，但无法正常工作。

这是我的代码的 sn-p：

api.py

from files import s3, AWS_S3_BUCKET_NAME, upload_file_to_bucket
    
@router.post('/api/customer/generate/upload',
                 name='Upload CSV to AWS S3 Bucket',
                 status_code=201)
    async def post_upload_user_csv(file_obj: UploadFile = File(...)):
        upload_obj = upload_file_to_bucket(s3_client=s3(),
                                           file_obj=file_obj.file,
                                           bucket=AWS_S3_BUCKET_NAME,
                                           folder='CSV',  # To Be updated
                                           object_name=file_obj.filename)
        if upload_obj:
            return JSONResponse(content="Object has been uploaded to bucket successfully",
                                status_code=status.HTTP_201_CREATED)
        else:
            raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                                detail="File could not be uploaded")

files.py

import os
import boto3
import logging

from botocore.client import BaseClient
from botocore.exceptions import ClientError

AWS_ACCESS_KEY_ID = os.getenv('POSTGRES_HOST')
AWS_SECRET_KEY = os.getenv('AWS_SECRET_KEY')
AWS_S3_BUCKET_NAME = os.getenv('AWS_S3_BUCKET_NAME')


def s3() -> BaseClient:
    client = boto3.client(service_name='s3',
                          aws_access_key_id=AWS_ACCESS_KEY_ID,
                          aws_secret_access_key=AWS_SECRET_KEY,
                          endpoint_url='http://localhost:4566/')  # Use LocalStack Endpoint

    return client


def upload_file_to_bucket(s3_client, file_obj, bucket, folder, object_name=None):
    """Upload a file to an S3 bucket

    :param s3_client: S3 Client
    :param file_obj: File to upload
    :param bucket: Bucket to upload to
    :param folder: Folder to upload to
    :param object_name: S3 object name. If not specified then file_name is used
    :return: True if file was uploaded, else False
    """
    # If S3 object_name was not specified, use file_name
    if object_name is None:
        object_name = file_obj

    # Upload the file
    try:
        # with open("files", "rb") as f:
        s3_client.upload_fileobj(file_obj, bucket, f"{folder}/{object_name}")
    except ClientError as e:
        logging.error(e)
        return False
    return True

问题是 s3_client 需要先以二进制模式打开文件，然后才能将其上传到 s3 存储桶。但是，这不能直接完成，文件需要临时保存在 FastAPI 服务器上，但出于明显的原因，我真的不想这样做。

任何帮助将不胜感激。提前谢谢！

【问题讨论】：

可以先将文件写入io.BytesIO吗？
@fishstix44 很抱歉造成混乱。请纠正我，我的逻辑有问题。当然，最好将文件写入变量而不是将其保存在磁盘上。据我了解，UploadFile 是一个类似文件的对象，这意味着它临时存储在内存上，而不是直接存储在磁盘上。我正在尝试做的是逐块或异步将提供的文件直接上传到 S3，只要可能。

标签： python amazon-s3 fastapi

【解决方案1】：

你能先显示错误吗？基本上。我以前做过。直接从前端上传一个文件到 aws-s3。您可以尝试将ContentType 添加到upload_fileobj。这是我的代码

 content_type = mimetypes.guess_type(fpath)[0]
 s3.Bucket(bucket_name).upload_fileobj(Fileobj=file, Key=file_path,
                                          ExtraArgs={"ACL": "public-read",
                                                     "ContentType": content_type})

另一种方式。您应该尝试将文件转换为 io.BytesIO

 def s3_upload(self, file, file_path, bucket_name, width=None, height=None, make_thumb=False, make_cover=False):
    s3 = boto3.resource(service_name='s3')
    obj = BytesIO(self.image_optimize_from_buffer(file, width, height, make_thumb, make_cover))
    s3.Bucket(bucket_name).upload_fileobj(Fileobj=obj, Key=file_path,
                                          ExtraArgs={"ACL": "public-read", "ContentType": file.content_type})
    return f'https://{bucket_name}.s3.amazonaws.com/{file_path}'

【讨论】：