ASJServer/Ansjer/test/osssigput.py

#!/usr/bin/env python3  
# -*- coding: utf-8 -*-  
"""
@Copyright (C) ansjer cop Video Technology Co.,Ltd.All rights reserved.
@AUTHOR: ASJRD018
@NAME: AnsjerFormal
@software: PyCharm
@DATE: 2018/11/8 9:23
@Version: python3.6
@MODIFY DECORD:ansjer dev
@file: oss_sts.py
@Contact: chanjunkai@163.com
"""
# -*- coding: utf-8 -*-
'''
生成sts上传授权
'''
from aliyunsdkcore import client
from aliyunsdksts.request.v20150401 import AssumeRoleRequest
import json
import oss2

endpoint = 'oss-cn-shenzhen.aliyuncs.com'

bucket_name = 'testrtmp'

tokenss={"AccessKeyId": "STS.NJRt22iBsPQy4AhgJi8NHPLBG", "AccessKeySecret": "5e3LjYrAsQcoNM1xfAmXYVfHiUst1i2PHTkKVKGFcGte", "SecurityToken": "CAIS1AJ1q6Ft5B2yfSjIr4nnP4iGhJ1S55OSNmfZg0o8NMFkv4nppTz2IHxLf3RuAe8dvvw+nGBV7vsdlqN4S5ZDR1HCbsJxtnbdcbo4J9ivgde8yJBZolnMewHKedCSvqL7Z+H+U6mMGJOEYEzFkSle2KbzcS7YMXWuLZyOj+wuDLEQRRLqVSdaI91UKwB+yrcdLmCDGfupPwLnpWDSAUF0wFYe71ly8qOi2MaRxwPDhVnhsI8vqp/2P4KvYrscXul2WMzn2/dtJOiTknxM5gNW7rtzlOsGg3Ol3/CFKlVd+RidKJCw+9Z0fjR+e7IxFo9c3oGe8qQi4qqJzNuxkksSe7gFDwanHdj6nJGfR7jzbY1iLe2gYk6giI7fasPH1Co/em8eOQ9wfN4sF2R9ExRESEuBdvD8pQ6SPVzzE/ndgPxujMVvr1Dh7dOGIlmJU0Y3c+Dtox9VGoABZmwK4f4Cc5EwoOItvda0tGfrLPUyjZLzzGCdvj8kgNn7YRIN66Rc8foeIIy54fiavbl2UTZyRG7Lk6DD5Ba0aQx145z1VX71OV0qwfr6CtzacfW+wdCW9/f5mv2fDv1/B6V6VobI6n65qQRtgEnnvi2BkNlS7DGO1+4hKtKnvIU=", "expire": 3600, "endpoint": "oss-cn-shenzhen.aliyuncs.com", "bucket_name": "testrtmp"}
# 使用RAM账号的AccessKeyId和AccessKeySecret向STS申请临时token。

# 使用临时token中的认证信息初始化StsAuth实例。
auth = oss2.StsAuth(tokenss['AccessKeyId'],
                    tokenss['AccessKeySecret'],
                    tokenss['SecurityToken'])
print(auth)
# exit()
# 使用StsAuth实例初始化存储空间。
bucket = oss2.Bucket(auth, endpoint, bucket_name)

# 上传一个字符串。
bucket.delete_object('a-name.txt')
exit()

bucket.put_object('a-name.txt', b'hello world')
res =bucket.get_object('object-name.txt')
print(res.read())