oss2/api.py (1,587 lines of code) (raw):
# -*- coding: utf-8 -*-
"""
文件上传方法中的data参数
------------------------
诸如 :func:`put_object <Bucket.put_object>` 这样的上传接口都会有 `data` 参数用于接收用户数据。`data` 可以是下述类型
- unicode类型(对于Python3则是str类型):内部会自动转换为UTF-8的bytes
- bytes类型:不做任何转换
- file-like object:对于可以seek和tell的file object,从当前位置读取直到结束。其他类型,请确保当前位置是文件开始。
- 可迭代类型:对于无法探知长度的数据,要求一定是可迭代的。此时会通过Chunked Encoding传输。
Bucket配置修改方法中的input参数
-----------------------------
诸如 :func:`put_bucket_cors <Bucket.put_bucket_cors>` 这样的Bucket配置修改接口都会有 `input` 参数接收用户提供的配置数据。
`input` 可以是下述类型
- Bucket配置信息相关的类,如 `BucketCors`
- unicode类型(对于Python3则是str类型)
- 经过utf-8编码的bytes类型
- file-like object
- 可迭代类型,会通过Chunked Encoding传输
也就是说 `input` 参数可以比 `data` 参数多接受第一种类型的输入。
返回值
------
:class:`Service` 和 :class:`Bucket` 类的大多数方法都是返回 :class:`RequestResult <oss2.models.RequestResult>`
及其子类。`RequestResult` 包含了HTTP响应的状态码、头部以及OSS Request ID,而它的子类则包含用户真正想要的结果。例如,
`ListBucketsResult.buckets` 就是返回的Bucket信息列表;`GetObjectResult` 则是一个file-like object,可以调用 `read()` 来获取响应的
HTTP包体。
:class:`CryptoBucket`:
加密接口
-------
CryptoBucket仅提供上传下载加密数据的接口,诸如`get_object` 、 `put_object` ,返回值与Bucket相应接口一致。
异常
----
一般来说Python SDK可能会抛出三种类型的异常,这些异常都继承于 :class:`OssError <oss2.exceptions.OssError>` :
- :class:`ClientError <oss2.exceptions.ClientError>` :由于用户参数错误而引发的异常;
- :class:`ServerError <oss2.exceptions.ServerError>` 及其子类:OSS服务器返回非成功的状态码,如4xx或5xx;
- :class:`RequestError <oss2.exceptions.RequestError>` :底层requests库抛出的异常,如DNS解析错误,超时等;
当然,`Bucket.put_object_from_file` 和 `Bucket.get_object_to_file` 这类函数还会抛出文件相关的异常。
.. _byte_range:
指定下载范围
------------
诸如 :func:`get_object <Bucket.get_object>` 以及 :func:`upload_part_copy <Bucket.upload_part_copy>` 这样的函数,可以接受
`byte_range` 参数,表明读取数据的范围。该参数是一个二元tuple:(start, last)。这些接口会把它转换为Range头部的值,如:
- byte_range 为 (0, 99) 转换为 'bytes=0-99',表示读取前100个字节
- byte_range 为 (None, 99) 转换为 'bytes=-99',表示读取最后99个字节
- byte_range 为 (100, None) 转换为 'bytes=100-',表示读取第101个字节到文件结尾的部分(包含第101个字节)
分页罗列
-------
罗列各种资源的接口,如 :func:`list_buckets <Service.list_buckets>` 、 :func:`list_objects <Bucket.list_objects>` 都支持
分页查询。通过设定分页标记(如:`marker` 、 `key_marker` )的方式可以指定查询某一页。首次调用将分页标记设为空(缺省值,可以不设),
后续的调用使用返回值中的 `next_marker` 、 `next_key_marker` 等。每次调用后检查返回值中的 `is_truncated` ,其值为 `False` 说明
已经到了最后一页。
.. _line_range:
指定查询CSV文件范围
------------
诸如 :func:`select_object <Bucket.select_object>` 以及 :func:`select_object_to_file <Bucket.select_object_to_file>` 这样的函数的select_csv_params参数,可以接受
`LineRange` 参数,表明读取CSV数据的范围。该参数是一个二元tuple:(start, last):
- LineRange 为 (0, 99) 表示读取前100行
- LineRange 为 (None, 99) 表示读取最后99行
- LineRange 为 (100, None) 表示读取第101行到文件结尾的部分(包含第101行)
.. _split_range:
指定查询CSV文件范围
------------
split可以认为是切分好的大小大致相等的csv行簇。每个Split大小大致相等,这样以便更好的做到负载均衡。
诸如 :func:`select_object <Bucket.select_object>` 以及 :func:`select_object_to_file <Bucket.select_object_to_file>` 这样的函数的select_csv_params参数,可以接受
`SplitRange` 参数,表明读取CSV数据的范围。该参数是一个二元tuple:(start, last):
- SplitRange 为 (0, 9) 表示读取前10个Split
- SplitRange 为 (None, 9) 表示读取最后9个split
- SplitRange 为 (10, None) 表示读取第11个split到文件结尾的部分(包含第11个Split)
分页查询
-------
和create_csv_object_meta配合使用,有两种方法:
- 方法1:先获取文件总的行数(create_csv_object_meta返回),然后把文件以line_range分成若干部分并行查询
- 方法2:先获取文件总的Split数(create_csv_object_meta返回), 然后把文件分成若干个请求,每个请求含有大致相等的Split
.. _progress_callback:
上传下载进度
-----------
上传下载接口,诸如 `get_object` 、 `put_object` 、`resumable_upload`,都支持进度回调函数,可以用它实现进度条等功能。
`progress_callback` 的函数原型如下 ::
def progress_callback(bytes_consumed, total_bytes):
'''进度回调函数。
:param int bytes_consumed: 已经消费的字节数。对于上传,就是已经上传的量;对于下载,就是已经下载的量。
:param int total_bytes: 总长度。
'''
其中 `total_bytes` 对于上传和下载有不同的含义:
- 上传:当输入是bytes或可以seek/tell的文件对象,那么它的值就是总的字节数;否则,其值为None
- 下载:当返回的HTTP相应中有Content-Length头部,那么它的值就是Content-Length的值;否则,其值为None
.. _unix_time:
Unix Time
---------
OSS Python SDK会把从服务器获得时间戳都转换为自1970年1月1日UTC零点以来的秒数,即Unix Time。
参见 `Unix Time <https://en.wikipedia.org/wiki/Unix_time>`_
OSS中常用的时间格式有
- HTTP Date格式,形如 `Sat, 05 Dec 2015 11:04:39 GMT` 这样的GMT时间。
用在If-Modified-Since、Last-Modified这些HTTP请求、响应头里。
- ISO8601格式,形如 `2015-12-05T00:00:00.000Z`。
用在生命周期管理配置、列举Bucket结果中的创建时间、列举文件结果中的最后修改时间等处。
`http_date` 函数把Unix Time转换为HTTP Date;而 `http_to_unixtime` 则做相反的转换。如 ::
>>> import oss2, time
>>> unix_time = int(time.time()) # 当前UNIX Time,设其值为 1449313829
>>> date_str = oss2.http_date(unix_time) # 得到 'Sat, 05 Dec 2015 11:10:29 GMT'
>>> oss2.http_to_unixtime(date_str) # 得到 1449313829
.. note::
生成HTTP协议所需的日期(即HTTP Date)时,请使用 `http_date` , 不要使用 `strftime` 这样的函数。因为后者是和locale相关的。
比如,`strftime` 结果中可能会出现中文,而这样的格式,OSS服务器是不能识别的。
`iso8601_to_unixtime` 把ISO8601格式转换为Unix Time;`date_to_iso8601` 和 `iso8601_to_date` 则在ISO8601格式的字符串和
datetime.date之间相互转换。如 ::
>>> import oss2
>>> d = oss2.iso8601_to_date('2015-12-05T00:00:00.000Z') # 得到 datetime.date(2015, 12, 5)
>>> date_str = oss2.date_to_iso8601(d) # 得到 '2015-12-05T00:00:00.000Z'
>>> oss2.iso8601_to_unixtime(date_str) # 得到 1449273600
.. _select_params:
指定OSS Select的文件格式。
对于Csv文件,支持如下Keys:
>>> CsvHeaderInfo: None|Use|Ignore #None表示没有CSV Schema头,Use表示启用CSV Schema头,可以在Select语句中使用Name,Ignore表示有CSV Schema头,但忽略它(Select语句中不可以使用Name)
默认值是None
>>> CommentCharacter: Comment字符,默认值是#,不支持多个字符
>>> RecordDelimiter: 行分隔符,默认是\n,最多支持两个字符分隔符(比如:\r\n)
>>> FieldDelimiter: 列分隔符,默认是逗号(,), 不支持多个字符
>>> QuoteCharacter: 列Quote字符,默认是双引号("),不支持多个字符。注意转义符合Quote字符相同。
>>> LineRange: 指定查询CSV文件的行范围,参见 `line_range`。
>>> SplitRange: 指定查询CSV文件的Split范围,参见 `split_range`.
注意LineRange和SplitRange两种不能同时指定。若同时指定LineRange会被忽略。
>>> CompressionType: 文件的压缩格式,默认值是None, 支持GZIP。
>>> OutputRawData: 指定是响应Body返回Raw数据,默认值是False.
>>> SkipPartialDataRecord: 当CSV行数据不完整时(select语句中出现的列在该行为空),是否跳过该行。默认是False。
>>> OutputHeader:是否输出CSV Header,默认是False.
>>> EnablePayloadCrc:是否启用对Payload的CRC校验,默认是False. 该选项不能和OutputRawData:True混用。
>>> MaxSkippedRecordsAllowed: 允许跳过的最大行数。默认值是0表示一旦有一行跳过就报错。当下列两种情况下该行CSV被跳过:1)当SkipPartialDataRecord为True时且该行不完整时 2)当该行的数据类型和SQL不匹配时
对于Json 文件, 支持如下Keys:
>>> Json_Type: DOCUMENT | LINES . DOCUMENT就是指一般的Json文件,LINES是指每一行是一个合法的JSON对象,文件由多行Json对象组成,整个文件本身不是合法的Json对象。
>>> LineRange: 指定查询JSON LINE文件的行范围,参见 `line_range`。注意该参数仅支持LINES类型
>>> SplitRange: 指定查询JSON LINE文件的Split范围,参见 `split_range`.注意该参数仅支持LINES类型
>>> CompressionType: 文件的压缩格式,默认值是None, 支持GZIP。
>>> OutputRawData: 指定是响应Body返回Raw数据,默认值是False.
>>> SkipPartialDataRecord: 当一条JSON记录数据不完整时(select语句中出现的Key在该对象为空),是否跳过该Json记录。默认是False。
>>> EnablePayloadCrc:是否启用对Payload的CRC校验,默认是False. 该选项不能和OutputRawData:True混用。
>>> MaxSkippedRecordsAllowed: 允许跳过的最大Json记录数。默认值是0表示一旦有一条Json记录跳过就报错。当下列两种情况下该JSON被跳过:1)当SkipPartialDataRecord为True时且该条Json记录不完整时 2)当该记录的数据类型和SQL不匹配时
>>> ParseJsonNumberAsString: 将Json文件中的数字解析成字符串。使用场景是当Json文件中的浮点数精度较高时,系统默认的浮点数精度无法达到要求,当解析成字符串时将完整保留原始数据精度,在Sql中使用Cast可以将字符串无精度损失地转成decimal.
>>> AllowQuotedRecordDelimiter: 允许CSV中的列包含转义过的换行符。默认为true。当值为False时,select API可以用Range:bytes来设置选取目标对象内容的范围
‘
.. _select_meta_params:
create_select_object_meta参数集合,支持如下Keys:
- RecordDelimiter: CSV换行符,最多支持两个字符
- FieldDelimiter: CSV列分隔符,最多支持一个字符
- QuoteCharacter: CSV转移Quote符,最多支持一个字符
- OverwriteIfExists: true|false. true表示重新获得csv meta,并覆盖原有的meta。一般情况下不需要使用
"""
import logging
from . import xml_utils
from . import http
from . import utils
from . import exceptions
from . import defaults
from . import models
from . import select_params
from .models import *
from .compat import urlquote, urlparse, to_unicode, to_string
from .headers import *
from .select_params import *
import time
import shutil
logger = logging.getLogger(__name__)
class _Base(object):
def __init__(self, auth, endpoint, is_cname, session, connect_timeout,
app_name='', enable_crc=True, proxies=None, region=None, cloudbox_id= None, is_path_style=False, is_verify_object_strict=True):
self.auth = auth
self.endpoint = _normalize_endpoint(endpoint.strip())
if utils.is_valid_endpoint(self.endpoint) is not True:
raise ClientError('The endpoint you has specified is not valid, endpoint: {0}'.format(endpoint))
self.session = session or http.Session()
self.timeout = defaults.get(connect_timeout, defaults.connect_timeout)
self.app_name = app_name
self.enable_crc = enable_crc
self.proxies = proxies
self.region = region
self.product = 'oss'
self.cloudbox_id = cloudbox_id
if self.cloudbox_id is not None:
self.product = 'oss-cloudbox'
self._make_url = _UrlMaker(self.endpoint, is_cname, is_path_style)
self.is_verify_object_strict = is_verify_object_strict
if hasattr(self.auth, 'auth_version') and self.auth.auth_version() != 'v1' :
self.is_verify_object_strict = False
def _do(self, method, bucket_name, key, **kwargs):
key = to_string(key)
req = http.Request(method, self._make_url(bucket_name, key),
app_name=self.app_name,
proxies=self.proxies,
region=self.region,
product=self.product,
cloudbox_id=self.cloudbox_id,
**kwargs)
self.auth._sign_request(req, bucket_name, key)
resp = self.session.do_request(req, timeout=self.timeout)
if resp.status // 100 != 2:
e = exceptions.make_exception(resp)
logger.info("Exception: {0}".format(e))
raise e
# Note that connections are only released back to the pool for reuse once all body data has been read;
# be sure to either set stream to False or read the content property of the Response object.
# For more details, please refer to http://docs.python-requests.org/en/master/user/advanced/#keep-alive.
content_length = models._hget(resp.headers, 'content-length', int)
if content_length is not None and content_length == 0:
resp.read()
return resp
def _do_url(self, method, sign_url, **kwargs):
req = http.Request(method, sign_url, app_name=self.app_name, proxies=self.proxies, **kwargs)
resp = self.session.do_request(req, timeout=self.timeout)
if resp.status // 100 != 2:
e = exceptions.make_exception(resp)
logger.info("Exception: {0}".format(e))
raise e
# Note that connections are only released back to the pool for reuse once all body data has been read;
# be sure to either set stream to False or read the content property of the Response object.
# For more details, please refer to http://docs.python-requests.org/en/master/user/advanced/#keep-alive.
content_length = models._hget(resp.headers, 'content-length', int)
if content_length is not None and content_length == 0:
resp.read()
return resp
@staticmethod
def _parse_result(resp, parse_func, klass):
result = klass(resp)
parse_func(result, resp.read())
return result
class Service(_Base):
"""用于Service操作的类,如罗列用户所有的Bucket。
用法 ::
>>> import oss2
>>> auth = oss2.Auth('your-access-key-id', 'your-access-key-secret')
>>> service = oss2.Service(auth, 'oss-cn-hangzhou.aliyuncs.com')
>>> service.list_buckets()
<oss2.models.ListBucketsResult object at 0x0299FAB0>
:param auth: 包含了用户认证信息的Auth对象
:type auth: oss2.Auth
:param str endpoint: 访问域名,如杭州区域的域名为oss-cn-hangzhou.aliyuncs.com
:param session: 会话。如果是None表示新开会话,非None则复用传入的会话
:type session: oss2.Session
:param float connect_timeout: 连接超时时间,以秒为单位。
:param str app_name: 应用名。该参数不为空,则在User Agent中加入其值。
注意到,最终这个字符串是要作为HTTP Header的值传输的,所以必须要遵循HTTP标准。
"""
QOS_INFO = 'qosInfo'
REGIONS = 'regions'
WRITE_GET_OBJECT_RESPONSE = 'x-oss-write-get-object-response'
PUBLIC_ACCESS_BLOCK = 'publicAccessBlock'
REQUESTER_QOS_INFO = 'requesterQosInfo'
QOS_REQUESTER = 'qosRequester'
RESOURCE_POOL_INFO = 'resourcePoolInfo'
RESOURCE_POOL = 'resourcePool'
RESOURCE_POOL_BUCKETS = 'resourcePoolBuckets'
def __init__(self, auth, endpoint,
session=None,
connect_timeout=None,
app_name='',
proxies=None,
region=None,
cloudbox_id=None,
is_path_style=False):
logger.debug("Init oss service, endpoint: {0}, connect_timeout: {1}, app_name: {2}, proxies: {3}".format(
endpoint, connect_timeout, app_name, proxies))
super(Service, self).__init__(auth, endpoint, False, session, connect_timeout,
app_name=app_name, proxies=proxies,
region=region, cloudbox_id=cloudbox_id, is_path_style=is_path_style)
def list_buckets(self, prefix='', marker='', max_keys=100, params=None, headers=None):
"""根据前缀罗列用户的Bucket。
:param str prefix: 只罗列Bucket名为该前缀的Bucket,空串表示罗列所有的Bucket
:param str marker: 分页标志。首次调用传空串,后续使用返回值中的next_marker
:param int max_keys: 每次调用最多返回的Bucket数目
:param dict params: list操作参数,传入'tag-key','tag-value'对结果进行过滤
:param headers: 用户指定的HTTP头部。可以指定Content-Type、Content-MD5、x-oss-meta-开头的头部等。可以是dict,建议是oss2.CaseInsensitiveDict
:return: 罗列的结果
:rtype: oss2.models.ListBucketsResult
"""
logger.debug("Start to list buckets, prefix: {0}, marker: {1}, max-keys: {2}".format(prefix, marker, max_keys))
listParam = {}
listParam['prefix'] = prefix
listParam['marker'] = marker
listParam['max-keys'] = str(max_keys)
headers = http.CaseInsensitiveDict(headers)
if params is not None:
listParam.update(params)
resp = self._do('GET', '', '', params=listParam, headers=headers)
logger.debug("List buckets done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_buckets, ListBucketsResult)
def get_user_qos_info(self):
"""获取User的QoSInfo
:return: :class:`GetUserQosInfoResult <oss2.models.GetUserQosInfoResult>`
"""
logger.debug("Start to get user qos info.")
resp = self._do('GET', '', '', params={Service.QOS_INFO: ''})
logger.debug("get use qos, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_qos_info, GetUserQosInfoResult)
def describe_regions(self, regions=''):
"""查询所有支持地域或者指定地域对应的Endpoint信息,包括外网Endpoint、内网Endpoint和传输加速Endpoint。
:param str regions : 地域。
:return: :class:`DescribeRegionsResult <oss2.models.DescribeRegionsResult>`
"""
logger.debug("Start to describe regions")
resp = self._do('GET', '', '', params={Service.REGIONS: regions})
logger.debug("Describe regions done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_describe_regions, DescribeRegionsResult)
def write_get_object_response(self, route, token, fwd_status, data, headers=None):
"""write get object response.
:param route: fc return route
:param token: fc return token
:param fwd_status: fwd_status
:param data: 待上传的内容。
:type data: bytes,str或file-like object
:param headers: 用户指定的HTTP头部。可以指定Content-Type、Content-MD5、x-oss-meta-开头的头部等
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to write get object response, headers: {0}".format(headers))
headers = http.CaseInsensitiveDict(headers)
if route:
headers['x-oss-request-route'] = route
if token:
headers['x-oss-request-token'] = token
if fwd_status:
headers['x-oss-fwd-status'] = fwd_status
resp = self._do('POST', '', '', params={Service.WRITE_GET_OBJECT_RESPONSE: ''}, headers=headers, data=data)
logger.debug("write get object response done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def list_user_data_redundancy_transition(self, continuation_token='', max_keys=100):
"""列举请求者所有的存储冗余转换任务。
:param str continuation_token: 分页标志,首次调用传空串
:param int max_keys: 最多返回数目
:return: :class:`ListUserDataRedundancyTransitionResult <oss2.models.ListUserDataRedundancyTransitionResult>`
"""
logger.debug("Start to list user data redundancy transition, continuation token: {0}, max keys: {1}".format(continuation_token, max_keys))
resp = self._do('GET', '', '', params={Bucket.REDUNDANCY_TRANSITION: '', 'continuation-token': continuation_token, 'max-keys': str(max_keys)})
logger.debug("List user data redundancy transition done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_user_data_redundancy_transition, ListUserDataRedundancyTransitionResult)
def list_access_points(self, max_keys=100, continuation_token=''):
"""查询某个Bucket下所有接入点。
param: int max_keys: 本次list返回access point的最大个数
param: str continuation_token: list时指定的起始标记
:return: :class:`ListBucketStyleResult <oss2.models.ListBucketStyleResult>`
"""
logger.debug("Start to list bucket access point")
resp = self._do('GET', '', '', params={Bucket.ACCESS_POINT: '', 'max-keys': str(max_keys), 'continuation-token': continuation_token})
logger.debug("query list access point done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_access_point_result, ListAccessPointResult)
def put_public_access_block(self, block_public_access=False):
"""为OSS全局开启阻止公共访问。
:param bool block_public_access : 是否开启阻止公共访问。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put public access block")
data = xml_utils.to_put_public_access_block_request(block_public_access)
resp = self._do('PUT', '', '', data=data, params={Service.PUBLIC_ACCESS_BLOCK: ''})
logger.debug("Put public access block done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_public_access_block(self):
"""获取OSS全局阻止公共访问的配置信息。
:return: :class:`GetPublicAccessBlockResult <oss2.models.GetPublicAccessBlockResult>`
"""
logger.debug("Start to get public access block")
resp = self._do('GET', '', '', params={Service.PUBLIC_ACCESS_BLOCK: ''})
logger.debug("Get public access block done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_public_access_block_result, GetPublicAccessBlockResult)
def delete_public_access_block(self):
"""删除OSS全局阻止公共访问配置信息。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete public access block")
resp = self._do('DELETE', '', '', params={Service.PUBLIC_ACCESS_BLOCK: ''})
logger.debug("Delete public access block done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def list_resource_pools(self, continuation_token='', max_keys=100):
"""列举当前主账号在当前地域下的资源池。
:param str continuation_token: 分页标志,首次调用传空串
:param int max_keys: 最多返回数目
:return: :class:`ListResourcePoolsResult <oss2.models.ListResourcePoolsResult>`
"""
logger.debug("Start to list resource pools, continuation_token: {0}, max_keys: {1}".format(continuation_token, max_keys))
resp = self._do('GET', '', '', params={Service.RESOURCE_POOL: '', 'continuation-token': continuation_token, 'max-keys': str(max_keys)})
logger.debug("List resource pools done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_resource_pools, ListResourcePoolsResult)
def get_resource_pool_info(self, resource_pool_name):
"""获取特定资源池的基本信息。
:param str resource_pool_name : 请求的资源池的名称。
:return: :class:`ResourcePoolInfoResult <oss2.models.ResourcePoolInfoResult>`
"""
logger.debug("Start to get resource pool info, uid: {0}.".format(resource_pool_name))
if not resource_pool_name:
raise ClientError('resource_pool_name should not be empty')
resp = self._do('GET', '', '', params={Service.RESOURCE_POOL_INFO: '', Service.RESOURCE_POOL: resource_pool_name})
logger.debug("Get resource pool info done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_resource_pool_info, ResourcePoolInfoResult)
def list_resource_pool_buckets(self, resource_pool_name, continuation_token='', max_keys=100):
"""获取特定资源池中的Bucket列表。
:param str resource_pool_name : 请求的资源池的名称。
:param str continuation_token: 分页标志,首次调用传空串
:param int max_keys: 最多返回数目
:return: :class:`ListResourcePoolBucketsResult <oss2.models.ListResourcePoolBucketsResult>`
"""
logger.debug("Start to list resource pool buckets, resource_pool_name:{0} continuation_token: {1}, max_keys: {2}".format(resource_pool_name, continuation_token, max_keys))
if not resource_pool_name:
raise ClientError('resource_pool_name should not be empty')
resp = self._do('GET', '', '', params={Service.RESOURCE_POOL_BUCKETS: '', Service.RESOURCE_POOL: resource_pool_name, 'continuation-token': continuation_token, 'max-keys': str(max_keys)})
logger.debug("List resource pool buckets done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_resource_pool_buckets, ListResourcePoolBucketsResult)
def put_resource_pool_requester_qos_info(self, uid, resource_pool_name, qos_configuration):
"""修改子账号在资源池的请求者流控配置。
:param str uid: 请求者UID
:param str resource_pool_name: 请求的资源池的名称
:param qos_configuration :class:`QoSConfiguration <oss2.models.QoSConfiguration>`
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put resource pool requester qos info, uid: {0}, resource_pool_name: {1}, qos_configuration: {2}".format(uid, resource_pool_name, qos_configuration))
if not uid:
raise ClientError('uid should not be empty')
if not resource_pool_name:
raise ClientError('resource_pool_name should not be empty')
data = xml_utils.to_put_qos_info(qos_configuration)
resp = self._do('PUT', '', '', data=data, params={Service.REQUESTER_QOS_INFO: '', Service.QOS_REQUESTER: uid, Service.RESOURCE_POOL: resource_pool_name})
logger.debug("Put resource pool requester qos info done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_resource_pool_requester_qos_info(self, uid, resource_pool_name):
"""获取子账号在资源池的流控配置。
:return: :class:`RequesterQoSInfoResult <oss2.models.RequesterQoSInfoResult>`
"""
logger.debug("Start to get resource pool requester qos info, uid: {0}, resource_pool_name: {1}.".format(uid, resource_pool_name))
if not uid:
raise ClientError('uid should not be empty')
if not resource_pool_name:
raise ClientError('resource_pool_name should not be empty')
resp = self._do('GET', '', '', params={Service.REQUESTER_QOS_INFO: '', Service.QOS_REQUESTER: uid, Service.RESOURCE_POOL: resource_pool_name})
logger.debug("Get resource pool requester qos info done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_requester_qos_info, RequesterQoSInfoResult)
def list_resource_pool_requester_qos_infos(self, resource_pool_name, continuation_token='', max_keys=100):
"""列举子账号账号在资源池的流控配置。
:param str resource_pool_name : 请求的资源池的名称。
:param str continuation_token: 分页标志,首次调用传空串
:param int max_keys: 最多返回数目
:return: :class:`ListResourcePoolRequesterQoSInfosResult <oss2.models.ListResourcePoolRequesterQoSInfosResult>`
"""
logger.debug("Start to list resource pool requester qos infos, resource_pool_name:{0} continuation_token: {1}, max_keys: {2}".format(resource_pool_name, continuation_token, max_keys))
if not resource_pool_name:
raise ClientError('resource_pool_name should not be empty')
resp = self._do('GET', '', '', params={Service.REQUESTER_QOS_INFO: '', Service.RESOURCE_POOL: resource_pool_name, 'continuation-token': continuation_token, 'max-keys': str(max_keys)})
logger.debug("List resource pool requester qos infos done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_resource_pool_requester_qos_infos, ListResourcePoolRequesterQoSInfosResult)
def delete_resource_pool_requester_qos_info(self, uid, resource_pool_name):
"""删除子账号在资源池的请求者流控配置。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete resource pool requester qos info, uid: {0}, resource_pool_name: {1}.".format(uid, resource_pool_name))
if not uid:
raise ClientError('uid should not be empty')
if not resource_pool_name:
raise ClientError('resource_pool_name should not be empty')
resp = self._do('DELETE', '', '', params={Service.REQUESTER_QOS_INFO: '', Service.QOS_REQUESTER: uid, Service.RESOURCE_POOL: resource_pool_name})
logger.debug("Delete resource pool requester qos info done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
class Bucket(_Base):
"""用于Bucket和Object操作的类,诸如创建、删除Bucket,上传、下载Object等。
用法(假设Bucket属于杭州区域) ::
>>> import oss2
>>> auth = oss2.Auth('your-access-key-id', 'your-access-key-secret')
>>> bucket = oss2.Bucket(auth, 'http://oss-cn-hangzhou.aliyuncs.com', 'your-bucket')
>>> bucket.put_object('readme.txt', 'content of the object')
<oss2.models.PutObjectResult object at 0x029B9930>
:param auth: 包含了用户认证信息的Auth对象
:type auth: oss2.ProviderAuth or oss2.ProviderAuthV4
:param str endpoint: 访问域名或者CNAME
:param str bucket_name: Bucket名
:param bool is_cname: 如果endpoint是CNAME则设为True;反之,则为False。
:param session: 会话。如果是None表示新开会话,非None则复用传入的会话
:type session: oss2.Session
:param float connect_timeout: 连接超时时间,以秒为单位。
:param str app_name: 应用名。该参数不为空,则在User Agent中加入其值。
注意到,最终这个字符串是要作为HTTP Header的值传输的,所以必须要遵循HTTP标准。
:param bool is_verify_object_strict: 严格验证对象名称的标志。默认为True。
"""
ACL = 'acl'
CORS = 'cors'
LIFECYCLE = 'lifecycle'
LOCATION = 'location'
LOGGING = 'logging'
REFERER = 'referer'
WEBSITE = 'website'
LIVE = 'live'
COMP = 'comp'
STATUS = 'status'
VOD = 'vod'
SYMLINK = 'symlink'
STAT = 'stat'
BUCKET_INFO = 'bucketInfo'
PROCESS = 'x-oss-process'
TAGGING = 'tagging'
ENCRYPTION = 'encryption'
VERSIONS = 'versions'
VERSIONING = 'versioning'
VERSIONID = 'versionId'
RESTORE = 'restore'
OBJECTMETA = 'objectMeta'
POLICY = 'policy'
REQUESTPAYMENT = 'requestPayment'
QOS_INFO = 'qosInfo'
USER_QOS = 'qos'
ASYNC_FETCH = 'asyncFetch'
SEQUENTIAL = 'sequential'
INVENTORY = "inventory"
INVENTORY_CONFIG_ID = "inventoryId"
CONTINUATION_TOKEN = "continuation-token"
WORM = "worm"
WORM_ID = "wormId"
WORM_EXTEND = "wormExtend"
REPLICATION = "replication"
REPLICATION_LOCATION = 'replicationLocation'
REPLICATION_PROGRESS = 'replicationProgress'
TRANSFER_ACCELERATION = 'transferAcceleration'
CNAME = 'cname'
META_QUERY = 'metaQuery'
ACCESS_MONITOR = 'accessmonitor'
RESOURCE_GROUP = 'resourceGroup'
STYLE = 'style'
STYLE_NAME = 'styleName'
ASYNC_PROCESS = 'x-oss-async-process'
CALLBACK = 'callback'
ARCHIVE_DIRECT_READ = "bucketArchiveDirectRead"
HTTPS_CONFIG = 'httpsConfig'
REDUNDANCY_TRANSITION = "redundancyTransition"
TARGET_REDUNDANCY_TYPE = "x-oss-target-redundancy-type"
REDUNDANCY_TRANSITION_TASK_ID = "x-oss-redundancy-transition-taskid"
ACCESS_POINT = 'accessPoint'
ACCESS_POINT_POLICY = 'accessPointPolicy'
OSS_ACCESS_POINT_NAME = 'x-oss-access-point-name'
PUBLIC_ACCESS_BLOCK = 'publicAccessBlock'
OSS_ACCESS_POINT_NAME = 'x-oss-access-point-name'
REQUESTER_QOS_INFO = 'requesterQosInfo'
QOS_REQUESTER = 'qosRequester'
RESOURCE_POOL_INFO = 'resourcePoolInfo'
RESOURCE_POOL = 'resourcePool'
def __init__(self, auth, endpoint, bucket_name,
is_cname=False,
session=None,
connect_timeout=None,
app_name='',
enable_crc=True,
proxies=None,
region=None,
cloudbox_id=None,
is_path_style=False,
is_verify_object_strict=True):
logger.debug("Init Bucket: {0}, endpoint: {1}, isCname: {2}, connect_timeout: {3}, app_name: {4}, enabled_crc: {5}, region: {6}"
", proxies: {6}".format(bucket_name, endpoint, is_cname, connect_timeout, app_name, enable_crc, proxies, region))
super(Bucket, self).__init__(auth, endpoint, is_cname, session, connect_timeout,
app_name=app_name, enable_crc=enable_crc, proxies=proxies,
region=region, cloudbox_id=cloudbox_id, is_path_style=is_path_style, is_verify_object_strict=is_verify_object_strict)
self.bucket_name = bucket_name.strip()
if utils.is_valid_bucket_name(self.bucket_name) is not True:
raise ClientError("The bucket_name is invalid, please check it.")
def sign_url(self, method, key, expires, headers=None, params=None, slash_safe=False, additional_headers=None):
"""生成签名URL。
常见的用法是生成加签的URL以供授信用户下载,如为log.jpg生成一个5分钟后过期的下载链接::
>>> bucket.sign_url('GET', 'log.jpg', 5 * 60)
r'http://your-bucket.oss-cn-hangzhou.aliyuncs.com/logo.jpg?OSSAccessKeyId=YourAccessKeyId\&Expires=1447178011&Signature=UJfeJgvcypWq6Q%2Bm3IJcSHbvSak%3D'
:param method: HTTP方法,如'GET'、'PUT'、'DELETE'等
:type method: str
:param key: 文件名
:param expires: 过期时间(单位:秒),链接在当前时间再过expires秒后过期
:param headers: 需要签名的HTTP头部,如名称以x-oss-meta-开头的头部(作为用户自定义元数据)、
Content-Type头部等。对于下载,不需要填。
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:param params: 需要签名的HTTP查询参数
:param slash_safe: 是否开启key名称中的‘/’转义保护,如果不开启'/'将会转义成%2F
:type slash_safe: bool
:param additional_headers: 额外的需要签名的HTTP头
:return: 签名URL。
"""
if key is None or len(key.strip()) <= 0:
raise ClientError("The key is invalid, please check it.")
key = to_string(key)
if self.is_verify_object_strict and key.startswith('?'):
raise ClientError("The key cannot start with `?`, please check it.")
logger.debug(
"Start to sign_url, method: {0}, bucket: {1}, key: {2}, expires: {3}, headers: {4}, params: {5}, slash_safe: {6}".format(
method, self.bucket_name, to_string(key), expires, headers, params, slash_safe))
req = http.Request(method, self._make_url(self.bucket_name, key, slash_safe),
headers=headers,
params=params,
region=self.region,
product=self.product,
cloudbox_id=self.cloudbox_id)
sign_url = ''
if additional_headers is None:
sign_url = self.auth._sign_url(req, self.bucket_name, key, expires)
else:
sign_url = self.auth._sign_url(req, self.bucket_name, key, expires, in_additional_headers=additional_headers)
return sign_url
def sign_rtmp_url(self, channel_name, playlist_name, expires):
"""生成RTMP推流的签名URL。
常见的用法是生成加签的URL以供授信用户向OSS推RTMP流。
:param channel_name: 直播频道的名称
:param expires: 过期时间(单位:秒),链接在当前时间再过expires秒后过期
:param playlist_name: 播放列表名称,注意与创建live channel时一致
:param params: 需要签名的HTTP查询参数
:return: 签名URL。
"""
logger.debug("Sign RTMP url, bucket: {0}, channel_name: {1}, playlist_name: {2}, expires: {3}".format(
self.bucket_name, channel_name, playlist_name, expires))
url = self._make_url(self.bucket_name, 'live').replace('http://', 'rtmp://').replace(
'https://', 'rtmp://') + '/' + channel_name
params = {}
if playlist_name is not None and playlist_name != "":
params['playlistName'] = playlist_name
return self.auth._sign_rtmp_url(url, self.bucket_name, channel_name, expires, params)
def list_objects(self, prefix='', delimiter='', marker='', max_keys=100, headers=None):
"""根据前缀罗列Bucket里的文件。
:param str prefix: 只罗列文件名为该前缀的文件
:param str delimiter: 分隔符。可以用来模拟目录
:param str marker: 分页标志。首次调用传空串,后续使用返回值的next_marker
:param int max_keys: 最多返回文件的个数,文件和目录的和不能超过该值
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`ListObjectsResult <oss2.models.ListObjectsResult>`
"""
headers = http.CaseInsensitiveDict(headers)
logger.debug(
"Start to List objects, bucket: {0}, prefix: {1}, delimiter: {2}, marker: {3}, max-keys: {4}".format(
self.bucket_name, to_string(prefix), delimiter, to_string(marker), max_keys))
resp = self.__do_bucket('GET',
params={'prefix': prefix,
'delimiter': delimiter,
'marker': marker,
'max-keys': str(max_keys),
'encoding-type': 'url'},
headers=headers)
logger.debug("List objects done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_objects, ListObjectsResult)
def list_objects_v2(self, prefix='', delimiter='', continuation_token='', start_after='', fetch_owner=False, encoding_type='url', max_keys=100, headers=None):
"""根据前缀罗列Bucket里的文件。
:param str prefix: 只罗列文件名为该前缀的文件
:param str delimiter: 分隔符。可以用来模拟目录
:param str continuation_token: 分页标志。首次调用传空串,后续使用返回值的next_continuation_token
:param str start_after: 起始文件名称,OSS会返回按照字典序排列start_after之后的文件。
:param bool fetch_owner: 是否获取文件的owner信息,默认不返回。
:param int max_keys: 最多返回文件的个数,文件和目录的和不能超过该值
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`ListObjectsV2Result <oss2.models.ListObjectsV2Result>`
"""
headers = http.CaseInsensitiveDict(headers)
logger.debug(
"Start to List objects, bucket: {0}, prefix: {1}, delimiter: {2}, continuation_token: {3}, "
"start-after: {4}, fetch-owner: {5}, encoding_type: {6}, max-keys: {7}".format(
self.bucket_name, to_string(prefix), delimiter, continuation_token, start_after, fetch_owner, encoding_type, max_keys))
resp = self.__do_bucket('GET',
params={'list-type': '2',
'prefix': prefix,
'delimiter': delimiter,
'continuation-token': continuation_token,
'start-after': start_after,
'fetch-owner': str(fetch_owner).lower(),
'max-keys': str(max_keys),
'encoding-type': encoding_type},
headers=headers)
logger.debug("List objects V2 done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_objects_v2, ListObjectsV2Result)
def put_object(self, key, data,
headers=None,
progress_callback=None):
"""上传一个普通文件。
用法 ::
>>> bucket.put_object('readme.txt', 'content of readme.txt')
>>> with open(u'local_file.txt', 'rb') as f:
>>> bucket.put_object('remote_file.txt', f)
:param key: 上传到OSS的文件名
:param data: 待上传的内容。
:type data: bytes,str或file-like object
:param headers: 用户指定的HTTP头部。可以指定Content-Type、Content-MD5、x-oss-meta-开头的头部等
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:param progress_callback: 用户指定的进度回调函数。可以用来实现进度条等功能。参考 :ref:`progress_callback` 。
:return: :class:`PutObjectResult <oss2.models.PutObjectResult>`
"""
headers = utils.set_content_type(http.CaseInsensitiveDict(headers), key)
if progress_callback:
data = utils.make_progress_adapter(data, progress_callback)
if self.enable_crc:
data = utils.make_crc_adapter(data)
logger.debug("Start to put object, bucket: {0}, key: {1}, headers: {2}".format(self.bucket_name, to_string(key),
headers))
resp = self.__do_object('PUT', key, data=data, headers=headers)
logger.debug("Put object done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
result = PutObjectResult(resp)
if self.enable_crc and result.crc is not None:
utils.check_crc('put object', data.crc, result.crc, result.request_id)
return result
def put_object_from_file(self, key, filename,
headers=None,
progress_callback=None):
"""上传一个本地文件到OSS的普通文件。
:param str key: 上传到OSS的文件名
:param str filename: 本地文件名,需要有可读权限
:param headers: 用户指定的HTTP头部。可以指定Content-Type、Content-MD5、x-oss-meta-开头的头部等
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:param progress_callback: 用户指定的进度回调函数。参考 :ref:`progress_callback`
:return: :class:`PutObjectResult <oss2.models.PutObjectResult>`
"""
headers = utils.set_content_type(http.CaseInsensitiveDict(headers), filename)
logger.debug("Put object from file, bucket: {0}, key: {1}, file path: {2}".format(
self.bucket_name, to_string(key), filename))
with open(to_unicode(filename), 'rb') as f:
return self.put_object(key, f, headers=headers, progress_callback=progress_callback)
def put_object_with_url(self, sign_url, data, headers=None, progress_callback=None):
""" 使用加签的url上传对象
:param sign_url: 加签的url
:param data: 待上传的数据
:param headers: 用户指定的HTTP头部。可以指定Content-Type、Content-MD5、x-oss-meta-开头的头部等,必须和签名时保持一致
:param progress_callback: 用户指定的进度回调函数。参考 :ref:`progress_callback`
:return:
"""
headers = http.CaseInsensitiveDict(headers)
if progress_callback:
data = utils.make_progress_adapter(data, progress_callback)
if self.enable_crc:
data = utils.make_crc_adapter(data)
logger.debug("Start to put object with signed url, bucket: {0}, sign_url: {1}, headers: {2}".format(
self.bucket_name, sign_url, headers))
resp = self._do_url('PUT', sign_url, data=data, headers=headers)
logger.debug("Put object with url done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
result = PutObjectResult(resp)
if self.enable_crc and result.crc is not None:
utils.check_crc('put object', data.crc, result.crc, result.request_id)
return result
def put_object_with_url_from_file(self, sign_url, filename,
headers=None,
progress_callback=None):
""" 使用加签的url上传本地文件到oss
:param sign_url: 加签的url
:param filename: 本地文件路径
:param headers: 用户指定的HTTP头部。可以指定Content-Type、Content-MD5、x-oss-meta-开头的头部等,必须和签名时保持一致
:param progress_callback: 用户指定的进度回调函数。参考 :ref:`progress_callback`
:return:
"""
logger.debug("Put object from file with signed url, bucket: {0}, sign_url: {1}, file path: {2}".format(
self.bucket_name, sign_url, filename))
with open(to_unicode(filename), 'rb') as f:
return self.put_object_with_url(sign_url, f, headers=headers, progress_callback=progress_callback)
def append_object(self, key, position, data,
headers=None,
progress_callback=None,
init_crc=None):
"""追加上传一个文件。
:param str key: 新的文件名,或已经存在的可追加文件名
:param int position: 追加上传一个新的文件, `position` 设为0;追加一个已经存在的可追加文件, `position` 设为文件的当前长度。
`position` 可以从上次追加的结果 `AppendObjectResult.next_position` 中获得。
:param data: 用户数据
:type data: str、bytes、file-like object或可迭代对象
:param headers: 用户指定的HTTP头部。可以指定Content-Type、Content-MD5、x-oss-开头的头部等
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:param progress_callback: 用户指定的进度回调函数。参考 :ref:`progress_callback`
:return: :class:`AppendObjectResult <oss2.models.AppendObjectResult>`
:raises: 如果 `position` 和当前文件长度不一致,抛出 :class:`PositionNotEqualToLength <oss2.exceptions.PositionNotEqualToLength>` ;
如果当前文件不是可追加类型,抛出 :class:`ObjectNotAppendable <oss2.exceptions.ObjectNotAppendable>` ;
还会抛出其他一些异常
"""
headers = utils.set_content_type(http.CaseInsensitiveDict(headers), key)
if progress_callback:
data = utils.make_progress_adapter(data, progress_callback)
if self.enable_crc and init_crc is not None:
data = utils.make_crc_adapter(data, init_crc)
logger.debug("Start to append object, bucket: {0}, key: {1}, headers: {2}, position: {3}".format(
self.bucket_name, to_string(key), headers, position))
resp = self.__do_object('POST', key,
data=data,
headers=headers,
params={'append': '', 'position': str(position)})
logger.debug("Append object done, req_id: {0}, statu_code: {1}".format(resp.request_id, resp.status))
result = AppendObjectResult(resp)
if self.enable_crc and result.crc is not None and init_crc is not None:
utils.check_crc('append object', data.crc, result.crc, result.request_id)
return result
def get_object(self, key,
byte_range=None,
headers=None,
progress_callback=None,
process=None,
params=None):
"""下载一个文件。
用法 ::
>>> result = bucket.get_object('readme.txt')
>>> print(result.read())
'hello world'
:param key: 文件名
:param byte_range: 指定下载范围。参见 :ref:`byte_range`
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:param progress_callback: 用户指定的进度回调函数。参考 :ref:`progress_callback`
:param process: oss文件处理,如图像服务等。指定后process,返回的内容为处理后的文件。
:param params: http 请求的查询字符串参数
:type params: dict
:return: file-like object
:raises: 如果文件不存在,则抛出 :class:`NoSuchKey <oss2.exceptions.NoSuchKey>` ;还可能抛出其他异常
"""
headers = http.CaseInsensitiveDict(headers)
range_string = _make_range_string(byte_range)
if range_string:
headers['range'] = range_string
params = {} if params is None else params
if process:
params.update({Bucket.PROCESS: process})
logger.debug("Start to get object, bucket: {0}, key: {1}, range: {2}, headers: {3}, params: {4}".format(
self.bucket_name, to_string(key), range_string, headers, params))
resp = self.__do_object('GET', key, headers=headers, params=params)
logger.debug("Get object done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return GetObjectResult(resp, progress_callback, self.enable_crc)
def select_object(self, key, sql,
progress_callback=None,
select_params=None,
byte_range=None,
headers=None
):
"""Select一个文件内容,支持(Csv,Json Doc,Json Lines及其GZIP压缩文件).
用法 ::
对于Csv:
>>> result = bucket.select_object('access.log', 'select * from ossobject where _4 > 40')
>>> print(result.read())
'hello world'
对于Json Doc: { contacts:[{"firstName":"abc", "lastName":"def"},{"firstName":"abc1", "lastName":"def1"}]}
>>> result = bucket.select_object('sample.json', 'select s.firstName, s.lastName from ossobject.contacts[*] s', select_params = {"Json_Type":"DOCUMENT"})
对于Json Lines: {"firstName":"abc", "lastName":"def"},{"firstName":"abc1", "lastName":"def1"}
>>> result = bucket.select_object('sample.json', 'select s.firstName, s.lastName from ossobject s', select_params = {"Json_Type":"LINES"})
:param key: 文件名
:param sql: sql statement
:param select_params: select参数集合,对于Json文件必须制定Json_Type类型。参见 :ref:`select_params`
:param progress_callback: 用户指定的进度回调函数。参考 :ref:`progress_callback`
:param byte_range: select content of specific range。可以设置Bytes header指定select csv时的文件起始offset和长度。
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: file-like object
:raises: 如果文件不存在,则抛出 :class:`NoSuchKey <oss2.exceptions.NoSuchKey>` ;还可能抛出其他异常
"""
range_select = False
headers = http.CaseInsensitiveDict(headers)
range_string = _make_range_string(byte_range)
if range_string:
headers['range'] = range_string
range_select = True
if (range_select == True and
(select_params is None or
(SelectParameters.AllowQuotedRecordDelimiter not in select_params or str(select_params[SelectParameters.AllowQuotedRecordDelimiter]).lower() != 'false'))):
raise ClientError('"AllowQuotedRecordDelimiter" must be specified in select_params as False when "Range" is specified in header.')
body = xml_utils.to_select_object(sql, select_params)
params = {'x-oss-process': 'csv/select'}
if select_params is not None and SelectParameters.Json_Type in select_params:
params['x-oss-process'] = 'json/select'
self.timeout = 3600
resp = self.__do_object('POST', key, data=body, headers=headers, params=params)
crc_enabled = False
if select_params is not None and SelectParameters.EnablePayloadCrc in select_params:
if str(select_params[SelectParameters.EnablePayloadCrc]).lower() == "true":
crc_enabled = True
return SelectObjectResult(resp, progress_callback, crc_enabled)
def get_object_to_file(self, key, filename,
byte_range=None,
headers=None,
progress_callback=None,
process=None,
params=None):
"""下载一个文件到本地文件。
:param key: 文件名
:param filename: 本地文件名。要求父目录已经存在,且有写权限。
:param byte_range: 指定下载范围。参见 :ref:`byte_range`
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:param progress_callback: 用户指定的进度回调函数。参考 :ref:`progress_callback`
:param process: oss文件处理,如图像服务等。指定后process,返回的内容为处理后的文件。
:param params: http 请求的查询字符串参数
:type params: dict
:return: 如果文件不存在,则抛出 :class:`NoSuchKey <oss2.exceptions.NoSuchKey>` ;还可能抛出其他异常
"""
logger.debug("Start to get object to file, bucket: {0}, key: {1}, file path: {2}".format(
self.bucket_name, to_string(key), filename))
with open(to_unicode(filename), 'wb') as f:
result = self.get_object(key, byte_range=byte_range, headers=headers, progress_callback=progress_callback,
process=process, params=params)
if result.content_length is None:
shutil.copyfileobj(result, f)
else:
utils.copyfileobj_and_verify(result, f, result.content_length, request_id=result.request_id)
if self.enable_crc and byte_range is None:
if (headers is None) or ('Accept-Encoding' not in headers) or (headers['Accept-Encoding'] != 'gzip'):
utils.check_crc('get', result.client_crc, result.server_crc, result.request_id)
return result
def get_object_with_url(self, sign_url,
byte_range=None,
headers=None,
progress_callback=None):
"""使用加签的url下载文件
:param sign_url: 加签的url
:param byte_range: 指定下载范围。参见 :ref:`byte_range`
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict,必须和签名时保持一致
:param progress_callback: 用户指定的进度回调函数。参考 :ref:`progress_callback`
:return: file-like object
:raises: 如果文件不存在,则抛出 :class:`NoSuchKey <oss2.exceptions.NoSuchKey>` ;还可能抛出其他异常
"""
headers = http.CaseInsensitiveDict(headers)
range_string = _make_range_string(byte_range)
if range_string:
headers['range'] = range_string
logger.debug("Start to get object with url, bucket: {0}, sign_url: {1}, range: {2}, headers: {3}".format(
self.bucket_name, sign_url, range_string, headers))
resp = self._do_url('GET', sign_url, headers=headers)
return GetObjectResult(resp, progress_callback, self.enable_crc)
def get_object_with_url_to_file(self, sign_url,
filename,
byte_range=None,
headers=None,
progress_callback=None):
"""使用加签的url下载文件
:param sign_url: 加签的url
:param filename: 本地文件名。要求父目录已经存在,且有写权限。
:param byte_range: 指定下载范围。参见 :ref:`byte_range`
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict,,必须和签名时保持一致
:param progress_callback: 用户指定的进度回调函数。参考 :ref:`progress_callback`
:return: file-like object
:raises: 如果文件不存在,则抛出 :class:`NoSuchKey <oss2.exceptions.NoSuchKey>` ;还可能抛出其他异常
"""
logger.debug(
"Start to get object with url, bucket: {0}, sign_url: {1}, file path: {2}, range: {3}, headers: {4}"
.format(self.bucket_name, sign_url, filename, byte_range, headers))
with open(to_unicode(filename), 'wb') as f:
result = self.get_object_with_url(sign_url, byte_range=byte_range, headers=headers,
progress_callback=progress_callback)
if result.content_length is None:
shutil.copyfileobj(result, f)
else:
utils.copyfileobj_and_verify(result, f, result.content_length, request_id=result.request_id)
return result
def select_object_to_file(self, key, filename, sql,
progress_callback=None,
select_params=None,
headers=None
):
"""Select一个文件的内容到本地文件
:param key: OSS文件名
:param filename: 本地文件名。其父亲目录已经存在且有写权限。
:param progress_callback: 调用进度的callback。参考 :ref:`progress_callback`
:param select_params: select参数集合。参见 :ref:`select_params`
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: 如果文件不存在, 抛出 :class:`NoSuchKey <oss2.exceptions.NoSuchKey>`
"""
with open(to_unicode(filename), 'wb') as f:
result = self.select_object(key, sql, progress_callback=progress_callback,
select_params=select_params, headers=headers)
for chunk in result:
f.write(chunk)
return result
def head_object(self, key, headers=None, params=None):
"""获取文件元信息。
HTTP响应的头部包含了文件元信息,可以通过 `RequestResult` 的 `headers` 成员获得。
用法 ::
>>> result = bucket.head_object('readme.txt')
>>> print(result.content_type)
text/plain
:param key: 文件名
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:param params: HTTP请求参数,传入versionId,获取指定版本Object元信息
:type params: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`HeadObjectResult <oss2.models.HeadObjectResult>`
:raises: 如果Bucket不存在或者Object不存在,则抛出 :class:`NotFound <oss2.exceptions.NotFound>`
"""
logger.debug("Start to head object, bucket: {0}, key: {1}, headers: {2}".format(
self.bucket_name, to_string(key), headers))
resp = self.__do_object('HEAD', key, headers=headers, params=params)
logger.debug("Head object done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_dummy_result, HeadObjectResult)
def create_select_object_meta(self, key, select_meta_params=None, headers=None):
"""获取或创建CSV,JSON LINES 文件元信息。如果元信息存在,返回之;不然则创建后返回之
HTTP响应的头部包含了文件元信息,可以通过 `RequestResult` 的 `headers` 成员获得。
CSV文件用法 ::
>>> select_meta_params = { 'FieldDelimiter': ',',
'RecordDelimiter': '\r\n',
'QuoteCharacter': '"',
'OverwriteIfExists' : 'false'}
>>> result = bucket.create_select_object_meta('csv.txt', select_meta_params)
>>> print(result.rows)
JSON LINES文件用法 ::
>>> select_meta_params = { 'Json_Type':'LINES', 'OverwriteIfExists':'False'}
>>> result = bucket.create_select_object_meta('jsonlines.json', select_meta_params)
:param key: 文件名
:param select_meta_params: 参数词典,可以是dict,参见ref:`csv_meta_params`
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`GetSelectObjectMetaResult <oss2.models.HeadObjectResult>`.
除了 rows 和splits 属性之外, 它也返回head object返回的其他属性。
rows表示该文件的总记录数。
splits表示该文件的总Split个数,一个Split包含若干条记录,每个Split的总字节数大致相当。用户可以以Split为单位进行分片查询。
:raises: 如果Bucket不存在或者Object不存在,则抛出:class:`NotFound <oss2.exceptions.NotFound>`
"""
headers = http.CaseInsensitiveDict(headers)
body = xml_utils.to_get_select_object_meta(select_meta_params)
params = {'x-oss-process': 'csv/meta'}
if select_meta_params is not None and 'Json_Type' in select_meta_params:
params['x-oss-process'] = 'json/meta'
self.timeout = 3600
resp = self.__do_object('POST', key, data=body, headers=headers, params=params)
return GetSelectObjectMetaResult(resp)
def get_object_meta(self, key, params=None, headers=None):
"""获取文件基本元信息,包括该Object的ETag、Size(文件大小)、LastModified,并不返回其内容。
HTTP响应的头部包含了文件基本元信息,可以通过 `GetObjectMetaResult` 的 `last_modified`,`content_length`,`etag` 成员获得。
:param key: 文件名
:param dict params: 请求参数
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`GetObjectMetaResult <oss2.models.GetObjectMetaResult>`
:raises: 如果文件不存在,则抛出 :class:`NoSuchKey <oss2.exceptions.NoSuchKey>` ;还可能抛出其他异常
"""
headers = http.CaseInsensitiveDict(headers)
logger.debug("Start to get object metadata, bucket: {0}, key: {1}".format(self.bucket_name, to_string(key)))
if params is None:
params = dict()
if Bucket.OBJECTMETA not in params:
params[Bucket.OBJECTMETA] = ''
resp = self.__do_object('HEAD', key, params=params, headers=headers)
logger.debug("Get object metadata done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return GetObjectMetaResult(resp)
def object_exists(self, key, headers=None):
"""如果文件存在就返回True,否则返回False。如果Bucket不存在,或是发生其他错误,则抛出异常。"""
#:param key: 文件名
#:param headers: HTTP头部
#:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
# 如果我们用head_object来实现的话,由于HTTP HEAD请求没有响应体,只有响应头部,这样当发生404时,
# 我们无法区分是NoSuchBucket还是NoSuchKey错误。
#
# 2.2.0之前的实现是通过get_object的if-modified-since头部,把date设为当前时间24小时后,这样如果文件存在,则会返回
# 304 (NotModified);不存在,则会返回NoSuchKey。get_object会受回源的影响,如果配置会404回源,get_object会判断错误。
#
# 目前的实现是通过get_object_meta判断文件是否存在。
# get_object_meta 为200时,不会返回响应体,所以该接口把GET方法修改为HEAD 方式
# 同时, 对于head 请求,服务端会通过x-oss-err 返回 错误响应信息,
# 考虑到兼容之前的行为,增加exceptions.NotFound 异常 当作NoSuchKey
logger.debug("Start to check if object exists, bucket: {0}, key: {1}".format(self.bucket_name, to_string(key)))
try:
self.get_object_meta(key, headers=headers)
except exceptions.NoSuchKey:
return False
except exceptions.NoSuchBucket:
raise
except exceptions.NotFound:
return False
except:
raise
return True
def copy_object(self, source_bucket_name, source_key, target_key, headers=None, params=None):
"""拷贝一个文件到当前Bucket。
:param str source_bucket_name: 源Bucket名
:param str source_key: 源文件名
:param str target_key: 目标文件名
:param dict params: 请求参数
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`PutObjectResult <oss2.models.PutObjectResult>`
"""
headers = http.CaseInsensitiveDict(headers)
if params and Bucket.VERSIONID in params:
headers[OSS_COPY_OBJECT_SOURCE] = '/' + source_bucket_name + \
'/' + urlquote(source_key, '') + '?versionId=' + params[Bucket.VERSIONID]
else:
headers[OSS_COPY_OBJECT_SOURCE] = '/' + source_bucket_name + '/' + urlquote(source_key, '')
logger.debug(
"Start to copy object, source bucket: {0}, source key: {1}, bucket: {2}, key: {3}, headers: {4}".format(
source_bucket_name, to_string(source_key), self.bucket_name, to_string(target_key), headers))
resp = self.__do_object('PUT', target_key, headers=headers)
logger.debug("Copy object done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return PutObjectResult(resp)
def update_object_meta(self, key, headers):
"""更改Object的元数据信息,包括Content-Type这类标准的HTTP头部,以及以x-oss-meta-开头的自定义元数据。
用户可以通过 :func:`head_object` 获得元数据信息。
:param str key: 文件名
:param headers: HTTP头部,包含了元数据信息
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`RequestResult <oss2.models.RequestResults>`
"""
if headers is not None:
headers[OSS_METADATA_DIRECTIVE] = 'REPLACE'
logger.debug("Start to update object metadata, bucket: {0}, key: {1}".format(self.bucket_name, to_string(key)))
return self.copy_object(self.bucket_name, key, key, headers=headers)
def delete_object(self, key, params=None, headers=None):
"""删除一个文件。
:param str key: 文件名
:param params: 请求参数
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
headers = http.CaseInsensitiveDict(headers)
logger.info("Start to delete object, bucket: {0}, key: {1}".format(self.bucket_name, to_string(key)))
resp = self.__do_object('DELETE', key, params=params, headers=headers)
logger.debug("Delete object done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def restore_object(self, key, params=None, headers=None, input=None):
"""restore an object
如果是第一次针对该object调用接口,返回RequestResult.status = 202;
如果已经成功调用过restore接口,且服务端仍处于解冻中,抛异常RestoreAlreadyInProgress(status=409)
如果已经成功调用过restore接口,且服务端解冻已经完成,再次调用时返回RequestResult.status = 200,且会将object的可下载时间延长一天,最多延长7天。
如果object不存在,则抛异常NoSuchKey(status=404);
对非Archive类型的Object提交restore,则抛异常OperationNotSupported(status=400)
也可以通过调用head_object接口来获取meta信息来判断是否可以restore与restore的状态
代码示例::
>>> meta = bucket.head_object(key)
>>> if meta.resp.headers['x-oss-storage-class'] == oss2.BUCKET_STORAGE_CLASS_ARCHIVE:
>>> bucket.restore_object(key)
>>> while True:
>>> meta = bucket.head_object(key)
>>> if meta.resp.headers['x-oss-restore'] == 'ongoing-request="true"':
>>> time.sleep(5)
>>> else:
>>> break
:param str key: object name
:param params: 请求参数
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:param input: 解冻配置。
:type input: class:`RestoreConfiguration <oss2.models.RestoreConfiguration>`
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
headers = http.CaseInsensitiveDict(headers)
logger.debug("Start to restore object, bucket: {0}, key: {1}".format(self.bucket_name, to_string(key)))
if params is None:
params = dict()
if Bucket.RESTORE not in params:
params[Bucket.RESTORE] = ''
data = self.__convert_data(RestoreConfiguration, xml_utils.to_put_restore_config, input)
resp = self.__do_object('POST', key, params=params, headers=headers, data=data)
logger.debug("Restore object done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def put_object_acl(self, key, permission, params=None, headers=None):
"""设置文件的ACL。
:param str key: 文件名
:param str permission: 可以是oss2.OBJECT_ACL_DEFAULT、oss2.OBJECT_ACL_PRIVATE、oss2.OBJECT_ACL_PUBLIC_READ或
oss2.OBJECT_ACL_PUBLIC_READ_WRITE。
:param dict params: 请求参数
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put object acl, bucket: {0}, key: {1}, acl: {2}".format(
self.bucket_name, to_string(key), permission))
headers = http.CaseInsensitiveDict(headers)
headers[OSS_OBJECT_ACL] = permission
if params is None:
params = dict()
if Bucket.ACL not in params:
params[Bucket.ACL] = ''
resp = self.__do_object('PUT', key, params=params, headers=headers)
logger.debug("Put object acl done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_object_acl(self, key, params=None, headers=None):
"""获取文件的ACL。
:param key: 文件名
:param params: 请求参数
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`GetObjectAclResult <oss2.models.GetObjectAclResult>`
"""
logger.debug("Start to get object acl, bucket: {0}, key: {1}".format(self.bucket_name, to_string(key)))
headers = http.CaseInsensitiveDict(headers)
if params is None:
params = dict()
if Bucket.ACL not in params:
params[Bucket.ACL] = ''
resp = self.__do_object('GET', key, params=params, headers=headers)
logger.debug("Get object acl done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_object_acl, GetObjectAclResult)
def batch_delete_objects(self, key_list, headers=None):
"""批量删除文件。待删除文件列表不能为空。
:param key_list: 文件名列表,不能为空。
:type key_list: list of str
:param headers: HTTP头部
:return: :class:`BatchDeleteObjectsResult <oss2.models.BatchDeleteObjectsResult>`
"""
if not key_list:
raise ClientError('key_list should not be empty')
logger.debug("Start to delete objects, bucket: {0}, keys: {1}".format(self.bucket_name, key_list))
data = xml_utils.to_batch_delete_objects_request(key_list, False)
headers = http.CaseInsensitiveDict(headers)
headers['Content-MD5'] = utils.content_md5(data)
resp = self.__do_bucket('POST',
data=data,
params={'delete': '', 'encoding-type': 'url'},
headers=headers)
logger.debug("Delete objects done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_batch_delete_objects, BatchDeleteObjectsResult)
def delete_object_versions(self, keylist_versions, headers=None):
"""批量删除带版本文件。待删除文件列表不能为空。
:param key_list_with_version: 带版本的文件名列表,不能为空。(如果传入,则不能为空)
:type key_list: list of BatchDeleteObjectsList
:param headers: HTTP头部
:return: :class:`BatchDeleteObjectsResult <oss2.models.BatchDeleteObjectsResult>`
"""
if not keylist_versions:
raise ClientError('keylist_versions should not be empty')
logger.debug("Start to delete object versions, bucket: {0}".format(self.bucket_name))
data = xml_utils.to_batch_delete_objects_version_request(keylist_versions, False)
headers = http.CaseInsensitiveDict(headers)
headers['Content-MD5'] = utils.content_md5(data)
resp = self.__do_bucket('POST',
data=data,
params={'delete': '', 'encoding-type': 'url'},
headers=headers)
logger.debug("Delete object versions done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_batch_delete_objects, BatchDeleteObjectsResult)
def init_multipart_upload(self, key, headers=None, params=None):
"""初始化分片上传。
返回值中的 `upload_id` 以及Bucket名和Object名三元组唯一对应了此次分片上传事件。
:param str key: 待上传的文件名
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`InitMultipartUploadResult <oss2.models.InitMultipartUploadResult>`
"""
headers = utils.set_content_type(http.CaseInsensitiveDict(headers), key)
if params is None:
tmp_params = dict()
else:
tmp_params = params.copy()
tmp_params['uploads'] = ''
logger.debug("Start to init multipart upload, bucket: {0}, keys: {1}, headers: {2}, params: {3}".format(
self.bucket_name, to_string(key), headers, tmp_params))
resp = self.__do_object('POST', key, params=tmp_params, headers=headers)
logger.debug("Init multipart upload done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_init_multipart_upload, InitMultipartUploadResult)
def upload_part(self, key, upload_id, part_number, data, progress_callback=None, headers=None):
"""上传一个分片。
:param str key: 待上传文件名,这个文件名要和 :func:`init_multipart_upload` 的文件名一致。
:param str upload_id: 分片上传ID
:param int part_number: 分片号,最小值是1.
:param data: 待上传数据。
:param progress_callback: 用户指定进度回调函数。可以用来实现进度条等功能。参考 :ref:`progress_callback` 。
:param headers: 用户指定的HTTP头部。可以指定Content-MD5头部等
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`PutObjectResult <oss2.models.PutObjectResult>`
"""
headers = http.CaseInsensitiveDict(headers)
if progress_callback:
data = utils.make_progress_adapter(data, progress_callback)
if self.enable_crc:
data = utils.make_crc_adapter(data)
logger.debug(
"Start to upload multipart, bucket: {0}, key: {1}, upload_id: {2}, part_number: {3}, headers: {4}".format(
self.bucket_name, to_string(key), upload_id, part_number, headers))
resp = self.__do_object('PUT', key,
params={'uploadId': upload_id, 'partNumber': str(part_number)},
headers=headers,
data=data)
logger.debug("Upload multipart done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
result = PutObjectResult(resp)
if self.enable_crc and result.crc is not None:
utils.check_crc('upload part', data.crc, result.crc, result.request_id)
return result
def complete_multipart_upload(self, key, upload_id, parts, headers=None):
"""完成分片上传,创建文件。
:param str key: 待上传的文件名,这个文件名要和 :func:`init_multipart_upload` 的文件名一致。
:param str upload_id: 分片上传ID
:param parts: PartInfo列表。PartInfo中的part_number和etag是必填项。其中的etag可以从 :func:`upload_part` 的返回值中得到。
:type parts: list of `PartInfo <oss2.models.PartInfo>`
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`PutObjectResult <oss2.models.PutObjectResult>`
"""
headers = http.CaseInsensitiveDict(headers)
data = None
if parts is not None:
parts = sorted(parts, key=lambda p: p.part_number)
data = xml_utils.to_complete_upload_request(parts)
logger.debug("Start to complete multipart upload, bucket: {0}, key: {1}, upload_id: {2}, parts: {3}".format(
self.bucket_name, to_string(key), upload_id, data))
resp = self.__do_object('POST', key,
params={'uploadId': upload_id},
data=data,
headers=headers)
logger.debug(
"Complete multipart upload done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
result = PutObjectResult(resp)
if self.enable_crc and parts is not None:
object_crc = utils.calc_obj_crc_from_parts(parts)
utils.check_crc('multipart upload', object_crc, result.crc, result.request_id)
return result
def abort_multipart_upload(self, key, upload_id, headers=None):
"""取消分片上传。
:param str key: 待上传的文件名,这个文件名要和 :func:`init_multipart_upload` 的文件名一致。
:param str upload_id: 分片上传ID
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to abort multipart upload, bucket: {0}, key: {1}, upload_id: {2}".format(
self.bucket_name, to_string(key), upload_id))
headers = http.CaseInsensitiveDict(headers)
resp = self.__do_object('DELETE', key,
params={'uploadId': upload_id}, headers=headers)
logger.debug("Abort multipart done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def list_multipart_uploads(self,
prefix='',
delimiter='',
key_marker='',
upload_id_marker='',
max_uploads=1000,
headers=None):
"""罗列正在进行中的分片上传。支持分页。
:param str prefix: 只罗列匹配该前缀的文件的分片上传
:param str delimiter: 目录分割符
:param str key_marker: 文件名分页符。第一次调用可以不传,后续设为返回值中的 `next_key_marker`
:param str upload_id_marker: 分片ID分页符。第一次调用可以不传,后续设为返回值中的 `next_upload_id_marker`
:param int max_uploads: 一次罗列最多能够返回的条目数
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`ListMultipartUploadsResult <oss2.models.ListMultipartUploadsResult>`
"""
logger.debug("Start to list multipart uploads, bucket: {0}, prefix: {1}, delimiter: {2}, key_marker: {3}, "
"upload_id_marker: {4}, max_uploads: {5}".format(self.bucket_name, to_string(prefix), delimiter,
to_string(key_marker), upload_id_marker,
max_uploads))
headers = http.CaseInsensitiveDict(headers)
resp = self.__do_bucket('GET',
params={'uploads': '',
'prefix': prefix,
'delimiter': delimiter,
'key-marker': key_marker,
'upload-id-marker': upload_id_marker,
'max-uploads': str(max_uploads),
'encoding-type': 'url'},
headers=headers)
logger.debug("List multipart uploads done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_multipart_uploads, ListMultipartUploadsResult)
def upload_part_copy(self, source_bucket_name, source_key, byte_range,
target_key, target_upload_id, target_part_number,
headers=None, params=None):
"""分片拷贝。把一个已有文件的一部分或整体拷贝成目标文件的一个分片。
:source_bucket_name: 源文件所在bucket的名称
:source_key:源文件名称
:param byte_range: 指定待拷贝内容在源文件里的范围。参见 :ref:`byte_range`
:target_key: 目的文件的名称
:target_upload_id: 目的文件的uploadid
:target_part_number: 目的文件的分片号
:param params: 请求参数
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`PutObjectResult <oss2.models.PutObjectResult>`
"""
headers = http.CaseInsensitiveDict(headers)
if params and Bucket.VERSIONID in params:
headers[OSS_COPY_OBJECT_SOURCE] = '/' + source_bucket_name + \
'/' + urlquote(source_key, '') + '?versionId=' + params[Bucket.VERSIONID]
else:
headers[OSS_COPY_OBJECT_SOURCE] = '/' + source_bucket_name + '/' + urlquote(source_key, '')
range_string = _make_range_string(byte_range)
if range_string:
headers[OSS_COPY_OBJECT_SOURCE_RANGE] = range_string
logger.debug("Start to upload part copy, source bucket: {0}, source key: {1}, bucket: {2}, key: {3}, range"
": {4}, upload id: {5}, part_number: {6}, headers: {7}".format(source_bucket_name,
to_string(source_key),self.bucket_name,to_string(target_key),
byte_range, target_upload_id,target_part_number, headers))
if params is None:
params = dict()
params['uploadId'] = target_upload_id
params['partNumber'] = str(target_part_number)
resp = self.__do_object('PUT', target_key,
params=params,headers=headers)
logger.debug("Upload part copy done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return PutObjectResult(resp)
def list_parts(self, key, upload_id,
marker='', max_parts=1000, headers=None):
"""列举已经上传的分片。支持分页。
:param headers: HTTP头部
:param str key: 文件名
:param str upload_id: 分片上传ID
:param str marker: 分页符
:param int max_parts: 一次最多罗列多少分片
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`ListPartsResult <oss2.models.ListPartsResult>`
"""
logger.debug("Start to list parts, bucket: {0}, key: {1}, upload_id: {2}, marker: {3}, max_parts: {4}".format(
self.bucket_name, to_string(key), upload_id, marker, max_parts))
headers = http.CaseInsensitiveDict(headers)
resp = self.__do_object('GET', key,
params={'uploadId': upload_id,
'part-number-marker': marker,
'max-parts': str(max_parts)},
headers=headers)
logger.debug("List parts done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_parts, ListPartsResult)
def put_symlink(self, target_key, symlink_key, headers=None):
"""创建Symlink。
:param str target_key: 目标文件,目标文件不能为符号连接
:param str symlink_key: 符号连接类文件,其实质是一个特殊的文件,数据指向目标文件
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
headers = http.CaseInsensitiveDict(headers)
headers[OSS_SYMLINK_TARGET] = urlquote(target_key, '')
logger.debug("Start to put symlink, bucket: {0}, target_key: {1}, symlink_key: {2}, headers: {3}".format(
self.bucket_name, to_string(target_key), to_string(symlink_key), headers))
resp = self.__do_object('PUT', symlink_key, headers=headers, params={Bucket.SYMLINK: ''})
logger.debug("Put symlink done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_symlink(self, symlink_key, params=None, headers=None):
"""获取符号连接文件的目标文件。
:param str symlink_key: 符号连接类文件
:param dict params: 请求参数
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`GetSymlinkResult <oss2.models.GetSymlinkResult>`
:raises: 如果文件的符号链接不存在,则抛出 :class:`NoSuchKey <oss2.exceptions.NoSuchKey>` ;还可能抛出其他异常
"""
logger.debug(
"Start to get symlink, bucket: {0}, symlink_key: {1}".format(self.bucket_name, to_string(symlink_key)))
headers = http.CaseInsensitiveDict(headers)
if params is None:
params = dict()
if Bucket.SYMLINK not in params:
params[Bucket.SYMLINK] = ''
resp = self.__do_object('GET', symlink_key, params=params, headers=headers)
logger.debug("Get symlink done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return GetSymlinkResult(resp)
def create_bucket(self, permission=None, input=None, headers=None):
"""创建新的Bucket。
:param str permission: 指定Bucket的ACL。可以是oss2.BUCKET_ACL_PRIVATE(推荐、缺省)、oss2.BUCKET_ACL_PUBLIC_READ或是
oss2.BUCKET_ACL_PUBLIC_READ_WRITE。
:param input: :class:`BucketCreateConfig <oss2.models.BucketCreateConfig>` object
"""
headers = http.CaseInsensitiveDict(headers)
if permission:
headers[OSS_CANNED_ACL] = permission
data = self.__convert_data(BucketCreateConfig, xml_utils.to_put_bucket_config, input)
logger.debug("Start to create bucket, bucket: {0}, permission: {1}, config: {2}".format(self.bucket_name,
permission, data))
resp = self.__do_bucket('PUT', headers=headers, data=data)
logger.debug("Create bucket done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def delete_bucket(self):
"""删除一个Bucket。只有没有任何文件,也没有任何未完成的分片上传的Bucket才能被删除。
:return: :class:`RequestResult <oss2.models.RequestResult>`
":raises: 如果试图删除一个非空Bucket,则抛出 :class:`BucketNotEmpty <oss2.exceptions.BucketNotEmpty>`
"""
logger.info("Start to delete bucket, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('DELETE')
logger.debug("Delete bucket done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def put_bucket_acl(self, permission):
"""设置Bucket的ACL。
:param str permission: 新的ACL,可以是oss2.BUCKET_ACL_PRIVATE、oss2.BUCKET_ACL_PUBLIC_READ或
oss2.BUCKET_ACL_PUBLIC_READ_WRITE
"""
logger.debug("Start to put bucket acl, bucket: {0}, acl: {1}".format(self.bucket_name, permission))
resp = self.__do_bucket('PUT', headers={OSS_CANNED_ACL: permission}, params={Bucket.ACL: ''})
logger.debug("Put bucket acl done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_acl(self):
"""获取Bucket的ACL。
:return: :class:`GetBucketAclResult <oss2.models.GetBucketAclResult>`
"""
logger.debug("Start to get bucket acl, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.ACL: ''})
logger.debug("Get bucket acl done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_acl, GetBucketAclResult)
def put_bucket_cors(self, input):
"""设置Bucket的CORS。
:param input: :class:`BucketCors <oss2.models.BucketCors>` 对象或其他
"""
data = self.__convert_data(BucketCors, xml_utils.to_put_bucket_cors, input)
logger.debug("Start to put bucket cors, bucket: {0}, cors: {1}".format(self.bucket_name, data))
resp = self.__do_bucket('PUT', data=data, params={Bucket.CORS: ''})
logger.debug("Put bucket cors done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_cors(self):
"""获取Bucket的CORS配置。
:return: :class:`GetBucketCorsResult <oss2.models.GetBucketCorsResult>`
"""
logger.debug("Start to get bucket CORS, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.CORS: ''})
logger.debug("Get bucket CORS done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_cors, GetBucketCorsResult)
def delete_bucket_cors(self):
"""删除Bucket的CORS配置。"""
logger.debug("Start to delete bucket CORS, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('DELETE', params={Bucket.CORS: ''})
logger.debug("Delete bucket CORS done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def put_bucket_lifecycle(self, input, headers=None):
"""设置生命周期管理的配置。
:param input: :class:`BucketLifecycle <oss2.models.BucketLifecycle>` 对象或其他
"""
headers = http.CaseInsensitiveDict(headers)
data = self.__convert_data(BucketLifecycle, xml_utils.to_put_bucket_lifecycle, input)
logger.debug("Start to put bucket lifecycle, bucket: {0}, lifecycle: {1}".format(self.bucket_name, data))
resp = self.__do_bucket('PUT', data=data, params={Bucket.LIFECYCLE: ''}, headers=headers)
logger.debug("Put bucket lifecycle done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_lifecycle(self):
"""获取生命周期管理配置。
:return: :class:`GetBucketLifecycleResult <oss2.models.GetBucketLifecycleResult>`
:raises: 如果没有设置Lifecycle,则抛出 :class:`NoSuchLifecycle <oss2.exceptions.NoSuchLifecycle>`
"""
logger.debug("Start to get bucket lifecycle, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.LIFECYCLE: ''})
logger.debug("Get bucket lifecycle done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_lifecycle, GetBucketLifecycleResult)
def delete_bucket_lifecycle(self):
"""删除生命周期管理配置。如果Lifecycle没有设置,也返回成功。"""
logger.debug("Start to delete bucket lifecycle, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('DELETE', params={Bucket.LIFECYCLE: ''})
logger.debug("Delete bucket lifecycle done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_location(self):
"""获取Bucket的数据中心。
:return: :class:`GetBucketLocationResult <oss2.models.GetBucketLocationResult>`
"""
logger.debug("Start to get bucket location, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.LOCATION: ''})
logger.debug("Get bucket location done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_location, GetBucketLocationResult)
def put_bucket_logging(self, input):
"""设置Bucket的访问日志功能。
:param input: :class:`BucketLogging <oss2.models.BucketLogging>` 对象或其他
"""
data = self.__convert_data(BucketLogging, xml_utils.to_put_bucket_logging, input)
logger.debug("Start to put bucket logging, bucket: {0}, logging: {1}".format(self.bucket_name, data))
resp = self.__do_bucket('PUT', data=data, params={Bucket.LOGGING: ''})
logger.debug("Put bucket logging done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_logging(self):
"""获取Bucket的访问日志功能配置。
:return: :class:`GetBucketLoggingResult <oss2.models.GetBucketLoggingResult>`
"""
logger.debug("Start to get bucket logging, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.LOGGING: ''})
logger.debug("Get bucket logging done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_logging, GetBucketLoggingResult)
def delete_bucket_logging(self):
"""关闭Bucket的访问日志功能。"""
logger.debug("Start to delete bucket loggging, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('DELETE', params={Bucket.LOGGING: ''})
logger.debug("Put bucket lifecycle done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def put_bucket_referer(self, input):
"""为Bucket设置防盗链。
:param input: :class:`BucketReferer <oss2.models.BucketReferer>` 对象或其他
"""
data = self.__convert_data(BucketReferer, xml_utils.to_put_bucket_referer, input)
logger.debug("Start to put bucket referer, bucket: {0}, referer: {1}".format(self.bucket_name, to_string(data)))
resp = self.__do_bucket('PUT', data=data, params={Bucket.REFERER: ''})
logger.debug("Put bucket referer done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_referer(self):
"""获取Bucket的防盗链配置。
:return: :class:`GetBucketRefererResult <oss2.models.GetBucketRefererResult>`
"""
logger.debug("Start to get bucket referer, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.REFERER: ''})
logger.debug("Get bucket referer done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_referer, GetBucketRefererResult)
def get_bucket_stat(self):
"""查看Bucket的状态,目前包括bucket大小,bucket的object数量,bucket正在上传的Multipart Upload事件个数等。
:return: :class:`GetBucketStatResult <oss2.models.GetBucketStatResult>`
"""
logger.debug("Start to get bucket stat, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.STAT: ''})
logger.debug("Get bucket stat done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_stat, GetBucketStatResult)
def get_bucket_info(self):
"""获取bucket相关信息,如创建时间,访问Endpoint,Owner与ACL等。
:return: :class:`GetBucketInfoResult <oss2.models.GetBucketInfoResult>`
"""
logger.debug("Start to get bucket info, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.BUCKET_INFO: ''})
logger.debug("Get bucket info done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_info, GetBucketInfoResult)
def put_bucket_website(self, input):
"""为Bucket配置静态网站托管功能。
:param input: :class:`BucketWebsite <oss2.models.BucketWebsite>`
"""
data = self.__convert_data(BucketWebsite, xml_utils.to_put_bucket_website, input)
headers = http.CaseInsensitiveDict()
headers['Content-MD5'] = utils.content_md5(data)
logger.debug("Start to put bucket website, bucket: {0}, website: {1}".format(self.bucket_name, to_string(data)))
resp = self.__do_bucket('PUT', data=data, params={Bucket.WEBSITE: ''}, headers=headers)
logger.debug("Put bucket website done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_website(self):
"""获取Bucket的静态网站托管配置。
:return: :class:`GetBucketWebsiteResult <oss2.models.GetBucketWebsiteResult>`
:raises: 如果没有设置静态网站托管,那么就抛出 :class:`NoSuchWebsite <oss2.exceptions.NoSuchWebsite>`
"""
logger.debug("Start to get bucket website, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.WEBSITE: ''})
logger.debug("Get bucket website done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_website, GetBucketWebsiteResult)
def delete_bucket_website(self):
"""关闭Bucket的静态网站托管功能。"""
logger.debug("Start to delete bucket website, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('DELETE', params={Bucket.WEBSITE: ''})
logger.debug("Delete bucket website done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def create_live_channel(self, channel_name, input):
"""创建推流直播频道
:param str channel_name: 要创建的live channel的名称
:param input: LiveChannelInfo类型,包含了live channel中的描述信息
:return: :class:`CreateLiveChannelResult <oss2.models.CreateLiveChannelResult>`
"""
data = self.__convert_data(LiveChannelInfo, xml_utils.to_create_live_channel, input)
logger.debug("Start to create live-channel, bucket: {0}, channel_name: {1}, info: {2}".format(
self.bucket_name, to_string(channel_name), to_string(data)))
resp = self.__do_object('PUT', channel_name, data=data, params={Bucket.LIVE: ''})
logger.debug("Create live-channel done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_create_live_channel, CreateLiveChannelResult)
def delete_live_channel(self, channel_name):
"""删除推流直播频道
:param str channel_name: 要删除的live channel的名称
"""
logger.debug("Start to delete live-channel, bucket: {0}, live_channel: {1}".format(
self.bucket_name, to_string(channel_name)))
resp = self.__do_object('DELETE', channel_name, params={Bucket.LIVE: ''})
logger.debug("Delete live-channel done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_live_channel(self, channel_name):
"""获取直播频道配置
:param str channel_name: 要获取的live channel的名称
:return: :class:`GetLiveChannelResult <oss2.models.GetLiveChannelResult>`
"""
logger.debug("Start to get live-channel info: bucket: {0}, live_channel: {1}".format(
self.bucket_name, to_string(channel_name)))
resp = self.__do_object('GET', channel_name, params={Bucket.LIVE: ''})
logger.debug("Get live-channel done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_live_channel, GetLiveChannelResult)
def list_live_channel(self, prefix='', marker='', max_keys=100):
"""列举出Bucket下所有符合条件的live channel
param: str prefix: list时channel_id的公共前缀
param: str marker: list时指定的起始标记
param: int max_keys: 本次list返回live channel的最大个数
return: :class:`ListLiveChannelResult <oss2.models.ListLiveChannelResult>`
"""
logger.debug("Start to list live-channels, bucket: {0}, prefix: {1}, marker: {2}, max_keys: {3}".format(
self.bucket_name, to_string(prefix), to_string(marker), max_keys))
resp = self.__do_bucket('GET', params={Bucket.LIVE: '',
'prefix': prefix,
'marker': marker,
'max-keys': str(max_keys)})
logger.debug("List live-channel done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_live_channel, ListLiveChannelResult)
def get_live_channel_stat(self, channel_name):
"""获取live channel当前推流的状态
param str channel_name: 要获取推流状态的live channel的名称
return: :class:`GetLiveChannelStatResult <oss2.models.GetLiveChannelStatResult>`
"""
logger.debug("Start to get live-channel stat, bucket: {0}, channel_name: {1}".format(
self.bucket_name, to_string(channel_name)))
resp = self.__do_object('GET', channel_name, params={Bucket.LIVE: '', Bucket.COMP: 'stat'})
logger.debug("Get live-channel stat done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_live_channel_stat, GetLiveChannelStatResult)
def put_live_channel_status(self, channel_name, status):
"""更改live channel的status,仅能在“enabled”和“disabled”两种状态中更改
param str channel_name: 要更改status的live channel的名称
param str status: live channel的目标status
"""
logger.debug("Start to put live-channel status, bucket: {0}, channel_name: {1}, status: {2}".format(
self.bucket_name, to_string(channel_name), status))
resp = self.__do_object('PUT', channel_name, params={Bucket.LIVE: '', Bucket.STATUS: status})
logger.debug("Put live-channel status done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_live_channel_history(self, channel_name):
"""获取live channel中最近的最多十次的推流记录,记录中包含推流的起止时间和远端的地址
param str channel_name: 要获取最近推流记录的live channel的名称
return: :class:`GetLiveChannelHistoryResult <oss2.models.GetLiveChannelHistoryResult>`
"""
logger.debug("Start to get live-channel history, bucket: {0}, channel_name: {1}".format(
self.bucket_name, to_string(channel_name)))
resp = self.__do_object('GET', channel_name, params={Bucket.LIVE: '', Bucket.COMP: 'history'})
logger.debug(
"Get live-channel history done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_live_channel_history, GetLiveChannelHistoryResult)
def post_vod_playlist(self, channel_name, playlist_name, start_time=0, end_time=0):
"""根据指定的playlist name以及startTime和endTime生成一个点播的播放列表
param str channel_name: 要生成点播列表的live channel的名称
param str playlist_name: 要生成点播列表m3u8文件的名称
param int start_time: 点播的起始时间,Unix Time格式,可以使用int(time.time())获取
param int end_time: 点播的结束时间,Unix Time格式,可以使用int(time.time())获取
"""
logger.debug("Start to post vod playlist, bucket: {0}, channel_name: {1}, playlist_name: {2}, start_time: "
"{3}, end_time: {4}".format(self.bucket_name, to_string(channel_name), playlist_name, start_time,
end_time))
key = channel_name + "/" + playlist_name
resp = self.__do_object('POST', key, params={Bucket.VOD: '', 'startTime': str(start_time),
'endTime': str(end_time)})
logger.debug("Post vod playlist done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_vod_playlist(self, channel_name, start_time, end_time):
"""查看指定时间段内的播放列表
param str channel_name: 要获取点播列表的live channel的名称
param int start_time: 点播的起始时间,Unix Time格式,可以使用int(time.time())获取
param int end_time: 点播的结束时间,Unix Time格式,可以使用int(time.time())获取
"""
logger.debug("Start to get vod playlist, bucket: {0}, channel_name: {1}, start_time: "
"{2}, end_time: {3}".format(self.bucket_name, to_string(channel_name), start_time, end_time))
resp = self.__do_object('GET', channel_name, params={Bucket.VOD: '', 'startTime': str(start_time),
'endTime': str(end_time)})
logger.debug("get vod playlist done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
result = GetVodPlaylistResult(resp)
return result
def process_object(self, key, process, headers=None):
"""处理图片的接口,支持包括调整大小,旋转,裁剪,水印,格式转换等,支持多种方式组合处理。
:param str key: 处理的图片的对象名称
:param str process: 处理的字符串,例如"image/resize,w_100|sys/saveas,o_dGVzdC5qcGc,b_dGVzdA"
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
"""
headers = http.CaseInsensitiveDict(headers)
logger.debug("Start to process object, bucket: {0}, key: {1}, process: {2}".format(
self.bucket_name, to_string(key), process))
process_data = "%s=%s" % (Bucket.PROCESS, process)
resp = self.__do_object('POST', key, params={Bucket.PROCESS: ''}, headers=headers, data=process_data)
logger.debug("Process object done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return ProcessObjectResult(resp)
def put_object_tagging(self, key, tagging, headers=None, params=None):
"""
:param str key: 上传tagging的对象名称,不能为空。
:param tagging: tag 标签内容
:type tagging: :class:`Tagging <oss2.models.Tagging>` 对象
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:param dict params: HTTP请求参数
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put object tagging, bucket: {0}, key: {1}, tagging: {2}".format(
self.bucket_name, to_string(key), tagging))
if headers is not None:
headers = http.CaseInsensitiveDict(headers)
if params is None:
params = dict()
params[Bucket.TAGGING] = ""
data = self.__convert_data(Tagging, xml_utils.to_put_tagging, tagging)
resp = self.__do_object('PUT', key, data=data, params=params, headers=headers)
return RequestResult(resp)
def get_object_tagging(self, key, params=None, headers=None):
"""
:param str key: 要获取tagging的对象名称
:param dict params: 请求参数
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`GetTaggingResult <oss2.models.GetTaggingResult>`
"""
logger.debug("Start to get object tagging, bucket: {0}, key: {1} params: {2}".format(
self.bucket_name, to_string(key), str(params)))
headers = http.CaseInsensitiveDict(headers)
if params is None:
params = dict()
params[Bucket.TAGGING] = ""
resp = self.__do_object('GET', key, params=params, headers=headers)
return self._parse_result(resp, xml_utils.parse_get_tagging, GetTaggingResult)
def delete_object_tagging(self, key, params=None, headers=None):
"""
:param str key: 要删除tagging的对象名称
:param dict params: 请求参数
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete object tagging, bucket: {0}, key: {1}".format(
self.bucket_name, to_string(key)))
headers = http.CaseInsensitiveDict(headers)
if params is None:
params = dict()
params[Bucket.TAGGING] = ""
resp = self.__do_object('DELETE', key, params=params, headers=headers)
logger.debug("Delete object tagging done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def put_bucket_encryption(self, rule):
"""设置bucket加密配置。
:param rule: :class:` ServerSideEncryptionRule<oss2.models.ServerSideEncryptionRule>` 对象
"""
data = self.__convert_data(ServerSideEncryptionRule, xml_utils.to_put_bucket_encryption, rule)
logger.debug("Start to put bucket encryption, bucket: {0}, rule: {1}".format(self.bucket_name, data))
resp = self.__do_bucket('PUT', data=data, params={Bucket.ENCRYPTION: ""})
logger.debug("Put bucket encryption done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_encryption(self):
"""获取bucket加密配置。
:return: :class:`GetServerSideEncryptionResult <oss2.models.GetServerSideEncryptionResult>`
:raises: 如果没有设置Bucket encryption,则抛出 :class:`NoSuchServerSideEncryptionRule <oss2.exceptions.NoSuchServerSideEncryptionRule>`
"""
logger.debug("Start to get bucket encryption, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.ENCRYPTION: ''})
logger.debug("Get bucket encryption done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_encryption, GetServerSideEncryptionResult)
def delete_bucket_encryption(self):
"""删除Bucket加密配置。如果Bucket加密没有设置,也返回成功。"""
logger.debug("Start to delete bucket encryption, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('DELETE', params={Bucket.ENCRYPTION: ''})
logger.debug("Delete bucket encryption done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def put_bucket_tagging(self, tagging, headers=None):
"""
:param str key: 上传tagging的对象名称,不能为空。
:param tagging: tag 标签内容
:type tagging: :class:`Tagging <oss2.models.Tagging>` 对象
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put object tagging, bucket: {0} tagging: {1}".format(
self.bucket_name, tagging))
headers = http.CaseInsensitiveDict(headers)
data = self.__convert_data(Tagging, xml_utils.to_put_tagging, tagging)
resp = self.__do_bucket('PUT', data=data, params={Bucket.TAGGING: ''}, headers=headers)
logger.debug("Put bucket tagging done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_tagging(self):
"""
:param str key: 要获取tagging的对象名称
:param dict params: 请求参数
:return: :class:`GetTaggingResult<oss2.models.GetTaggingResult>`
"""
logger.debug("Start to get bucket tagging, bucket: {0}".format(
self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.TAGGING: ''})
logger.debug("Get bucket tagging done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_tagging, GetTaggingResult)
def delete_bucket_tagging(self, params=None):
"""
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete bucket tagging, bucket: {0}".format(
self.bucket_name))
if params is None:
params = dict()
if Bucket.TAGGING not in params:
params[Bucket.TAGGING] = ''
resp = self.__do_bucket('DELETE', params=params)
logger.debug("Delete bucket tagging done, req_id: {0}, status_code: {1}".format(
resp.request_id, resp.status))
return RequestResult(resp)
def list_object_versions(self, prefix='', delimiter='', key_marker='',
max_keys=100, versionid_marker='', headers=None):
"""根据前缀罗列Bucket里的文件的版本。
:param str prefix: 只罗列文件名为该前缀的文件
:param str delimiter: 分隔符。可以用来模拟目录
:param str key_marker: 分页标志。首次调用传空串,后续使用返回值的next_marker
:param int max_keys: 最多返回文件的个数,文件和目录的和不能超过该值
:param str versionid_marker: 设定结果从key-marker对象的
versionid-marker之后按新旧版本排序开始返回,该版本不会在返回的结果当中。
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
:return: :class:`ListObjectVersionsResult <oss2.models.ListObjectVersionsResult>`
"""
logger.debug(
"Start to List object versions, bucket: {0}, prefix: {1}, delimiter: {2},"
+"key_marker: {3}, versionid_marker: {4}, max-keys: {5}".format(
self.bucket_name, to_string(prefix), delimiter, to_string(key_marker),
to_string(versionid_marker), max_keys))
headers = http.CaseInsensitiveDict(headers)
resp = self.__do_bucket('GET',
params={'prefix': prefix,
'delimiter': delimiter,
'key-marker': key_marker,
'version-id-marker': versionid_marker,
'max-keys': str(max_keys),
'encoding-type': 'url',
Bucket.VERSIONS: ''},
headers=headers)
logger.debug("List object versions done, req_id: {0}, status_code: {1}"
.format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_object_versions, ListObjectVersionsResult)
def put_bucket_versioning(self, config, headers=None):
"""
:param str operation: 设置bucket是否开启多版本特性,可取值为:[Enabled,Suspend]
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put object versioning, bucket: {0}".format(self.bucket_name))
data = self.__convert_data(BucketVersioningConfig, xml_utils.to_put_bucket_versioning, config)
headers = http.CaseInsensitiveDict(headers)
headers['Content-MD5'] = utils.content_md5(data)
resp = self.__do_bucket('PUT', data=data, params={Bucket.VERSIONING: ''}, headers=headers)
logger.debug("Put bucket versiong done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_versioning(self):
"""
:return: :class:`GetBucketVersioningResult<oss2.models.GetBucketVersioningResult>`
"""
logger.debug("Start to get bucket versioning, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.VERSIONING: ''})
logger.debug("Get bucket versiong done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_versioning, GetBucketVersioningResult)
def put_bucket_policy(self, policy):
"""设置bucket授权策略, 具体policy书写规则请参考官方文档
:param str policy: 授权策略
"""
logger.debug("Start to put bucket policy, bucket: {0}, policy: {1}".format(self.bucket_name, policy))
resp = self.__do_bucket('PUT', data=policy, params={Bucket.POLICY: ''}, headers={'Content-MD5': utils.content_md5(policy)})
logger.debug("Put bucket policy done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_policy(self):
"""获取bucket授权策略
:return: :class:`GetBucketPolicyResult <oss2.models.GetBucketPolicyResult>`
"""
logger.debug("Start to get bucket policy, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.POLICY:''})
logger.debug("Get bucket policy done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return GetBucketPolicyResult(resp)
def delete_bucket_policy(self):
"""删除bucket授权策略
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete bucket policy, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('DELETE', params={Bucket.POLICY: ''})
logger.debug("Delete bucket policy done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def put_bucket_request_payment(self, payer):
"""设置付费者。
:param input: :class: str
"""
data = xml_utils.to_put_bucket_request_payment(payer)
logger.debug("Start to put bucket request payment, bucket: {0}, payer: {1}".format(self.bucket_name, payer))
resp = self.__do_bucket('PUT', data=data, params={Bucket.REQUESTPAYMENT: ''}, headers={'Content-MD5': utils.content_md5(data)})
logger.debug("Put bucket request payment done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_request_payment(self):
"""获取付费者设置。
:return: :class:`GetBucketRequestPaymentResult <oss2.models.GetBucketRequestPaymentResult>`
"""
logger.debug("Start to get bucket request payment, bucket: {0}.".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.REQUESTPAYMENT: ''})
logger.debug("Get bucket request payment done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_request_payment, GetBucketRequestPaymentResult)
def put_bucket_qos_info(self, bucket_qos_info):
"""配置bucket的QoSInfo
:param bucket_qos_info :class:`BucketQosInfo <oss2.models.BucketQosInfo>`
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put bucket qos info, bucket: {0}".format(self.bucket_name))
data = self.__convert_data(BucketQosInfo, xml_utils.to_put_qos_info, bucket_qos_info)
headers = http.CaseInsensitiveDict()
headers['Content-MD5'] = utils.content_md5(data)
resp = self.__do_bucket('PUT', data=data, params={Bucket.QOS_INFO: ''}, headers=headers)
logger.debug("Get bucket qos info done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_qos_info(self):
"""获取bucket的QoSInfo
:return: :class:`GetBucketQosInfoResult <oss2.models.GetBucketQosInfoResult>`
"""
logger.debug("Start to get bucket qos info, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.QOS_INFO: ''})
logger.debug("Get bucket qos info, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_qos_info, GetBucketQosInfoResult)
def delete_bucket_qos_info(self):
"""删除bucket的QoSInfo
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete bucket qos info, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('DELETE', params={Bucket.QOS_INFO: ''})
logger.debug("Delete bucket qos info done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def set_bucket_storage_capacity(self, user_qos):
"""设置Bucket的容量,单位GB
:param user_qos :class:`BucketUserQos <oss2.models.BucketUserQos>`
"""
logger.debug("Start to set bucket storage capacity: {0}".format(self.bucket_name))
data = xml_utils.to_put_bucket_user_qos(user_qos)
resp = self.__do_bucket('PUT', data=data, params={Bucket.USER_QOS: ''})
logger.debug("Set bucket storage capacity done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_storage_capacity(self):
"""获取bucket的容量信息。
:return: :class:`GetBucketUserQosResult <oss2.models.GetBucketUserQosResult>`
"""
logger.debug("Start to get bucket storage capacity, bucket:{0}".format(self.bucket_name))
resp = self._Bucket__do_bucket('GET', params={Bucket.USER_QOS: ''})
logger.debug("Get bucket storage capacity done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_user_qos, GetBucketUserQosResult)
def put_async_fetch_task(self, task_config):
"""创建一个异步获取文件到bucket的任务。
:param task_config: 任务配置
:type task_config: class:`AsyncFetchTaskConfiguration <oss2.models.AsyncFetchTaskConfiguration>`
:return: :class:`PutAsyncFetchTaskResult <oss2.models.PutAsyncFetchTaskResult>`
"""
logger.debug("Start to put async fetch task, bucket:{0}".format(self.bucket_name))
data = xml_utils.to_put_async_fetch_task(task_config)
headers = http.CaseInsensitiveDict()
headers['Content-MD5'] = utils.content_md5(data)
resp = self._Bucket__do_bucket('POST', data=data, params={Bucket.ASYNC_FETCH: ''}, headers=headers)
logger.debug("Put async fetch task done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_put_async_fetch_task_result, PutAsyncFetchTaskResult)
def get_async_fetch_task(self, task_id):
"""获取一个异步获取文件到bucket的任务信息。
:param str task_id: 任务id
:return: :class:`GetAsyncFetchTaskResult <oss2.models.GetAsyncFetchTaskResult>`
"""
logger.debug("Start to get async fetch task, bucket:{0}, task_id:{1}".format(self.bucket_name, task_id))
resp = self._Bucket__do_bucket('GET', headers={OSS_TASK_ID: task_id}, params={Bucket.ASYNC_FETCH: ''})
logger.debug("Put async fetch task done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_async_fetch_task_result, GetAsyncFetchTaskResult)
def put_bucket_inventory_configuration(self, inventory_configuration):
"""设置bucket清单配置
:param inventory_configuration :class:`InventoryConfiguration <oss2.models.InventoryConfiguration>`
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put bucket inventory configuration, bucket: {0}".format(self.bucket_name))
data = self.__convert_data(InventoryConfiguration, xml_utils.to_put_inventory_configuration, inventory_configuration)
headers = http.CaseInsensitiveDict()
headers['Content-MD5'] = utils.content_md5(data)
resp = self.__do_bucket('PUT', data=data, params={Bucket.INVENTORY: '', Bucket.INVENTORY_CONFIG_ID:inventory_configuration.inventory_id}, headers=headers)
logger.debug("Put bucket inventory configuration done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_inventory_configuration(self, inventory_id):
"""获取指定的清单配置。
:param str inventory_id : 清单配置id
:return: :class:`GetInventoryConfigurationResult <oss2.models.GetInventoryConfigurationResult>`
"""
logger.debug("Start to get bucket inventory configuration, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.INVENTORY: '', Bucket.INVENTORY_CONFIG_ID:inventory_id})
logger.debug("Get bucket inventory cinfguration done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_inventory_configuration, GetInventoryConfigurationResult)
def list_bucket_inventory_configurations(self, continuation_token=None):
"""罗列清单配置,默认单次最大返回100条配置,如果存在超过100条配置,罗列结果将会分页,
分页信息保存在 class:`ListInventoryConfigurationResult <oss2.models.ListInventoryConfigurationResult>`中。
:param str continuation_token: 分页标识, 默认值为None, 如果上次罗列不完整,这里设置为上次罗列结果中的next_continuation_token值。
:return: :class:`ListInventoryConfigurationResult <oss2.models.ListInventoryConfigurationResult>`
"""
logger.debug("Start to list bucket inventory configuration, bucket: {0}".format(self.bucket_name))
params = {Bucket.INVENTORY:''}
if continuation_token is not None:
params[Bucket.CONTINUATION_TOKEN] = continuation_token
resp = self.__do_bucket('GET', params=params)
logger.debug("List bucket inventory configuration done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_bucket_inventory_configurations, ListInventoryConfigurationsResult)
def delete_bucket_inventory_configuration(self, inventory_id):
"""删除指定的清单配置
:param str inventory_id : 清单配置id
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete bucket inventory configuration, bucket: {0}, configuration id: {1}.".format(self.bucket_name, inventory_id))
resp = self.__do_bucket('DELETE', params={Bucket.INVENTORY:'', Bucket.INVENTORY_CONFIG_ID:inventory_id})
logger.debug("Delete bucket inventory configuration, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def init_bucket_worm(self, retention_period_days=None):
"""创建一条合规保留策略
:param int retention_period_days : 指定object的保留天数
:return: :class:`InitBucketWormResult <oss2.models.InitBucketWormResult>`
"""
logger.debug("Start to init bucket worm, bucket: {0}, retention_period_days: {1}."
.format(self.bucket_name, retention_period_days))
data = xml_utils.to_put_init_bucket_worm(retention_period_days)
headers = http.CaseInsensitiveDict()
headers['Content-MD5'] = utils.content_md5(data)
resp = self.__do_bucket('POST', data=data, params={Bucket.WORM: ''}, headers=headers)
logger.debug("init bucket worm done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
result = InitBucketWormResult(resp)
result.worm_id = resp.headers.get("x-oss-worm-id")
return result
def abort_bucket_worm(self):
"""删除一条合规保留策略
只有未锁定保留策略的状态下才能删除,一旦锁定bucket数据将处于保护状态。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to abort bucket worm, bucket: {0}.".format(self.bucket_name))
resp = self.__do_bucket('DELETE', params={Bucket.WORM: ''})
logger.debug("abort bucket worm done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def complete_bucket_worm(self, worm_id=None):
"""锁定一条合规保留策略
:param str worm_id : 合规保留策略的id。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to complete bucket worm, bucket: {0}, worm_id: {1}.".format(self.bucket_name, worm_id))
resp = self.__do_bucket('POST', params={Bucket.WORM_ID: worm_id})
logger.debug("complete bucket worm done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def extend_bucket_worm(self, worm_id=None, retention_period_days=None):
"""延长已经锁定的合规保留策略的object保护天数。
:param str worm_id : 合规保留策略的id。
:param int retention_period_days : 指定object的保留天数
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
data = xml_utils.to_put_extend_bucket_worm(retention_period_days)
headers = http.CaseInsensitiveDict()
headers['Content-MD5'] = utils.content_md5(data)
logger.debug("Start to extend bucket worm, bucket: {0}, worm_id: {1}, retention_period_days."
.format(self.bucket_name, worm_id, retention_period_days))
resp = self.__do_bucket('POST', data=data, params={Bucket.WORM_ID: worm_id, Bucket.WORM_EXTEND: ''}, headers=headers)
logger.debug("extend bucket worm done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_worm(self):
"""获取合规保留策略
:return: :class:`GetBucketWormResult <oss2.models.GetBucketWormResult>`
"""
logger.debug("Start to get bucket worm, bucket: {0}.".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.WORM: ''})
logger.debug("get bucket worm done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_worm_result, GetBucketWormResult)
def put_bucket_replication(self, rule):
"""设置bucket跨区域复制规则
:param rule :class:`ReplicationRule <oss2.models.ReplicationRule>`
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put bucket replication: {0}".format(self.bucket_name))
data = xml_utils.to_put_bucket_replication(rule)
headers = http.CaseInsensitiveDict()
headers['Content-MD5'] = utils.content_md5(data)
resp = self.__do_bucket('POST', data=data, params={Bucket.REPLICATION: '', 'comp': 'add'}, headers=headers)
logger.debug("Put bucket replication done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_replication(self):
"""获取bucket跨区域复制规则
:return: :class:`GetBucketReplicationResult <oss2.models.GetBucketReplicationResult>`
"""
logger.debug("Start to get bucket replication: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.REPLICATION: ''})
logger.debug("Get bucket replication done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_replication_result, GetBucketReplicationResult)
def delete_bucket_replication(self, rule_id):
"""停止Bucket的跨区域复制并删除Bucket的复制配置
:param str rule_id: Bucket跨区域复制规则的id。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete bucket replication: {0}".format(self.bucket_name))
data = xml_utils.to_delete_bucket_replication(rule_id)
headers = http.CaseInsensitiveDict()
headers['Content-MD5'] = utils.content_md5(data)
resp = self.__do_bucket('POST', data=data, params={Bucket.REPLICATION: '', 'comp': 'delete'}, headers=headers)
logger.debug("Delete bucket replication done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_replication_location(self):
"""获取可复制到的Bucket所在的地域
:return: :class:`ReplicationLocation <oss2.models.GetBucketReplicationLocationResult>`
"""
logger.debug("Start to get bucket replication location: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.REPLICATION_LOCATION: ''})
logger.debug("Get bucket replication location done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_replication_location_result, GetBucketReplicationLocationResult)
def get_bucket_replication_progress(self, rule_id):
"""获取获取某个Bucket的跨区域复制进度
:param str rule_id: Bucket跨区域复制规则的id。
:return: :class:`GetBucketReplicationProgressResult <oss2.models.GetBucketReplicationProgressResult>`
"""
logger.debug("Start to get bucket replication progress: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.REPLICATION_PROGRESS: '', 'rule-id': rule_id})
logger.debug("Get bucket replication progress done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_replication_progress_result, GetBucketReplicationProgressResult)
def _get_bucket_config(self, config):
"""获得Bucket某项配置,具体哪种配置由 `config` 指定。该接口直接返回 `RequestResult` 对象。
通过read()接口可以获得XML字符串。不建议使用。
:param str config: 可以是 `Bucket.ACL` 、 `Bucket.LOGGING` 等。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to get bucket config, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={config: ''})
logger.debug("Get bucket config done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return resp
def put_bucket_transfer_acceleration(self, enabled):
"""为存储空间(Bucket)配置传输加速
:param str enabled : 是否开启传输加速。true:开启传输加速; false:关闭传输加速.
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to bucket transfer acceleration, bucket: {0}, enabled: {1}."
.format(self.bucket_name, enabled))
data = xml_utils.to_put_bucket_transfer_acceleration(enabled)
headers = http.CaseInsensitiveDict()
headers['Content-MD5'] = utils.content_md5(data)
resp = self.__do_bucket('PUT', data=data, params={Bucket.TRANSFER_ACCELERATION: ''}, headers=headers)
logger.debug("bucket transfer acceleration done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_transfer_acceleration(self):
"""获取目标存储空间(Bucket)的传输加速配置
:return: :class:`GetBucketTransferAccelerationResult <oss2.models.GetBucketTransferAccelerationResult>`
"""
logger.debug("Start to get bucket transfer acceleration: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.TRANSFER_ACCELERATION: ''})
logger.debug("Get bucket transfer acceleration done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_transfer_acceleration_result, GetBucketTransferAccelerationResult)
def create_bucket_cname_token(self, domain):
"""创建域名所有权验证所需的CnameToken。
:param str domain : 绑定的Cname名称。
:return: :class:`CreateBucketCnameTokenResult <oss2.models.CreateBucketCnameTokenResult>`
"""
logger.debug("Start to create bucket cname token, bucket: {0}.".format(self.bucket_name))
data = xml_utils.to_bucket_cname_configuration(domain)
resp = self.__do_bucket('POST', data=data, params={Bucket.CNAME: '', Bucket.COMP: 'token'})
logger.debug("bucket cname token done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_create_bucket_cname_token, CreateBucketCnameTokenResult)
def get_bucket_cname_token(self, domain):
"""获取已创建的CnameToken。
:param str domain : 绑定的Cname名称。
:return: :class:`GetBucketCnameTokenResult <oss2.models.GetBucketCnameTokenResult>`
"""
logger.debug("Start to get bucket cname: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.CNAME: domain, Bucket.COMP: 'token'})
logger.debug("Get bucket cname done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_cname_token, GetBucketCnameTokenResult)
def put_bucket_cname(self, input):
"""为某个存储空间(Bucket)绑定自定义域名。
:param input: PutBucketCnameRequest类型,包含了证书和自定义域名信息
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to add bucket cname, bucket: {0}.".format(self.bucket_name))
data = xml_utils.to_bucket_cname_configuration(input.domain, input.cert)
resp = self.__do_bucket('POST', data=data, params={Bucket.CNAME: '', Bucket.COMP: 'add'})
logger.debug("bucket cname done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def list_bucket_cname(self):
"""查询某个存储空间(Bucket)下绑定的所有Cname列表。
:return: :class:`ListBucketCnameResult <oss2.models.ListBucketCnameResult>`
"""
logger.debug("Start to do query list bucket cname: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.CNAME: ''})
logger.debug("query list bucket cname done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_bucket_cname, ListBucketCnameResult)
def delete_bucket_cname(self, domain):
"""删除某个存储空间(Bucket)已绑定的Cname
:param str domain : 绑定的Cname名称。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete bucket cname: {0}".format(self.bucket_name))
data = xml_utils.to_bucket_cname_configuration(domain)
resp = self.__do_bucket('POST', data=data, params={Bucket.CNAME: '', Bucket.COMP: 'delete'})
logger.debug("delete bucket cname done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def open_bucket_meta_query(self):
"""为存储空间(Bucket)开启元数据管理功能
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to bucket meta query, bucket: {0}.".format(self.bucket_name))
resp = self.__do_bucket('POST', params={Bucket.META_QUERY: '', 'comp': 'add'})
logger.debug("bucket meta query done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_meta_query_status(self):
"""获取指定存储空间(Bucket)的元数据索引库信息。
:return: :class:`GetBucketMetaQueryResult <oss2.models.GetBucketMetaQueryResult>`
"""
logger.debug("Start to get bucket meta query: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.META_QUERY: ''})
logger.debug("Get bucket meta query done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_meta_query_result, GetBucketMetaQueryResult)
def do_bucket_meta_query(self, do_meta_query_request):
"""查询满足指定条件的文件(Object),并按照指定字段和排序方式列出文件信息。
:param do_meta_query_request :class:`MetaQuery <oss2.models.MetaQuery>`
:return: :class:`DoBucketMetaQueryResult <oss2.models.DoBucketMetaQueryResult>`
"""
logger.debug("Start to do bucket meta query: {0}".format(self.bucket_name))
data = self.__convert_data(MetaQuery, xml_utils.to_do_bucket_meta_query_request, do_meta_query_request)
resp = self.__do_bucket('POST', data=data, params={Bucket.META_QUERY: '', Bucket.COMP: 'query'})
logger.debug("do bucket meta query done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_do_bucket_meta_query_result, DoBucketMetaQueryResult)
def close_bucket_meta_query(self):
"""关闭存储空间(Bucket)的元数据管理功能
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to close bucket meta query: {0}".format(self.bucket_name))
resp = self.__do_bucket('POST', params={Bucket.META_QUERY: '', Bucket.COMP: 'delete'})
logger.debug("bucket meta query done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def put_bucket_access_monitor(self, status):
"""更新 Bucket 访问跟踪状态。
:param str status : bucket访问跟踪的开启状态
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put bucket access monitor, bucket: {0}.".format(self.bucket_name))
data = xml_utils.to_put_bucket_access_monitor(status)
resp = self.__do_bucket('PUT', data=data, params={Bucket.ACCESS_MONITOR: ''})
logger.debug("bucket access monitor done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_access_monitor(self):
"""获取当前Bucket的访问跟踪的状态。
:return: :class:`GetBucketAccessMonitorResult <oss2.models.GetBucketAccessMonitorResult>`
"""
logger.debug("Start to get bucket access monitor: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.ACCESS_MONITOR: ''})
logger.debug("query list bucket cname done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_access_monitor_result, GetBucketAccessMonitorResult)
def get_bucket_resource_group(self):
"""查询存储空间(Bucket)的资源组ID。
:return: :class:`GetBucketResourceGroupResult <oss2.models.GetBucketResourceGroupResult>`
"""
logger.debug("Start to get bucket resource group: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.RESOURCE_GROUP: ''})
logger.debug("Get bucket resource group done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_resource_group_result, GetBucketResourceGroupResult)
def put_bucket_resource_group(self, resourceGroupId):
"""为存储空间(Bucket)配置所属资源组。
:param str resourceGroupId : Bucket所属的资源组ID。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put bucket resource group, bucket: {0}.".format(self.bucket_name))
data = xml_utils.to_put_bucket_resource_group(resourceGroupId)
resp = self.__do_bucket('PUT', data=data, params={Bucket.RESOURCE_GROUP: ''})
logger.debug("bucket resource group done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def put_bucket_style(self, styleName, content):
"""新增图片样式。
:param str styleName : 样式名称。
:param str content : 图片样式内容,图片样式可以包含一个或多个图片处理操作。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put bucket style, bucket: {0}.".format(self.bucket_name))
data = xml_utils.to_put_bucket_style(content)
resp = self.__do_bucket('PUT', data=data, params={Bucket.STYLE: '', Bucket.STYLE_NAME: styleName})
logger.debug("bucket style done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_style(self, styleName):
"""查询某个Bucket下指定的图片样式信息。
:param str styleName : 样式名称。
:return: :class:`GetBucketStyleResult <oss2.models.GetBucketStyleResult>`
"""
logger.debug("Start to get bucket style: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.STYLE: '', Bucket.STYLE_NAME: styleName})
logger.debug("Get bucket style done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_style_result, GetBucketStyleResult)
def list_bucket_style(self):
"""查询某个Bucket下已创建的所有图片样式。
:return: :class:`ListBucketStyleResult <oss2.models.ListBucketStyleResult>`
"""
logger.debug("Start to list bucket style: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.STYLE: ''})
logger.debug("query list bucket style done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_bucket_style, ListBucketStyleResult)
def delete_bucket_style(self, styleName):
"""删除某个Bucket下指定的图片样式。
:param str styleName : 样式名称。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete bucket style: {0}".format(self.bucket_name))
resp = self.__do_bucket('DELETE', params={Bucket.STYLE: '', Bucket.STYLE_NAME: styleName})
logger.debug("delete bucket style done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def async_process_object(self, key, process, headers=None):
"""异步处理多媒体接口。
:param str key: 处理的多媒体的对象名称
:param str process: 处理的字符串,例如"video/convert,f_mp4,vcodec_h265,s_1920x1080,vb_2000000,fps_30,acodec_aac,ab_100000,sn_1|sys/saveas,o_dGVzdC5qcGc,b_dGVzdA"
:param headers: HTTP头部
:type headers: 可以是dict,建议是oss2.CaseInsensitiveDict
"""
headers = http.CaseInsensitiveDict(headers)
logger.debug("Start to async process object, bucket: {0}, key: {1}, process: {2}".format(
self.bucket_name, to_string(key), process))
process_data = "%s=%s" % (Bucket.ASYNC_PROCESS, process)
resp = self.__do_object('POST', key, params={Bucket.ASYNC_PROCESS: ''}, headers=headers, data=process_data)
logger.debug("Async process object done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_async_process_object, AsyncProcessObject)
def put_bucket_callback_policy(self, callbackPolicy):
"""设置bucket回调策略
:param str callbackPolicy: 回调策略
"""
logger.debug("Start to put bucket callback policy, bucket: {0}, callback policy: {1}".format(self.bucket_name, callbackPolicy))
data = xml_utils.to_do_bucket_callback_policy_request(callbackPolicy)
resp = self.__do_bucket('PUT', data=data, params={Bucket.POLICY: '', Bucket.COMP: Bucket.CALLBACK})
logger.debug("Put bucket callback policy done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_callback_policy(self):
"""获取bucket回调策略
:return: :class:`GetBucketPolicyResult <oss2.models.CallbackPolicyResult>`
"""
logger.debug("Start to get bucket callback policy, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.POLICY: '', Bucket.COMP: Bucket.CALLBACK})
logger.debug("Get bucket callback policy done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_callback_policy_result, CallbackPolicyResult)
def delete_bucket_callback_policy(self):
"""删除bucket回调策略
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete bucket callback policy, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('DELETE', params={Bucket.POLICY: '', Bucket.COMP: Bucket.CALLBACK})
logger.debug("Delete bucket callback policy done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def put_bucket_archive_direct_read(self, enabled=False):
"""设置归档直读
:param boolean enabled: Bucket是否开启归档直读
"""
logger.debug("Start to put bucket archive direct read, bucket: {0}, enabled: {1}".format(self.bucket_name, enabled))
data = xml_utils.to_put_bucket_archive_direct_read(enabled)
resp = self.__do_bucket('PUT', data=data, params={Bucket.ARCHIVE_DIRECT_READ: ''})
logger.debug("bucket archive direct read done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_archive_direct_read(self):
"""获取归档直读
:return: :class:`GetBucketArchiveDirectReadResult <oss2.models.GetBucketArchiveDirectReadResult>`
"""
logger.debug("Start to get bucket archive direct read, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.ARCHIVE_DIRECT_READ: ''})
logger.debug("Get bucket archive direct read done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_archive_direct_read, GetBucketArchiveDirectReadResult)
def put_bucket_https_config(self, httpsConfig):
"""Bucket开启或关闭TLS版本设置。
:param httpsConfig: TLS版本信息设置
"""
logger.debug("Start to put bucket https config, bucket: {0}, https config: {1}".format(self.bucket_name, httpsConfig))
data = xml_utils.to_do_bucket_https_config_request(httpsConfig)
resp = self.__do_bucket('PUT', data=data, params={Bucket.HTTPS_CONFIG: ''})
logger.debug("Put bucket https config done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def create_bucket_data_redundancy_transition(self, targetType):
"""为Bucket创建存储冗余转换任务。
:param str targetType: 目标存储冗余类型
"""
logger.debug("Start to create bucket data redundancy transition, bucket: {0}, target type: {1}".format(self.bucket_name, targetType))
resp = self.__do_bucket('POST', params={Bucket.REDUNDANCY_TRANSITION: '', Bucket.TARGET_REDUNDANCY_TYPE: targetType})
logger.debug("Create bucket data redundancy transition done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_create_data_redundancy_transition_result, CreateDataRedundancyTransitionResult)
def get_bucket_data_redundancy_transition(self, taskId):
"""获取存储冗余转换任务。
:return: :class:`DataRedundancyTransitionInfoResult <oss2.models.DataRedundancyTransitionInfoResult>`
"""
logger.debug("Start to get bucket data redundancy transition, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.REDUNDANCY_TRANSITION: '', Bucket.REDUNDANCY_TRANSITION_TASK_ID: taskId})
logger.debug("Get bucket data redundancy transition done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_data_redundancy_transition, DataRedundancyTransitionInfoResult)
def delete_bucket_data_redundancy_transition(self, taskId):
"""删除存储冗余转换任务。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete bucket data redundancy transition, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('DELETE', params={Bucket.REDUNDANCY_TRANSITION: '', Bucket.REDUNDANCY_TRANSITION_TASK_ID: taskId})
logger.debug("Delete bucket data redundancy transition done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_https_config(self):
"""查看Bucket的TLS版本设置。
:return: :class:`HttpsConfigResult <oss2.models.HttpsConfigResult>`
"""
logger.debug("Start to get bucket https config, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.HTTPS_CONFIG: ''})
logger.debug("Get bucket https config done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_bucket_https_config, HttpsConfigResult)
def list_bucket_data_redundancy_transition(self):
"""列举某个Bucket下所有的存储冗余转换任务。
:return: :class:`ListBucketDataRedundancyTransitionResult <oss2.models.ListBucketDataRedundancyTransitionResult>`
"""
logger.debug("Start to do query list bucket data redundancy transition: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.REDUNDANCY_TRANSITION: ''})
logger.debug("query list bucket data redundancy transition done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_bucket_data_redundancy_transition, ListBucketDataRedundancyTransitionResult)
def create_access_point(self, accessPoint):
"""创建接入点
:param accessPoint :class:`CreateAccessPointRequest <oss2.models.CreateAccessPointRequest>`
:return: :class:`CreateAccessPointResult <oss2.models.CreateAccessPointResult>`
"""
logger.debug("Start to create access point, bucket: {0}".format(self.bucket_name))
data = xml_utils.to_do_create_access_point_request(accessPoint)
resp = self.__do_bucket('PUT', data=data, params={Bucket.ACCESS_POINT: ''})
logger.debug("Create access point done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_create_access_point_result, CreateAccessPointResult)
def get_access_point(self, accessPointName):
"""获取接入点信息
:param str accessPointName: 接入点名称
:return: :class:`GetAccessPointResult <oss2.models.GetAccessPointResult>`
"""
logger.debug("Start to get access point, bucket: {0}".format(self.bucket_name))
headers = http.CaseInsensitiveDict()
headers['x-oss-access-point-name'] = accessPointName
resp = self.__do_bucket('GET', params={Bucket.ACCESS_POINT: ''}, headers=headers)
logger.debug("Get access point done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_access_point_result, GetAccessPointResult)
def delete_access_point(self, accessPointName):
"""删除接入点
:param str accessPointName: 接入点名称
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete access point, bucket: {0}".format(self.bucket_name))
headers = http.CaseInsensitiveDict()
headers['x-oss-access-point-name'] = accessPointName
resp = self.__do_bucket('DELETE', params={Bucket.ACCESS_POINT: ''}, headers=headers)
logger.debug("Delete access point done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def list_bucket_access_points(self, max_keys=100, continuation_token=''):
"""查询某个Bucket下所有接入点。
param: int max_keys: 本次list返回access point的最大个数
param: str continuation_token: list时指定的起始标记
:return: :class:`ListAccessPointResult <oss2.models.ListAccessPointResult>`
"""
logger.debug("Start to list bucket access point: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.ACCESS_POINT: '', 'max-keys': str(max_keys), 'continuation-token': continuation_token})
logger.debug("query list bucket access point done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_access_point_result, ListAccessPointResult)
def put_access_point_policy(self, accessPointName, accessPointPolicy):
"""设置接入点策略
:param str accessPointName: 接入点名称
:param str accessPointPolicy : 接入点策略
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put access point policy, bucket: {0}, accessPointPolicy: {1}".format(self.bucket_name, accessPointPolicy))
headers = http.CaseInsensitiveDict()
headers['x-oss-access-point-name'] = accessPointName
resp = self.__do_bucket('PUT', data=accessPointPolicy, params={Bucket.ACCESS_POINT_POLICY: ''}, headers=headers)
logger.debug("Create access point policy done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_access_point_policy(self, accessPointName):
"""获取接入点策略
:param str accessPointName: 接入点名称
:return: :class:`GetAccessPointPolicyResult <oss2.models.GetAccessPointPolicyResult>`
"""
logger.debug("Start to get access point policy, bucket: {0}".format(self.bucket_name))
headers = http.CaseInsensitiveDict()
headers['x-oss-access-point-name'] = accessPointName
resp = self.__do_bucket('GET', params={Bucket.ACCESS_POINT_POLICY: ''}, headers=headers)
logger.debug("Get access point policy done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return GetAccessPointPolicyResult(resp)
def delete_access_point_policy(self, accessPointName):
"""删除接入点策略
:param str accessPointName: 接入点名称
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete access point policy, bucket: {0}".format(self.bucket_name))
headers = http.CaseInsensitiveDict()
headers['x-oss-access-point-name'] = accessPointName
resp = self.__do_bucket('DELETE', params={Bucket.ACCESS_POINT_POLICY: ''}, headers=headers)
logger.debug("Delete access point policy done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def put_bucket_public_access_block(self, block_public_access=False):
"""为Bucket开启阻止公共访问。
:param bool block_public_access : 是否开启阻止公共访问。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to bucket put public access block, bucket: {0}, enabled: {1}."
.format(self.bucket_name, block_public_access))
data = xml_utils.to_put_public_access_block_request(block_public_access)
resp = self.__do_bucket('PUT', data=data, params={Bucket.PUBLIC_ACCESS_BLOCK: ''})
logger.debug("bucket public access block done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_public_access_block(self):
"""获取指定Bucket的阻止公共访问配置信息。
:return: :class:`GetBucketPublicAccessBlockResult <oss2.models.GetBucketPublicAccessBlockResult>`
"""
logger.debug("Start to get bucket public access block: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.PUBLIC_ACCESS_BLOCK: ''})
logger.debug("Get bucket public access block done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_public_access_block_result, GetBucketPublicAccessBlockResult)
def delete_bucket_public_access_block(self):
"""删除指定Bucket的阻止公共访问配置信息。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete bucket public access block, bucket: {0}".format(self.bucket_name))
resp = self.__do_bucket('DELETE', params={Bucket.PUBLIC_ACCESS_BLOCK: ''})
logger.debug("Delete bucket public access block done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def put_access_point_public_access_block(self, access_point_name, block_public_access=False):
"""为接入点开启阻止公共访问。
:param bool block_public_access : 是否开启阻止公共访问。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put access point public access block, bucket: {0}, access point name: {1}, block public access: {2}."
.format(self.bucket_name, access_point_name, block_public_access))
data = xml_utils.to_put_public_access_block_request(block_public_access)
resp = self.__do_bucket('PUT', data=data, params={Bucket.PUBLIC_ACCESS_BLOCK: '', Bucket.OSS_ACCESS_POINT_NAME: access_point_name})
logger.debug("access point public access block done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_access_point_public_access_block(self, access_point_name):
"""获取指定接入点的阻止公共访问配置信息。
:return: :class:`GetBucketPublicAccessBlockResult <oss2.models.GetBucketPublicAccessBlockResult>`
"""
logger.debug("Start to get access point public access block: {0}, access point name: {1}.".format(self.bucket_name, access_point_name))
resp = self.__do_bucket('GET', params={Bucket.PUBLIC_ACCESS_BLOCK: '', Bucket.OSS_ACCESS_POINT_NAME: access_point_name})
logger.debug("Get access point public access block done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_public_access_block_result, GetBucketPublicAccessBlockResult)
def delete_access_point_public_access_block(self, access_point_name):
"""删除指定接入点的阻止公共访问配置信息。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete access point public access block, bucket: {0}, access point name: {1}.".format(self.bucket_name, access_point_name))
resp = self.__do_bucket('DELETE', params={Bucket.PUBLIC_ACCESS_BLOCK: '', Bucket.OSS_ACCESS_POINT_NAME: access_point_name})
logger.debug("Delete access point public access block done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def put_bucket_requester_qos_info(self, uid, qos_configuration):
"""修改请求者在Bucket上的流控配置。
:param str uid: 请求者UID
:param qos_configuration :class:`QoSConfiguration <oss2.models.QoSConfiguration>`
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to put bucket requester qos info, bucket: {0}, uid: {1}, qos_configuration: {2}."
.format(self.bucket_name, uid, qos_configuration))
if not uid:
raise ClientError('uid should not be empty')
data = xml_utils.to_put_qos_info(qos_configuration)
resp = self.__do_bucket('PUT', data=data, params={Bucket.REQUESTER_QOS_INFO: '', Bucket.QOS_REQUESTER: uid})
logger.debug("put bucket requester qos info done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def get_bucket_requester_qos_info(self, uid):
"""获取请求者在Bucket上的流控配置。
:return: :class:`RequesterQoSInfoResult <oss2.models.RequesterQoSInfoResult>`
"""
logger.debug("Start to get bucket requester qos info: {0}, uid: {1}.".format(self.bucket_name, uid))
if not uid:
raise ClientError('uid should not be empty')
resp = self.__do_bucket('GET', params={Bucket.REQUESTER_QOS_INFO: '', Bucket.QOS_REQUESTER: uid})
logger.debug("Get bucket requester qos info, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_get_requester_qos_info, RequesterQoSInfoResult)
def list_bucket_requester_qos_infos(self, continuation_token='', max_keys=100):
"""列举所有对该Bucket的请求者流控配置。
:param str continuation_token: 分页标志,首次调用传空串
:param int max_keys: 最多返回数目
:return: :class:`ListBucketRequesterQoSInfosResult <oss2.models.ListBucketRequesterQoSInfosResult>`
"""
logger.debug("Start to do query list bucket requester qos infos: {0}".format(self.bucket_name))
resp = self.__do_bucket('GET', params={Bucket.REQUESTER_QOS_INFO: '', 'continuation-token': continuation_token, 'max-keys': str(max_keys)})
logger.debug("query list bucket requester qos infos done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return self._parse_result(resp, xml_utils.parse_list_bucket_requester_qos_infos, ListBucketRequesterQoSInfosResult)
def delete_bucket_requester_qos_info(self, uid):
"""删除在Bucket上的请求者流控配置。
:return: :class:`RequestResult <oss2.models.RequestResult>`
"""
logger.debug("Start to delete bucket requester qos info, bucket: {0}, uid: {1}.".format(self.bucket_name, uid))
if not uid:
raise ClientError('uid should not be empty')
resp = self.__do_bucket('DELETE', params={Bucket.REQUESTER_QOS_INFO: '', Bucket.QOS_REQUESTER: uid})
logger.debug("Delete bucket requester qos info done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
return RequestResult(resp)
def __do_object(self, method, key, **kwargs):
if not self.bucket_name:
raise ClientError("Bucket name should not be null or empty.")
if not key:
raise ClientError("key should not be null or empty.")
return self._do(method, self.bucket_name, key, **kwargs)
def __do_bucket(self, method, **kwargs):
return self._do(method, self.bucket_name, '', **kwargs)
def __convert_data(self, klass, converter, data):
if isinstance(data, klass):
return converter(data)
else:
return data
def _normalize_endpoint(endpoint):
url = endpoint
if not endpoint.startswith('http://') and not endpoint.startswith('https://'):
url = 'http://' + endpoint
p = urlparse(url)
if p.port is not None:
return p.scheme + '://' + p.hostname + ':' + str(p.port)
else:
return p.scheme + '://' + p.hostname
_ENDPOINT_TYPE_ALIYUN = 0
_ENDPOINT_TYPE_CNAME = 1
_ENDPOINT_TYPE_IP = 2
_ENDPOINT_TYPE_PATH_STYLE = 3
def _make_range_string(range):
if range is None:
return ''
start = range[0]
last = range[1]
if start is None and last is None:
return ''
return 'bytes=' + _range(start, last)
def _range(start, last):
def to_str(pos):
if pos is None:
return ''
else:
return str(pos)
return to_str(start) + '-' + to_str(last)
def _determine_endpoint_type(netloc, is_cname, bucket_name, is_path_style):
if utils.is_ip_or_localhost(netloc):
return _ENDPOINT_TYPE_IP
if is_cname:
return _ENDPOINT_TYPE_CNAME
if is_path_style:
return _ENDPOINT_TYPE_PATH_STYLE
if utils.is_valid_bucket_name(bucket_name):
return _ENDPOINT_TYPE_ALIYUN
else:
return _ENDPOINT_TYPE_IP
class _UrlMaker(object):
def __init__(self, endpoint, is_cname, is_path_style):
p = urlparse(endpoint)
self.scheme = p.scheme
self.netloc = p.netloc
self.is_cname = is_cname
self.is_path_style = is_path_style
def __call__(self, bucket_name, key, slash_safe=False):
self.type = _determine_endpoint_type(self.netloc, self.is_cname, bucket_name, self.is_path_style)
safe = '/' if slash_safe is True else ''
key = urlquote(key, safe=safe)
if self.type == _ENDPOINT_TYPE_CNAME:
return '{0}://{1}/{2}'.format(self.scheme, self.netloc, key)
if self.type == _ENDPOINT_TYPE_IP or self.type == _ENDPOINT_TYPE_PATH_STYLE:
if bucket_name:
return '{0}://{1}/{2}/{3}'.format(self.scheme, self.netloc, bucket_name, key)
else:
return '{0}://{1}/{2}'.format(self.scheme, self.netloc, key)
if not bucket_name:
assert not key
return '{0}://{1}'.format(self.scheme, self.netloc)
return '{0}://{1}.{2}/{3}'.format(self.scheme, bucket_name, self.netloc, key)