以下是一个使用Python语言的示例代码,用于跟踪AWS预签名URL的数据上传进度。
import requests
import os
import boto3
import threading
class ProgressPercentage(object):
def __init__(self, filename):
self._filename = filename
self._size = float(os.path.getsize(filename))
self._lock = threading.Lock()
self._uploaded = 0
def __call__(self, bytes_amount):
with self._lock:
self._uploaded += bytes_amount
percentage = (self._uploaded / self._size) * 100
print(f"Uploaded {self._uploaded} bytes out of {self._size} ({percentage:.2f}%)")
# 获取预签名URL
def get_presigned_url(bucket, key):
s3_client = boto3.client('s3')
response = s3_client.generate_presigned_url(
'put_object',
Params={
'Bucket': bucket,
'Key': key,
},
ExpiresIn=3600 # URL有效期为1小时
)
return response
# 使用预签名URL上传文件
def upload_file_with_progress(url, file_path):
with open(file_path, 'rb') as file:
total_size = os.path.getsize(file_path)
headers = {'Content-Length': str(total_size)}
response = requests.put(url, data=file, headers=headers, stream=True)
if response.status_code == 200:
print("File uploaded successfully")
else:
print("File upload failed")
# 示例用法
def main():
bucket = 'your-bucket-name'
key = 'your-file-key'
file_path = 'path/to/your/file'
url = get_presigned_url(bucket, key)
progress = ProgressPercentage(file_path)
upload_thread = threading.Thread(
target=upload_file_with_progress,
args=(url, file_path),
kwargs={'progress_callback': progress}
)
upload_thread.start()
# 等待上传完成
upload_thread.join()
if __name__ == '__main__':
main()
上述代码中,我们定义了一个ProgressPercentage类,用于跟踪上传进度。在每次上传数据时,会调用该类的__call__方法来更新进度信息。
get_presigned_url函数使用AWS SDK的generate_presigned_url方法来获取预签名URL。upload_file_with_progress函数使用requests库发送PUT请求来上传文件,并将进度信息传递给ProgressPercentage类。
在主函数main中,我们可以设置bucket、key和file_path来指定要上传的文件。然后,我们获取预签名URL并创建一个ProgressPercentage对象。接下来,我们创建一个线程来执行文件上传操作,并在最后等待上传完成。
请确保已安装所需的Python库(boto3和requests),并将示例代码中的your-bucket-name、your-file-key和path/to/your/file替换为实际的值。
上一篇:AWS预签名URL查询字符串解释