【關于boto3】

Boto3是亞馬遜AWS提供的python SDK，最為常用的功能是S3對象存儲的訪問。作為標準的S3 SDK，除了訪問AWS，也可以訪問其他兼容S3 API的云存儲廠商。

Boto3的項目地址為：https://github.com/boto/boto3.git

Boto3的AWS doc地址為：https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#service-resource

?

需要指出的是，AWS的文檔說明較為混亂，現在給大家提供一個簡單的使用說明和范例。

?

【準備】

首先需要安裝python，建議使用Python 3.6及以后的版本。

通過pip安裝boto3的包即可：

pip install boto3

?

在你的python腳本直接引用boto3即可：

import boto3

?

正式使用boto3之前，你還需要具備云存儲廠商的access?key和secret_key。先給出一對土星云的測試key，有需要的可以去土星云存儲官網注冊獲取。

param_endpoint='https://s3.local-north-1.saturncloud.com.cn:6666'

param_access_key='3ABC8E3ABC387442B3936F9426B11C1B'

param_secret_key='0428D37B0FDC14805AF0153969C272F3'

?

【創建client】

先初始化一個client對象，我們可以定義一個方法如下：

def setup_client(endpoint, access_key_id, secret_access_key):

return boto3.client(

's3',

aws_access_key_id=access_key_id,

aws_secret_access_key=secret_access_key,

use_ssl=True,

region_name='cn',

endpoint_url=endpoint,

config=Config(s3={"addressing_style": "virtual", "signature_version": 's3v4'}))

再定義一個判斷http響應狀態的方法：

def is_result_ok(return_obj):

ifcode = return_obj['ResponseMetadata']['HTTPStatusCode']

if ifcode == 200 or ifcode == 204:

return True

else:

return False

?

然后在main函數中調用創建client的方法：

if __name__ == '__main__':

print('Start SCS testing at '+datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))

?

s3=setup_client(param_endpoint, param_access_key, param_secret_key)

?

【創建存儲桶】

以當前時間為名創建一個存儲桶：

bucket_id=datetime.datetime.now().strftime('test-%Y%m%d-%H%M%S')

?

result=cre_bucket(s3, bucket_id)

?

if is_result_ok(result):

print('bucket ['+bucket_id+'] created ok.')

else:

print('bucket ['+bucket_id+'] created fail.')

print(result)

sys.exit(1)

?

【上傳文件】

定義一個上傳文件的方法如下：

def put_obj(s3, bucket_id, filename, local_dir):

return s3.put_object(Bucket=bucket_id, Body=open(local_dir+'/'+filename, 'rb'), Key=filename)

選擇一個本地文件上傳到剛剛創建的存儲桶

param_test_local_dir='/tmp/'

filename='test.txt’

result=put_obj(s3, bucket_id, filename, param_test_local_dir)

if is_result_ok(result):

print('put_obj ['+filename+'] ok.')

else:

print('put_obj ['+filename+'] fail.')

print(result)

return

?

【下載文件】

定義一個下載文件的方法：

def get_obj(s3, bucket_id, filename):

return s3.get_object(Bucket=bucket_id, Key=filename)

我們再把剛才上傳的文件下載下來

result=get_obj(s3, bucket_id, filename)

?

if is_result_ok(result):

print('get_obj ['+filename+'] ok.')

else:

print('get_obj ['+filename+'] fail.')

print(result)

return

?

【刪除文件】

首先定義一個刪除S3存儲桶文件的方法

def del_obj(s3, bucket_id, filename):

return s3.delete_object(Bucket=bucket_id, Key=filename)

然后調用此方法即可刪除剛剛上傳到云端的文件：

result=del_obj(s3, bucket_id, filename)

?

if is_result_ok(result):

print('del_obj ['+filename+'] ok.')

else:

print('del_obj ['+filename+'] fail.')

print(result)

return

?

【結語】

上面已經給出了一個簡單的使用范例，更多的接口和使用方法，可以查看AWS的文檔和云存儲廠商的文檔。

?

【附 完整代碼】

?

############################################################################################################

'''

Random test script for saturn cloud systems.

?

Usage:

1. Change test parameters through modifying param_XXXX in test configurations.

2. Run 'python scs_rand_test.py'

3. For better performance, please set param_test_local_dir to a ramdisk path, such as '/run' in Linux.

?

Requirement:

1. Python 3.6 or higher version.

2. boto3, use 'pip install boto3 --user' to install.

?

by sam 2021-10-06

'''

############################################################################################################

?

?

import time

import datetime

import os

import sys

import subprocess

