python使用

官方文档

官方模块

BOTO3

# 专门安装 Boto3 1.0 版
pip install boto3==1.0.0

# 确保 Boto3 不低于 1.15.0 版本
pip install boto3>=1.15.0

# 避免使用比 1.15.3 更新的 Boto3 版本
pip install boto3<=1.15.3

Boto3 默认不使用 AWS CRT

AWS CRT 官方的加速模块，建议使用

pip install boto3[crt]

.\python.exe -m pip install pip install boto3[crt]

基本使用

import boto3, os
from os import path
from botocore.exceptions import ClientError


class AwsS3:
    def __init__(self, BUCKET_NAME: str):
        self.s3 = boto3.client('s3')
        self.BUCKET_NAME = BUCKET_NAME

    def upload_file(self, file_name: str, cdn_path: str = "") -> bool:
        if not cdn_path:
            cdn_path = os.path.basename(file_name)

        # Upload the file
        try:
            self.s3.upload_file(file_name, self.BUCKET_NAME, cdn_path)
        except ClientError as e:
            print(f'上传文件：{file_name} 失败', e)
            return False
        return True

    def download_file(self, cdn_path: str, output_name: str = ""):
        if not output_name:
            output_name = path.basename(cdn_path)

        self.s3 = boto3.client('s3')
        with open(output_name, 'wb') as f:
            self.s3.download_fileobj(self.BUCKET_NAME, cdn_path, f)


if __name__ == '__main__':
    target = path.abspath('test.py')

    s3 = AwsS3('test-bucket-psd-tools')

    # s3.upload_file(target, 'test/test.py')
    s3.download_file('psd/1-2.psd/1-2_small.psd', '1-2_small.psd')

下载文件

import boto3
s3 = boto3.client('s3')


s3.download_file('BUCKET_NAME', 'OBJECT_NAME', 'FILE_NAME')
# or
with open('FILE_NAME', 'wb') as f:
    s3.download_fileobj('BUCKET_NAME', 'OBJECT_NAME', f)

上传文件

生成带key的url

bucket_name = ""
bucket_region = ""
upload_path = "path/output.jpg"

s3 = boto3.client("s3", region_name=bucket_region)


url = s3.generate_presigned_url(
                operation, # "get_object"|"put_object"
                Params={"Bucket": bucket_name, "Key": upload_path},
                ExpiresIn=1800,
            )
            
# 通过生成的图片，上传文件
response = None
with open(r"d:/1.jpg", "rb") as object_file:
    object_text = object_file.read()
    
    # 正式上传文件
    response = requests.put(url, data=object_text)
print("response: ", response)