電子發(fā)燒友App
【關(guān)于boto3】
Boto3是亞馬遜AWS提供的python SDK，最為常用的功能是S3對(duì)象存儲(chǔ)的訪問。作為標(biāo)準(zhǔn)的S3 SDK，除了訪問AWS，也可以訪問其他兼容S3 API的云存儲(chǔ)廠商。
Boto3的項(xiàng)目地址為：https://github.com/boto/boto3.git
Boto3的AWS doc地址為：https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#service-resource
?
需要指出的是，AWS的文檔說明較為混亂，現(xiàn)在給大家提供一個(gè)簡(jiǎn)單的使用說明和范例。
?
【準(zhǔn)備】
首先需要安裝python，建議使用Python 3.6及以后的版本。
通過pip安裝boto3的包即可：
pip install boto3
?
在你的python腳本直接引用boto3即可：
import boto3
?
正式使用boto3之前，你還需要具備云存儲(chǔ)廠商的access?key和secret_key。先給出一對(duì)土星云的測(cè)試key，有需要的可以去土星云存儲(chǔ)官網(wǎng)注冊(cè)獲取。
param_endpoint='https://s3.local-north-1.saturncloud.com.cn:6666'
param_access_key='3ABC8E3ABC387442B3936F9426B11C1B'
param_secret_key='0428D37B0FDC14805AF0153969C272F3'
?
【創(chuàng)建client】
先初始化一個(gè)client對(duì)象，我們可以定義一個(gè)方法如下：
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'}))
再定義一個(gè)判斷http響應(yīng)狀態(tài)的方法：
def is_result_ok(return_obj):
ifcode = return_obj['ResponseMetadata']['HTTPStatusCode']
if ifcode == 200 or ifcode == 204:
return True
else:
return False
?
然后在main函數(shù)中調(diào)用創(chuàng)建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)
?
【創(chuàng)建存儲(chǔ)桶】
以當(dāng)前時(shí)間為名創(chuàng)建一個(gè)存儲(chǔ)桶：
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)
?
【上傳文件】
定義一個(gè)上傳文件的方法如下：
def put_obj(s3, bucket_id, filename, local_dir):
return s3.put_object(Bucket=bucket_id, Body=open(local_dir+'/'+filename, 'rb'), Key=filename)
選擇一個(gè)本地文件上傳到剛剛創(chuàng)建的存儲(chǔ)桶
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
?
【下載文件】
定義一個(gè)下載文件的方法：
def get_obj(s3, bucket_id, filename):
return s3.get_object(Bucket=bucket_id, Key=filename)
我們?cè)侔褎偛派蟼鞯奈募螺d下來
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
?
【刪除文件】
首先定義一個(gè)刪除S3存儲(chǔ)桶文件的方法
def del_obj(s3, bucket_id, filename):
return s3.delete_object(Bucket=bucket_id, Key=filename)
然后調(diào)用此方法即可刪除剛剛上傳到云端的文件：
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
?
【結(jié)語(yǔ)】
上面已經(jīng)給出了一個(gè)簡(jiǎn)單的使用范例，更多的接口和使用方法，可以查看AWS的文檔和云存儲(chǔ)廠商的文檔。
?
【附 完整代碼】
?
############################################################################################################
'''
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
工商網(wǎng)監(jiān)
評(píng)論