Amazon Simple Storage Service (Amazon S3) 是一种面向 Internet 的存储服务。您可以通过 Amazon S3 随时在 Web 上的任何位置存储和检索的任意大小的数据
官方文档
运维把服务配好以后，给了一个账号，可以登录查看S3的存储效果
作为一个职业前端选手，看到官方提供了Node的开发文档，果断使用Node进行效果测试
在配置文件配置 aws_key 和 aws_secret 以及 region_name (注意：国内北京区域是 cn-north-1)
配置好以后发现通过官方提供的 Demo 连接不上，总是提示 Key 错误，找了好多方案都没解决，当时就无语了，后来继续看官方文档，后来想到 python 也有实现，要不换 python 试试？配置之类的都没变，python 代码拿来就能跑，既然这么顺，那就用 python 做开发测试吧
获取存储桶名称
import boto3 s3 = boto3.resource('s3') # 获取存储桶名称 for bucket in s3.buckets.all(): print(bucket.name)
上传
import boto3 s3 = boto3.client('s3') # 上传本地图片 data = open('test.jpg', 'rb') s3.Bucket('skin-test').put_object(Key='test.jpg', Body=data) # 通过名称上传文件 s3.upload_file('test.jpg', 'skin-test', 'test-file2.jpg') # 上传时 ExtraArgs 用于指定附加参数，例如用户提供元数据 s3.upload_file( "test.jpg", "skin-test", "test-file3.jpg", ExtraArgs={"Metadata": {"mykey": "myvalue"}} ) # 上传一个可读的类文件对象 with open('package-lock.json', 'rb') as f: s3.upload_fileobj(f, 'skin-test', 'package-lock.json')
获取上传进度
# 为了跟踪传输的进度，可以提供进度回调，以便每次在传输进度时调用回调 import os import sys import threading import boto3 class ProgressPercentage(object): def __init__(self, filename): self._filename = filename self._size = float(os.path.getsize(filename)) self._seen_so_far = 0 self._lock = threading.Lock() def __call__(self, bytes_amount): with self._lock: self._seen_so_far += bytes_amount percentage = (self._seen_so_far / self._size) * 100 sys.stdout.write( "\r%s %s / %s (%.2f%%)" % ( self._filename, self._seen_so_far, self._size, percentage)) sys.stdout.flush() s3 = boto3.client('s3') s3.upload_file( "001.tif", "skin-test", "001.tif", Callback=ProgressPercentage("001.tif"))
下载
# 根据名称下载文件 import boto3 s3 = boto3.client('s3') s3.download_file("skin-test", "001.tif", "001.tif")
显示下载进度
import sys import threading import boto3 class ProgressPercentage(object): def __init__(self, filename): self._filename = filename self._seen_so_far = 0 self._lock = threading.Lock() def __call__(self, bytes_amount): with self._lock: self._seen_so_far += bytes_amount sys.stdout.write( "\r%s --> %s bytes transferred" % ( slf._filename, self._seen_so_far)) sys.stdout.flush() s3 = boto3.client('s3') s3.download_file( "skin-test", "001.tif", "download_001.tif", Callback=ProgressPercentage("download_001.tif"))
生成预签名url链接
import boto3 import requests s3 = boto3.client('s3') url = s3.generate_presigned_url( ClientMethod='get_object', Params={ 'Bucket': 'skin', 'Key': 'server2.js' } ) print(url)
注意：这里需要进行ICP备案以后url才能正常运行，否则会提示无权操作
You are not authorized to perform this operation
![无缝融入，即刻智能[二]：Dify-LLM平台(聊天智能助手、AI工作流)快速使用指南，42K+星标见证专属智能方案](https://ucc.alicdn.com/fnj5anauszhew_20240819_1d1a574e34e6463ab6585a9d8f594cd6.png?x-oss-process=image/format,webp/resize,h_160,m_lfit)