import random

import numpy as np

import threading

from queue import Queue

?

import boto3

from botocore.config import Config

?

############################################################################################################

'''

Test configurations

'''

############################################################################################################

?

param_endpoint='https://s3.local-north-1.saturncloud.com.cn:6666'

param_access_key='3ABC8E3ABC387442B3936F9426B11C1B'

param_secret_key='0428D37B0FDC14805AF0153969C272F3'

?

param_test_local_dir='/tmp/'

?

param_test_max_blocks=33

param_block_size=256*1024

?

############################################################################################################

'''

Common used API fuctions

'''

############################################################################################################

?

def setup_client(endpoint, access_key_id, secret_access_key):

return boto3.client(

's3',

aws_access_key_id=access_key_id,

aws_secret_access_key=secret_access_key,

use_ssl=True,

region_name='cn',

endpoint_url=endpoint,

config=Config(s3={"addressing_style": "virtual", "signature_version": 's3v4'}))

?

def cre_bucket(s3, bucket_id):

????return s3.create_bucket(Bucket=bucket_id,)

?

def del_bucket(s3, bucket_id):

return s3.delete_bucket(Bucket=bucket_id,)

?

def put_obj(s3, bucket_id, filename, local_dir):

????return s3.put_object(Bucket=bucket_id, Body=open(local_dir+'/'+filename, 'rb'), Key=filename)

?

def get_obj(s3, bucket_id, filename):

return s3.get_object(Bucket=bucket_id, Key=filename)

?

def del_obj(s3, bucket_id, filename):

????return s3.delete_object(Bucket=bucket_id, Key=filename)

?

def is_result_ok(return_obj):

ifcode = return_obj['ResponseMetadata']['HTTPStatusCode']

if ifcode == 200 or ifcode == 204:

return True

else:

return False

?

############################################################################################################

'''

Test applications

'''

############################################################################################################

def create_dummy_file(file_path):

blocks=random.randrange(0, param_test_max_blocks, 1)

last_bytes=random.randrange(1, param_block_size+1, 1)

total_size=blocks*param_block_size+last_bytes

?

file=open(file_path, mode='wb')

file.write(np.random.bytes(total_size))

file.close()

?

def test_task(bucket_id):

s3=setup_client(param_endpoint, param_access_key, param_secret_key)

?

filename='put_obj'

create_dummy_file(param_test_local_dir+'/'+filename)

result=put_obj(s3, bucket_id, filename, param_test_local_dir)

?

if is_result_ok(result):

print('put_obj ['+filename+'] ok.')

else:

print('put_obj ['+filename+'] fail.')

print(result)

return

?

result=get_obj(s3, bucket_id, filename)

?

if is_result_ok(result):

print('get_obj ['+filename+'] ok.')

else:

print('get_obj ['+filename+'] fail.')

print(result)

return

?

filename_new='get_obj'

file=open(param_test_local_dir+'/'+filename_new, mode='wb')

file.write(result['Body'].read())

file.close()

?

cmd_string='cmp '+param_test_local_dir+'/'+filename+' '+param_test_local_dir+'/'+filename_new

cmd_result=subprocess.getoutput(cmd_string)

????

if len(cmd_result)>1:

print('content error at ['+filename+'] and ['+filename_new+']')

return

????

result=del_obj(s3, bucket_id, filename)

?

if is_result_ok(result):

print('del_obj ['+filename+'] ok.')

else:

print('del_obj ['+filename+'] fail.')

print(result)

return

?

cmd_string='rm '+param_test_local_dir+'/'+filename+' '+param_test_local_dir+'/'+filename_new

os.system(cmd_string)

?

?

?

############################################################################################################

'''

Script entrance

'''

############################################################################################################

if __name__ == '__main__':

print('Start SCS random testing at '+datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))

?

s3=setup_client(param_endpoint, param_access_key, param_secret_key)

?

bucket_id=datetime.datetime.now().strftime('test-%Y%m%d-%H%M%S')

?

result=cre_bucket(s3, bucket_id)

?

if is_result_ok(result):

print('bucket ['+bucket_id+'] created ok.')

else:

print('bucket ['+bucket_id+'] created fail.')

print(result)

sys.exit(1)

pass

?

test_task(bucket_id)

?

result=del_bucket(s3, bucket_id)

?

if is_result_ok(result):

print('bucket ['+bucket_id+'] deleted ok.')

else:

print('bucket ['+bucket_id+'] deleted fail.')

print(result)

sys.exit(1)

pass

?

print('End SCS random testing at '+datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))

編輯：fqj