批量异步上传aws图片脚本（python）

背景

工作中需要上传一些测试图片，于是网上找找资料（官方说明），前置步骤如下。

1. python需要3.8以上，安装最新的boto3库：

   pip install boto3

2. 有一个S3权限的aws账户，得到访问密钥ACCESS_KEY与SECRET_KEY，以及上传图片的存储桶位置
3. 安装异步编程asyncio，aiohttp库，方便本地异步上传图片

代码实现

 1 # -*- coding: utf-8 -*-
 2 """
 3 # @Author : qgc
 4 # @Time : 2024/3/13 14:28
 5 # @File : upload_to_aws.py
 6 # Description : 文件说明
 7 """
 8 import os
 9 import boto3
10 import asyncio
11 import aiohttp
12 
13 aws_access_key_id = 'xxxxxx'
14 aws_secret_access_key = 'xxxxxxx'
15 bucket_name = 'deme-normal'   # 存储桶的具体路由，路由不存在会上传失败，但不会报错
16 
17 async def upload_to_aws(local_file, s3_file):
18     s3 = boto3.client('s3', aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key)
19     try:
20         s3.upload_file(local_file, bucket_name, s3_file)
21         print(f"{local_file} upload to s3://{bucket_name}/{s3_file} successed")
22         return True
23     except FileNotFoundError:
24         print('本地文件未找到')
25         return False
26 
27 async def upload_task(url, session, semaphore):
28     try:
29         if url:
30             local_filename = url
31             s3_path = 'qgctest/' + os.path.basename(url)
32             await upload_to_aws(local_filename, s3_path)
33     except Exception as e:
34         print(f'处理{url}:{e}')
35 
36 def get_image_files(directory):
37     image_files = []
38     if directory:
39         for file in os.listdir(directory):
40             if file.endswith('.jpg') or file.endswith('.jpeg') or file.endswith('.png') or file.endswith('.gif'):
41                 image_files.append(os.path.join(directory, file))
42     return image_files
43 
44 async def main(image_files):
45     semaphore = asyncio.Semaphore(10)
46     connector = aiohttp.TCPConnector(limit=10)
47     async with aiohttp.ClientSession(connector=connector) as session:
48         tasks = [upload_task(url, session, semaphore) for url in image_files]
49         await asyncio.gather(*tasks)
50 
51 
52 if __name__ == '__main__':
53     directory = f'E:\图片\images2'  # 本地图片路径
54     image_files = get_image_files(directory)
55     # image_files = ['E:\\图片\\images2\\0.jpg']
56     print(f'共找到{len(image_files)}个图片')
57     asyncio.get_event_loop().run_until_complete(main(image_files))