这次的案例是利用阿里云oss2进行简单的bucket查询以及上传下载等基础操作
- 对象存储
对象存储的概念我就不做太多概述了，详情可以参考阿里云 - sdk
python的sdk不仅提供对文件的处理，包括其它内容比如图片也有很好的支持，可以实现对图片在线缩放、水印等一系列操作，详情可参考官方图片处理文档 - 文件处理代码
# -*- coding: utf-8 -*
import oss2
import os
access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', 'your key')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', 'your secret')
endpoint = os.getenv('OSS_TEST_ENDPOINT', 'http://oss-cn-beijing.aliyuncs.com')
path = os.path.abspath('.')
dir_path = os.path.abspath('dir')
dirs = os.listdir(path)
def showBucket():
print("********** 获取bucket信息 *******")
service = oss2.Service(oss2.Auth(access_key_id, access_key_secret), endpoint)
print("*****************************")
print(" 现有bucket有： ")
print('\n'.join(info.name for info in oss2.BucketIterator(service)))
print("*****************************")
def createBucket():
print("********** 创建 *******")
bucket_input = input("请输入想创建的bucket名： ")
# 创建Bucket对象，所有Object相关的接口都可以通过Bucket对象来进行
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name=bucket_input)
# 带权限与存储类型创建bucket
bucket.create_bucket(permission=oss2.BUCKET_ACL_PRIVATE,
input=oss2.models.BucketCreateConfig(oss2.BUCKET_STORAGE_CLASS_STANDARD))
if oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name=bucket_input):
print(" 成功创建%s" %bucket_input)
showBucket()
print("***************************")
def bucketInfo():
print("********** 获取bucket_info *******")
bucket_input = input('请输入bucket名： ')
# 获取bucket相关信息
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name=bucket_input)
bucket_info = bucket.get_bucket_info()
print(" bucket_info:")
print(' name: ' + bucket_info.name)
print(' storage class: ' + bucket_info.storage_class)
print(' creation date: ' + bucket_info.creation_date)
print("*******************************")
print("*******************************")
print(" bucket_stat:")
bucket_stat = bucket.get_bucket_stat()
print(' storage: ' + str(bucket_stat.storage_size_in_bytes))
print(' object count: ' + str(bucket_stat.object_count))
print(' multi part upload count: ' + str(bucket_stat.multi_part_upload_count))
print("********************************")
def upload():
print("********** 上传 *******")
bucket_input = input('请输入要传入的bucket名： ')
print("**************************")
print(" 当前目录下所有文件：")
for file in dirs:
print(file)
print("***************************")
filename = input('请输入要上传的文件名： ')
cloud_name = input('请输入云端文件名： ')
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name=bucket_input)
with open(oss2.to_unicode(filename), 'rb') as f:
bucket.put_object(cloud_name, f)
meta = bucket.get_object_meta(cloud_name)
if meta:
print(" 上传成功")
print(" 云端所有文件：")
for i in oss2.ObjectIterator(bucket):
print(i.key)
else:
print(" 上传失败")
def download():
print("********** 下载 *******")
bucket_input = input('请输入bucket名：')
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name=bucket_input)
print(" %s下有如下文件：" %bucket_input)
for i in oss2.ObjectIterator(bucket):
print(i.key)
print("***************************")
cloud_name = input('请输入要下载的文件名：')
file_name = input('请输入保存至本地文件名：')
bucket.get_object_to_file(cloud_name, file_name)
print(file_name[4:])
if file_name[4:] in os.listdir(dir_path):
print(" 成功下载%s" %cloud_name)
print("**************************")
print(" 当前目录下所有文件：")
for file in os.listdir(dir_path):
print(file)
print("***************************")
def remove():
print("********** 删除 *******")
bucket_input = input('请输入bucket名：' )
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name=bucket_input)
print(" %s下有如下文件(删除前)：" % bucket_input)
for i in oss2.ObjectIterator(bucket):
print(i.key)
print("***************************")
file_name = input('请输入要删除的文件名：' )
# 删除名为motto.txt的Object
bucket.delete_object(file_name)
print(" 成功删除%s" %file_name)
print(" %s下有如下文件(删除后)：" % bucket_input)
for i in oss2.ObjectIterator(bucket):
print(i.key)
def main():
print("************************")
print(" 「指令数字」：")
print("0. 获取bucket信息")
print("1. 查看具体bucket信息")
print("2. 创建bucket")
print("3. 上传文件")
print("4. 下载文件")
print("5. 删除文件")
print("6. 退出")
print("************************")
while 1:
num = int(input('请输入指令数：'))
if num == 0:
showBucket()
elif num == 1:
bucketInfo()
elif num == 2:
createBucket()
elif num == 3:
upload()
elif num == 4:
download()
elif num == 5:
remove()
else:
break
if __name__ == '__main__':
main()
- 这里是小编用python3实现的简单功能，具体细节大家看注释应该都能理解，主要是对上传下载地址路径的板块可以详细注意一下，小编当时也有点混乱，所以专门用了测试文件来调试这版块的内容，大家也可以参考一下，github里面有其他版本的代码，欢迎参考、建议
- 如果内容中有疑问欢迎留言或者参考：
