AutoAndroidController/py/venv/Lib/site-packages/baidubce/services/bos/bos_client.py

# Copyright 2014 Baidu, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
# except in compliance with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the
# License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
# either express or implied. See the License for the specific language governing permissions
# and limitations under the License.

"""
This module provides a client class for BOS.
"""

import io
import copy
import http.client
import os
import json
import logging
import shutil
import struct
from builtins import str
from builtins import bytes
from future.utils import iteritems, iterkeys, itervalues
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED
import threading
import functools
import multiprocessing

import baidubce
from baidubce import bce_client_configuration
from baidubce import utils
from baidubce.auth import bce_v1_signer
from baidubce.bce_base_client import BceBaseClient
from baidubce.exception import BceClientError
from baidubce.exception import BceServerError
from baidubce.exception import BceHttpClientError
from baidubce.http import bce_http_client
from baidubce.http import handler
from baidubce.http import http_content_types
from baidubce.http import http_headers
from baidubce.http import http_methods
from baidubce.services import bos
from baidubce.services.bos import bos_handler
from baidubce.services.bos import storage_class
from baidubce.utils import required
from baidubce import compat

_logger = logging.getLogger(__name__)

FETCH_MODE_SYNC = b"sync"
FETCH_MODE_ASYNC = b"async"

ENCRYPTION_ALGORITHM= "AES256"

HTTP_PROTOCOL_HEAD = b'http'


class UploadTaskHandle:
    """
    handle to control multi upload file with multi-thread
    """
    def __init__(self):
        self.cancel_flag = False
        self.cancel_lock = threading.Lock()

    def cancel(self):
        """
        cancel putting super object from file with multi-thread
        """
        self.cancel_lock.acquire()
        self.cancel_flag= True
        self.cancel_lock.release()

    def is_cancel(self):
        """
        get cancel flag
        """
        self.cancel_lock.acquire()
        result = self.cancel_flag
        self.cancel_lock.release()
        return result


class BosClient(BceBaseClient):
    """
    sdk client
    """
    def __init__(self, config=None):
        BceBaseClient.__init__(self, config)

    def list_buckets(self, config=None):
        """
        List buckets of user

        :param config: None
        :type config: BceClientConfiguration
        :returns: all buckets owned by the user.
        :rtype: baidubce.bce_response.BceResponse
        """
        return self._send_request(http_methods.GET, config=config)

    @required(bucket_name=(bytes, str))
    def get_bucket_location(self, bucket_name, config=None):
        """
        Get the region which the bucket located in.

        :param bucket_name: the name of bucket
        :type bucket_name: string or unicode
        :param config: None
        :type config: BceClientConfiguration

        :return: region of the bucket
        :rtype: str
        """
        params = {b'location': b''}
        response = self._send_request(http_methods.GET, bucket_name, params=params, config=config)
        return response.location_constraint

    @required(bucket_name=(bytes, str))
    def create_bucket(self, bucket_name, config=None):
        """
        Create bucket with specific name

        :param bucket_name: the name of bucket
        :type bucket_name: string or unicode
        :param config: None
        :type config: BceClientConfiguration
        :returns:
        :rtype: baidubce.bce_response.BceResponse
        """
        return self._send_request(http_methods.PUT, bucket_name, config=config)

    @required(bucket_name=(bytes, str))
    def does_bucket_exist(self, bucket_name, config=None):
        """
        Check whether there is a bucket with specific name

        :param bucket_name: None
        :type bucket_name: str
        :return:True or False
        :rtype: bool
        """
        try:
            self._send_request(http_methods.HEAD, bucket_name, config=config)
            return True
        except BceHttpClientError as e:
            if isinstance(e.last_error, BceServerError):
                if e.last_error.status_code == http.client.FORBIDDEN:
                    return True
                if e.last_error.status_code == http.client.NOT_FOUND:
                    return False
            raise e

    @required(bucket_name=(bytes, str))
    def get_bucket_acl(self, bucket_name, config=None):
        """
        Get Access Control Level of bucket

        :type bucket: string
        :param bucket: None
        :return:
            **json text of acl**
        """
        return self._send_request(
                http_methods.GET,
                bucket_name,
                params={b'acl': b''},
                config=config)

    @staticmethod
    def _dump_acl_object(acl):
        result = {}
        for k, v in iteritems(acl.__dict__):
            if not k.startswith('_'):
                result[k] = v
        return result

    @required(bucket_name=(bytes, str), acl=(list, dict))
    def set_bucket_acl(self, bucket_name, acl, config=None):
        """
        Set Access Control Level of bucket

        :type bucket: string
        :param bucket: None

        :type grant_list: list of grant
        :param grant_list: None
        :return:
            **HttpResponse Class**
        """
        self._send_request(http_methods.PUT,
                           bucket_name,
                           body=json.dumps({'accessControlList': acl},
                                           default=BosClient._dump_acl_object),
                           headers={http_headers.CONTENT_TYPE: http_content_types.JSON},
                           params={b'acl': b''},
                           config=config)

    @required(bucket_name=(bytes, str), canned_acl=bytes)
    def set_bucket_canned_acl(self, bucket_name, canned_acl, config=None):
        """

        :param bucket_name:
        :param canned_acl:
        :param config:
        :return:
        """
        self._send_request(http_methods.PUT,
                           bucket_name,
                           headers={http_headers.BCE_ACL: canned_acl},
                           params={b'acl': b''},
                           config=config)

    @required(bucket_name=(bytes, str))
    def set_bucket_storage_class(self, bucket_name, storage_class, config=None):
        """

        :param bucket_name:
        :param config:
        :return:
        """
        storage_class = compat.convert_to_string(storage_class)
        return self._send_request(http_methods.PUT,
                           bucket_name,
                           body=json.dumps({'storageClass': storage_class},
                                           default=BosClient._dump_acl_object),
                           headers={http_headers.CONTENT_TYPE: http_content_types.JSON},
                           params={b'storageClass': b''},
                           config=config)


    @required(bucket_name=(bytes, str))
    def get_bucket_storage_class(self, bucket_name, config=None):
        """

        :param bucket_name:
        :param config:
        :return:
        """
        return self._send_request(http_methods.GET,
                           bucket_name,
                           params={b'storageClass': b''},
                           config=config)

    @required(bucket_name=(bytes, str))
    def delete_bucket(self, bucket_name, config=None):
        """
        Delete a Bucket(Must Delete all the Object in Bucket before)

        :type bucket: string
        :param bucket: None
        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.DELETE, bucket_name, config=config)

# bucket static website
    @required(bucket_name=(bytes, str))
    def put_bucket_static_website(self, bucket_name, index=None, not_found=None, config=None):
        """
        Set index page and not_found 404 page for static website trusteeship

        :type bucket_name: string
        :param bucket_name: None

        :type index:string
        :param index:object name of index page for static website trusteeship

        :type not_found:string
        :param not_found:object name of not_found 404 page for static website trusteeship

        :return:
            **HttpResponse Class**
        """
        body = {}
        if index is not None:
            body['index'] = index
        if not_found is not None:
            body['notFound'] = not_found
        return self._send_request(http_methods.PUT,
                           bucket_name,
                           body=json.dumps(body,
                                           default=BosClient._dump_acl_object),
                           headers={http_headers.CONTENT_TYPE: http_content_types.JSON},
                           params={b'website': b''},
                           config=config)

    @required(bucket_name=(bytes, str))
    def get_bucket_static_website(self, bucket_name, config=None):
        """
        Get Information of static website trusteeship

        :type bucket: string
        :param bucket: None

        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.GET,
                           bucket_name,
                           params={b'website': b''},
                           config=config)


    @required(bucket_name=(bytes, str))
    def delete_bucket_static_website(self, bucket_name, config=None):
        """
        Delete Information of static website trusteeship to be closed

        :type bucket: string
        :param bucket: None

        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.DELETE,
                           bucket_name,
                           params={b'website': b''},
                           config=config)

# bucket encryption
    @required(bucket_name=(bytes, str))
    def put_bucket_encryption(self, bucket_name, encryption_algorithm=ENCRYPTION_ALGORITHM, config=None):
        """
        Set server encryption for bucket

        :type bucket: string
        :param bucket: None

        :type encryption_algorithm: string
        :param grant_list: server encryption algorithm for bucekt.Now the value of encryption_algorithm
        only is 'AES256'

        :return:
            **HttpResponse Class**
        """
        encryption_algorithm = compat.convert_to_string(encryption_algorithm)
        return self._send_request(http_methods.PUT,
                           bucket_name,
                           body=json.dumps({"encryptionAlgorithm":encryption_algorithm},
                                           default=BosClient._dump_acl_object),
                           headers={http_headers.CONTENT_TYPE: http_content_types.JSON},
                           params={b'encryption': b''},
                           config=config)

    @required(bucket_name=(bytes, str))
    def get_bucket_encryption(self, bucket_name, config=None):
        """
        Get status of server encryption

        :type bucket: string
        :param bucket: None

        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.GET,
                           bucket_name,
                           params={b'encryption': b''},
                           config=config)


    @required(bucket_name=(bytes, str))
    def delete_bucket_encryption(self, bucket_name, config=None):
        """
        Close server encryption

        :type bucket: string
        :param bucket: None

        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.DELETE,
                           bucket_name,
                           params={b'encryption': b''},
                           config=config)

# Bucket Copyright Protection

    @required(bucket_name=(bytes, str), resource=(list))
    def put_bucket_copyright_protection(self, bucket_name, resource, config=None):
        """
        Open image copyright protection and set resource

        :type bucket: string
        :param bucket: None

        :type resource: list of  string
        :param grant_list: resource range to be protected

        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.PUT,
                           bucket_name,
                           body=json.dumps({"resource": resource},
                                           default=BosClient._dump_acl_object),
                           headers={http_headers.CONTENT_TYPE: http_content_types.JSON},
                           params={b'copyrightProtection': b''},
                           config=config)

    @required(bucket_name=(bytes, str))
    def get_bucket_copyright_protection(self, bucket_name, config=None):
        """
        Get configuration of image copyright protection

        :type bucket: string
        :param grant_list: None

        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.GET,
                           bucket_name,
                           params={b'copyrightProtection': b''},
                           config=config)


    @required(bucket_name=(bytes, str))
    def delete_bucket_copyright_protection(self, bucket_name, config=None):
        """
        Close image copyright protection

        :type bucket: string
        :param bucket: None

        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.DELETE,
                           bucket_name,
                           params={b'copyrightProtection': b''},
                           config=config)

# bucket replication
    @required(bucket_name=(bytes, str), replication=(dict))
    def put_bucket_replication(self, bucket_name, replication, config=None):
        """
        Open cross-region replication

        :type bucket: string
        :param bucket: None

        :type replication: dict
        :type replication: configuration for cross-region replication

        :return:
            **HttpResponse Class**
        """
        params={b'replication': b''}
        if "id" in replication:
            params[b"id"] = compat.convert_to_bytes(replication["id"])
        return self._send_request(http_methods.PUT,
                           bucket_name,
                           body=json.dumps(replication,
                                           default=BosClient._dump_acl_object),
                           headers={http_headers.CONTENT_TYPE: http_content_types.JSON},
                           params=params,
                           config=config)

    @required(bucket_name=(bytes, str))
    def get_bucket_replication(self, bucket_name, id=None, config=None):
        """
        Get configuration of cross-region replication 

        :type bucket: string
        :param bucket: None

        :type id: string
        :param id: replication rule id

        :return:
            **HttpResponse Class**
        """
        params={b'replication': b''}
        if id is not None:
            params[b"id"] = compat.convert_to_bytes(id)
        return self._send_request(http_methods.GET,
                           bucket_name,
                           params=params,
                           config=config)


    @required(bucket_name=(bytes, str))
    def delete_bucket_replication(self, bucket_name, id=None, config=None):
        """
        Delete configuration of cross-region replication and close it 

        :type bucket: string
        :param bucket: None

        :type id: string
        :param id: replication rule id

        :return:
            **HttpResponse Class**
        """
        params={b'replication': b''}
        if id is not None:
            params[b"id"] = compat.convert_to_bytes(id)
        return self._send_request(http_methods.DELETE,
                           bucket_name,
                           params=params,
                           config=config)


    @required(bucket_name=(bytes, str))
    def get_bucket_replication_progress(self, bucket_name, id=None, config=None):
        """
        Get status of cross-region replication,for exapmle 'historyReplicationPercent',
        'latestReplicationTime'

        :type bucket: string
        :param bucket: None

        :type id: string
        :param id: replication rule id

        :return:
            **HttpResponse Class**
        """
        params={b'replicationProgress': b''}
        if id is not None:
            params[b"id"] = compat.convert_to_bytes(id)
        return self._send_request(http_methods.GET,
                           bucket_name,
                           params=params,
                           config=config)
    
    @required(bucket_name=(bytes, str))
    def list_bucket_replication(self, bucket_name, config=None):
        """
        list configuration of cross-region replication rule

        :type bucket: string
        :param bucket: None

        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.GET,
                           bucket_name,
                           params={b'replication': b'', b'list': b''},
                           config=config)

    @required(bucket_name=(bytes, str), inventory=(dict))
    def put_bucket_inventory(self, bucket_name, inventory, config=None):
        """
        set bucket inventoru

        :type bucket: string
        :param bucket: None

        :type inventory: dict
        :param inventory: configuration for bucket inventory

        :return:
            **HttpResponse Class**
        """
        conf_id = compat.convert_to_bytes(inventory["id"])
        return self._send_request(http_methods.PUT,
                           bucket_name,
                           body=json.dumps(inventory,
                                           default=BosClient._dump_acl_object),
                           headers={http_headers.CONTENT_TYPE: http_content_types.JSON},
                           params={b'inventory': b'', b'id': conf_id},
                           config=config)

    @required(bucket_name=(bytes, str), inventory_conf_id=(bytes, str))
    def get_bucket_inventory(self, bucket_name, inventory_conf_id, config=None):
        """
        Get configuration of bucket inventory

        :type bucket: string
        :param bucket: None

        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.GET,
                           bucket_name,
                           params={b'inventory': b'', b'id': compat.convert_to_bytes(inventory_conf_id)},
                           config=config)

    @required(bucket_name=(bytes, str), inventory_conf_id=(bytes, str))
    def delete_bucket_inventory(self, bucket_name, inventory_conf_id, config=None):
        """
        Delete configuration of bucket inventory

        :type bucket: string
        :param bucket: None

        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.DELETE,
                           bucket_name,
                           params={b'inventory': b'', b'id': compat.convert_to_bytes(inventory_conf_id)},
                           config=config)

    @required(bucket_name=(bytes, str))
    def list_bucket_inventory(self, bucket_name, config=None):
        """
        list configuration of bucket inventory

        :type bucket: string
        :param bucket: None

        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.GET,
                           bucket_name,
                           params={b'inventory': b''},
                           config=config)

    @required(bucket_name=(bytes, str))
    def put_bucket_trash(self, bucket_name, trash_dir=None, config=None):
        """
        Open bucket trash function

        :type bucket: string
        :param bucket: None

        :type trash_dir: string
        :param trash_dir: directory of trash,optional

        :return:
            **HttpResponse Class**
        """
        if trash_dir is not None:
            trash_dir = compat.convert_to_string(trash_dir)
        return self._send_request(http_methods.PUT,
                           bucket_name,
                           body=json.dumps({"trashDir": trash_dir},
                                           default=BosClient._dump_acl_object),
                           headers={http_headers.CONTENT_TYPE: http_content_types.JSON},
                           params={b'trash': b''},
                           config=config)

    @required(bucket_name=(bytes, str))
    def get_bucket_trash(self, bucket_name, config=None):
        """
        Get status of bucket trash

        :type bucket: string
        :param grant_list: None

        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.GET,
                           bucket_name,
                           params={b'trash': b''},
                           config=config)


    @required(bucket_name=(bytes, str))
    def delete_bucket_trash(self, bucket_name, config=None):
        """
        Close bucket trash

        :type bucket: string
        :param bucket: None

        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.DELETE,
                           bucket_name,
                           params={b'trash': b''},
                           config=config)

    @required(bucket_name=(bytes, str), key=(bytes, str))
    def generate_pre_signed_url(self,
                                bucket_name,
                                key,
                                timestamp=0,
                                expiration_in_seconds=1800,
                                headers=None,
                                params=None,
                                headers_to_sign=None,
                                protocol=None,
                                config=None,
                                httpmethod=http_methods.GET):
        """
        Get an authorization url with expire time.
        specified  protocol in endpoint > protocal > default protocol in config.

        :type timestamp: int
        :param timestamp: None

        :type expiration_in_seconds: int
        :param expiration_in_seconds: None

        :type options: dict
        :param options: None

        :param is_official_domain: default use not official domain,example: bucket.bj.bcebos.com

        :return:
            **URL string**
        """
        key = compat.convert_to_bytes(key)
        if len(key) == 0 or key == b'v1':
            raise ValueError('generate url the key param error!')

        config = self._merge_config(config, bucket_name)
        headers = headers or {}
        params = params or {}

        # specified  protocol in endpoint > protocal > default protocol in config
        if protocol is not None:
            config.protocol = protocol
        endpoint_protocol, endpoint_host, endpoint_port = \
            utils.parse_host_port(config.endpoint, config.protocol)

        full_host = endpoint_host
        if endpoint_port != endpoint_protocol.default_port:
            full_host += b':' + compat.convert_to_bytes(endpoint_port)
        headers[http_headers.HOST] = full_host

        path = self._get_path(config, bucket_name, key)
        if httpmethod != http_methods.GET and httpmethod != http_methods.HEAD:
            headers_to_sign = set([b'host'])

        # Compatible with STS request acquisition
        if config.security_token is not None:
            params[http_headers.STS_SECURITY_TOKEN.lower()] = config.security_token

        params[http_headers.AUTHORIZATION.lower()] = bce_v1_signer.sign(
            config.credentials,
            httpmethod,
            path,
            headers,
            params,
            timestamp,
            expiration_in_seconds,
            headers_to_sign)
        
        return b"%s://%s%s?%s" % (compat.convert_to_bytes(endpoint_protocol.name),
                                 full_host,
                                 path,
                                 utils.get_canonical_querystring(params, False))

    @required(bucket_name=(bytes, str), rules=(list, dict))
    def put_bucket_lifecycle(self, 
                             bucket_name,
                             rules,
                             config=None):
        """
        Put Bucket Lifecycle
       
        :type bucket: string
        :param bucket: None

        :type rules: list
        :param rules: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.PUT, 
                                  bucket_name,
                                  params={b'lifecycle': b''},
                                  body=json.dumps({'rule': rules}),
                                  config=config)

    @required(bucket_name=(bytes, str))
    def get_bucket_lifecycle(self, bucket_name, config=None):
        """
        Get Bucket Lifecycle

        :type bucket: string
        :param bucket: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.GET, 
                                  bucket_name, 
                                  params={b'lifecycle': b''},
                                  config=config) 

    @required(bucket_name=(bytes, str))
    def delete_bucket_lifecycle(self, bucket_name, config=None):
        """
        Delete Bucket Lifecycle
        
        :type bucket: string
        :param bucket: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.DELETE,
                                  bucket_name,
                                  params={b'lifecycle': b''},
                                  config=config)       
    
    @required(bucket_name=(bytes, str), cors_configuration=list)
    def put_bucket_cors(self, 
                        bucket_name,
                        cors_configuration,
                        config=None):
        """
        Put Bucket Cors
        :type bucket: string
        :param bucket: None

        :type cors_configuration: list
        :param cors_configuration: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.PUT,
                                  bucket_name,
                                  params={b'cors': b''},
                                  body=json.dumps({'corsConfiguration': cors_configuration}),
                                  config=config)

    @required(bucket_name=(bytes, str))
    def get_bucket_cors(self, bucket_name, config=None):
        """
        Get Bucket Cors

        :type bucket: string
        :param bucket: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.GET,
                                  bucket_name,
                                  params={b'cors': b''},
                                  config=config)

    @required(bucket_name=(bytes, str))
    def delete_bucket_cors(self, bucket_name, config=None):
        """
        Delete Bucket Cors

        :type bucket: string
        :param bucket: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.DELETE,
                                  bucket_name,
                                  params={b'cors': b''},
                                  config=config)

    @required(bucket_name=(bytes, str))        
    def list_objects(self, bucket_name,
                     max_keys=1000, prefix=None, marker=None, delimiter=None,
                     config=None):
        """
        Get Object Information of bucket

        :type bucket: string
        :param bucket: None

        :type delimiter: string
        :param delimiter: None

        :type marker: string
        :param marker: None

        :type max_keys: int
        :param max_keys: value <= 1000

        :type prefix: string
        :param prefix: None

        :return:
            **_ListObjectsResponse Class**
        """
        params = {}
        if max_keys is not None:
            params[b'maxKeys'] = max_keys
        if prefix is not None:
            params[b'prefix'] = prefix
        if marker is not None:
            params[b'marker'] = marker
        if delimiter is not None:
            params[b'delimiter'] = delimiter

        return self._send_request(http_methods.GET, bucket_name, params=params, config=config)

    @required(bucket_name=(bytes, str))
    def list_all_objects(self, bucket_name, prefix=None, delimiter=None, config=None):
        """

        :param bucket_name:
        :param prefix:
        :param delimiter:
        :param config:
        :return:
        """
        marker = None
        while True:
            response = self.list_objects(
                bucket_name, marker=marker, prefix=prefix, delimiter=delimiter, config=config)
            for item in response.contents:
                yield item
            if response.is_truncated:
                marker = response.next_marker
            else:
                break

    @staticmethod
    def _get_range_header_dict(range):
        if range is None:
            return None
        if not isinstance(range, (list, tuple)):
            raise TypeError('range should be a list or a tuple')
        if len(range) != 2:
            raise ValueError('range should have length of 2')
        return {http_headers.RANGE: b'bytes=%d-%d' % tuple(range)}


    @staticmethod
    def _parse_bos_object(http_response, response):
        """Sets response.body to http_response and response.user_metadata to a dict consists of all http
        headers starts with 'x-bce-meta-'.

        :param http_response: the http_response object returned by HTTPConnection.getresponse()
        :type http_response: httplib.HTTPResponse

        :param response: general response object which will be returned to the caller
        :type response: baidubce.BceResponse

        :return: always true
        :rtype bool
        """
        user_metadata = {}
        headers_list = http_response.getheaders()
        if compat.PY3:
            temp_heads = []
            for k, v in headers_list:
                k = k.lower()
                temp_heads.append((k, v))
            headers_list = temp_heads

        prefix = compat.convert_to_string(
                http_headers.BCE_USER_METADATA_PREFIX
        )
        for k, v in headers_list:
            if k.startswith(prefix):
                k = k[len(prefix):]
                user_metadata[compat.convert_to_unicode(k)] = \
                    compat.convert_to_unicode(v)
        response.metadata.user_metadata = user_metadata
        response.data = http_response
        return True

    @required(bucket_name=(bytes, str), key=(bytes, str))
    def get_object(self, bucket_name, key, range=None, traffic_limit=None, version_id=None, 
                   cond_read_write=None, config=None):
        """

        :param bucket_name:
        :param key:
        :param range:
        :param traffic_limit:
        :param version_id:
        :param cond_read_write:
        :param config:
        :return:
        """
        query_params = {}
        if version_id is not None:
            version_id = compat.convert_to_bytes(version_id)
            query_params={b'versionId': version_id}
        key = compat.convert_to_bytes(key)
        if len(key) == 0 or key.startswith(b"/"):
            raise BceClientError("Key can not be empty or start with '/' .")
        range_header = BosClient._get_range_header_dict(range)

        if traffic_limit is not None:
            if range_header is None:
                range_header = {}
            range_header[http_headers.BOS_TRAFFIC_LIMIT] = traffic_limit
        
        if cond_read_write is not None:
            if range_header is None:
                range_header = {}
            range_header = self._get_cond_read_write_headers(http_methods.GET, range_header, cond_read_write)
        
        return self._send_request(
            http_methods.GET,
            bucket_name,
            key,
            headers=range_header,
            params=query_params,
            config=config,
            body_parser=BosClient._parse_bos_object)

    # restore object
    @required(bucket_name=(bytes, str), key=(bytes, str))
    def restore_object(self, bucket_name, key, days=None, tier="Standard", config=None):
        """

        :param bucket_name:
        :param key:
        :param config:
        :return:
        """
        key = compat.convert_to_bytes(key)
        headers = {}
        if days is not None:
            headers[http_headers.BOS_RESTORE_DAYS] = days
        if compat.convert_to_string(tier) not in ("Standard", "Expedited", "LowCost"):
            raise ValueError('invalid tier:{} for restore_object.The valid value is \"Standard\" or \"Expedited\" or '\
                    '\"LowCost\"'.format(tier) )
        headers[http_headers.BOS_RESTORE_TIER] = compat.convert_to_bytes(tier)
        return self._send_request(
            http_methods.POST,
            bucket_name,
            key,
            headers=headers,
            params={b'restore': b''},
            config=config,
            body_parser=BosClient._parse_bos_object)

    @staticmethod
    def _save_body_to_file(http_response, response, file_name, buf_size=16 * 1024, progress_callback=None):
        f = open(file_name, 'wb')
        try:
            # Added progress bar monitoring
            if progress_callback:
                file_size = int(response.metadata.content_length)
                stream = utils.make_progress_adapter(http_response, progress_callback, file_size)
            else:
                stream = http_response

            shutil.copyfileobj(stream, f, buf_size)
            http_response.close()
        finally:
            f.close()
        return True

    @staticmethod
    def _parse_select_message(http_response, response, select_response):
        select_response.init_from_http_response(http_response, response)
        return True

    @required(bucket_name=(bytes, str), key=(bytes, str))
    def get_object_as_string(self, bucket_name, key, range=None, version_id=None, cond_read_write=None, config=None):
        """

        :param bucket_name:
        :param key:
        :param range:
        :param config:
        :return:
        """
        key = compat.convert_to_bytes(key)
        response = self.get_object(bucket_name, key, range=range, version_id = version_id,
                                   cond_read_write = cond_read_write, config=config)
        s = response.data.read()
        response.data.close()
        return s

    @required(bucket_name=(bytes, str), key=(bytes, str), file_name=(bytes, str))
    def get_object_to_file(self, bucket_name, key, file_name, range=None, config=None, 
                            progress_callback=None, traffic_limit=None, cond_read_write=None, version_id=None):
        """
        Get Content of Object and Put Content to File

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None

        :type file_name: string
        :param file_name: None

        :type range: tuple
        :param range: (0,9) represent get object contents of 0-9 in bytes. 10 bytes date in total.
        :return:
            **HTTP Response**
        """
        query_params = {}
        if version_id is not None:
            version_id = compat.convert_to_bytes(version_id)
            query_params={b'versionId': version_id}
        key = compat.convert_to_bytes(key)
        if len(key) == 0 or key.startswith(b"/"):
            raise BceClientError("Key can not be empty or start with '/' .")
        file_name = compat.convert_to_bytes(file_name)
        range_header = BosClient._get_range_header_dict(range)
        if traffic_limit is not None:
            if range_header is None:
                range_header = {}
            range_header[http_headers.BOS_TRAFFIC_LIMIT] = traffic_limit
        
        if cond_read_write is not None:
            if range_header is None:
                range_header = {}
            range_header = self._get_cond_read_write_headers(http_methods.GET, range_header, cond_read_write)
        
        return self._send_request(
            http_methods.GET,
            bucket_name,
            key,
            headers=range_header,
            params=query_params,
            config=config,
            body_parser=lambda http_response, response: BosClient._save_body_to_file(
                http_response,
                response,
                file_name,
                self._get_config_parameter(config, 'recv_buf_size'),
                progress_callback=progress_callback))


    @required(bucket_name=(bytes, str), key=(bytes, str))
    def get_object_meta_data(self, bucket_name, key, version_id=None, cond_read_write=None, config=None):
        """
        Get head of object

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None
        :return:
            **_GetObjectMetaDataResponse Class**
        """
        query_params = {}
        if version_id is not None:
            version_id = compat.convert_to_bytes(version_id)
            query_params={b'versionId': version_id}

        headers = {}
        if cond_read_write is not None:
            headers = self._get_cond_read_write_headers(http_methods.HEAD, headers, cond_read_write)
        
        key = compat.convert_to_bytes(key)
        return self._send_request(http_methods.HEAD, bucket_name, key,
                                  headers=headers, params=query_params, config=config)

    @required(bucket_name=(bytes, str),
              key=(bytes, str),
              data=object,
              content_length=compat.integer_types,
              content_md5=(bytes, str))
    def append_object(self, bucket_name, key, data,
                     content_md5,
                     content_length,
                     offset=None,
                     content_type=None,
                     user_metadata=None,
                     content_sha256=None,
                     storage_class=None,
                     user_headers=None,
                     progress_callback=None,
                     traffic_limit=None,
                     object_tagging=None,
                     config=None):
        """
        Put an appendable object to BOS or add content to an appendable object

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None

        :type content_length: long
        :type offset: long
        :return:
            **HTTP Response**
        """
        key = compat.convert_to_bytes(key)
        content_md5 = compat.convert_to_bytes(content_md5)
        headers = self._prepare_object_headers(
            content_length=content_length,
            content_md5=content_md5,
            content_type=content_type,
            content_sha256=content_sha256,
            user_metadata=user_metadata,
            storage_class=storage_class,
            user_headers=user_headers,
            traffic_limit=traffic_limit,
            object_tagging=object_tagging)

        if content_length > bos.MAX_APPEND_OBJECT_LENGTH:
            raise ValueError('Object length should be less than %d. '
                             'Use multi-part upload instead.' % bos.MAX_APPEND_OBJECT_LENGTH)

        params = {b'append': b''}
        if offset is not None:
            params[b'offset'] = offset
        
        if progress_callback:
            data = utils.make_progress_adapter(data, progress_callback)

        return self._send_request(
            http_methods.POST,
            bucket_name,
            key,
            body=data,
            headers=headers,
            params=params,
            config=config)

    @required(bucket_name=(bytes, str),
                           key=(bytes, str),
                           data=(bytes, str))
    def append_object_from_string(self, bucket_name, key, data,
                                  content_md5=None,
                                  offset=None,
                                  content_type=None,
                                  user_metadata=None,
                                  content_sha256=None,
                                  storage_class=None,
                                  user_headers=None,
                                  progress_callback=None,
                                  traffic_limit=None,
                                  config=None):
        """
        Create an appendable object and put content of string to the object
        or add content of string to an appendable object
        """
        key = compat.convert_to_bytes(key)
        if isinstance(data, str):
            data = data.encode(baidubce.DEFAULT_ENCODING)

        fp = None
        try:
            fp = io.BytesIO(data)
            if content_md5 is None:
                content_md5 = utils.get_md5_from_fp(
                    fp, buf_size=self._get_config_parameter(config, 'recv_buf_size'))

            return self.append_object(bucket_name=bucket_name,
                                      key=key,
                                      data=fp,
                                      content_md5=content_md5,
                                      content_length=len(data),
                                      offset=offset,
                                      content_type=content_type,
                                      user_metadata=user_metadata,
                                      content_sha256=content_sha256,
                                      storage_class=storage_class,
                                      user_headers=user_headers,
                                      progress_callback=progress_callback,
                                      traffic_limit=traffic_limit,
                                      config=config)
        finally:
            if fp is not None:
                fp.close()

    @required(bucket_name=(bytes, str),
              key=(bytes, str),
              data=object,
              content_length=compat.integer_types,
              content_md5=(bytes, str))
    def put_object(self, bucket_name, key, data,
                   content_length,
                   content_md5,
                   content_type=None,
                   content_sha256=None,
                   user_metadata=None,
                   storage_class=None,
                   user_headers=None,
                   encryption=None,
                   customer_key=None,
                   customer_key_md5=None,
                   progress_callback=None,
                   traffic_limit=None,
                   object_tagging=None,
                   cond_read_write=None,
                   config=None):
        """
        Put object and put content of file to the object

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None

        :type fp: FILE
        :param fp: None

        :type file_size: long
        :type offset: long
        :type content_length: long
        :return:
            **HTTP Response**
        """
        key = compat.convert_to_bytes(key)
        content_md5 = compat.convert_to_bytes(content_md5)
        headers = self._prepare_object_headers(
            content_length=content_length,
            content_md5=content_md5,
            content_type=content_type,
            content_sha256=content_sha256,
            user_metadata=user_metadata,
            storage_class=storage_class,
            user_headers=user_headers,
            traffic_limit=traffic_limit,
            object_tagging=object_tagging,)

        if cond_read_write is not None:
            headers = self._get_cond_read_write_headers(http_methods.PUT, headers, cond_read_write)

        buf_size = self._get_config_parameter(config, 'recv_buf_size')

        if content_length > bos.MAX_PUT_OBJECT_LENGTH:
            raise ValueError('Object length should be less than %d. '
                             'Use multi-part upload instead.' % bos.MAX_PUT_OBJECT_LENGTH)
        
        if progress_callback:
                data = utils.make_progress_adapter(data, progress_callback)

        return self._send_request(
            http_methods.PUT,
            bucket_name,
            key,
            body=data,
            headers=headers,
            config=config)

    @required(bucket=(bytes, str), key=(bytes, str), data=(bytes, str))
    def put_object_from_string(self, bucket, key, data,
                               content_md5=None,
                               content_type=None,
                               content_sha256=None,
                               user_metadata=None,
                               storage_class=None,
                               user_headers=None,
                               encryption=None,
                               customer_key=None,
                               customer_key_md5=None,
                               progress_callback=None,
                               traffic_limit=None,
                               object_tagging=None,
                               cond_read_write=None,
                               config=None):
        """
        Create object and put content of string to the object

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None

        :type input_content: string
        :param input_content: None

        :type options: dict
        :param options: None
        :return:
            **HTTP Response**
        """
        key = compat.convert_to_bytes(key)
        if isinstance(data, str):
            data = data.encode(baidubce.DEFAULT_ENCODING)

        fp = None
        try:
            fp = io.BytesIO(data)
            if content_md5 is None:
                content_md5 = utils.get_md5_from_fp(
                    fp, buf_size=self._get_config_parameter(config, 'recv_buf_size'))
            return self.put_object(bucket, key, fp,
                                   content_length=len(data),
                                   content_md5=content_md5,
                                   content_type=content_type,
                                   content_sha256=content_sha256,
                                   user_metadata=user_metadata,
                                   storage_class=storage_class,
                                   user_headers=user_headers,
                                   encryption=encryption,
                                   customer_key=customer_key,
                                   customer_key_md5=customer_key_md5,
                                   progress_callback = progress_callback,
                                   traffic_limit=traffic_limit,
                                   object_tagging=object_tagging,
                                   cond_read_write=cond_read_write,
                                   config=config)
        finally:
            if fp is not None:
                fp.close()

    @required(bucket=(bytes, str), key=(bytes, str), file_name=(bytes, str))
    def put_object_from_file(self, bucket, key, file_name,
                             content_length=None,
                             content_md5=None,
                             content_type=None,
                             content_sha256=None,
                             user_metadata=None,
                             storage_class=None,
                             user_headers=None,
                             encryption=None,
                             customer_key=None,
                             customer_key_md5=None,
                             progress_callback=None,
                             traffic_limit=None,
                             object_tagging=None,
                             cond_read_write=None,
                             config=None,
                             ):

        """
        Put object and put content of file to the object

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None

        :type file_name: string
        :param file_name: None

        :type options: dict
        :param options: None
        :return:
            **HttpResponse Class**
        """
        key = compat.convert_to_bytes(key)
        fp = open(file_name, 'rb')
        try:
            if content_length is None:
                fp.seek(0, os.SEEK_END)
                content_length = fp.tell()
                fp.seek(0)
            if content_md5 is None:
                recv_buf_size = self._get_config_parameter(config, 'recv_buf_size')
                content_md5 = utils.get_md5_from_fp(fp, length=content_length,
                                                    buf_size=recv_buf_size)
            if content_type is None:
                content_type = utils.guess_content_type_by_file_name(file_name)
            return self.put_object(bucket, key, fp,
                                   content_length=content_length,
                                   content_md5=content_md5,
                                   content_type=content_type,
                                   content_sha256=content_sha256,
                                   user_metadata=user_metadata,
                                   storage_class=storage_class,
                                   user_headers=user_headers,
                                   encryption=encryption,
                                   customer_key=customer_key,
                                   customer_key_md5=customer_key_md5,
                                   progress_callback=progress_callback,
                                   traffic_limit=traffic_limit,
                                   object_tagging=object_tagging,
                                   cond_read_write=cond_read_write,
                                   config=config)
        finally:
            fp.close()

    @required(source_bucket_name=(bytes, str),
              source_key=(bytes, str),
              target_bucket_name=(bytes, str),
              target_key=(bytes, str))
    def copy_object(self,
                    source_bucket_name, source_key,
                    target_bucket_name, target_key,
                    etag=None,
                    content_type=None,
                    user_metadata=None,
                    storage_class=None,
                    user_headers=None,
                    copy_object_user_headers=None,
                    traffic_limit=None,
                    object_tagging=None,
                    source_version_id=None,
                    config=None):
        """
        Copy one object to another object

        :type source_bucket: string
        :param source_bucket: None

        :type source_key: string
        :param source_key: None

        :type target_bucket: string
        :param target_bucket: None

        :type target_key: string
        :param target_key: None
        :return:
            **HttpResponse Class**
        """
        source_key = compat.convert_to_bytes(source_key)
        target_key = compat.convert_to_bytes(target_key)
        headers = self._prepare_object_headers(
            content_type=content_type,
            user_metadata=user_metadata,
            storage_class=storage_class,
            user_headers=user_headers,
            traffic_limit=traffic_limit,
            object_tagging=object_tagging)
        
        merge_source_key = utils.normalize_string(
            b'/%s/%s' % (
                compat.convert_to_bytes(source_bucket_name), 
                source_key), False)
        if source_version_id is not None:
            merge_source_key = merge_source_key + b'?versionId=%s' % compat.convert_to_bytes(source_version_id)
        headers[http_headers.BCE_COPY_SOURCE] = merge_source_key
        if etag is not None:
            headers[http_headers.BCE_COPY_SOURCE_IF_MATCH] = etag
        if user_metadata is not None or content_type is not None:
            headers[http_headers.BCE_COPY_METADATA_DIRECTIVE] = b'replace'
        else:
            headers[http_headers.BCE_COPY_METADATA_DIRECTIVE] = b'copy'

        if copy_object_user_headers is not None:
            try:
                headers = BosClient._get_user_header(headers, copy_object_user_headers, True)
            except Exception as e:
                raise e

        return self._send_request(
            http_methods.PUT,
            target_bucket_name,
            target_key,
            headers=headers,
            config=config,
            body_parser=bos_handler.parse_copy_object_response)

    @required(bucket_name=(bytes, str), key=(bytes, str))
    def delete_object(self, bucket_name, key, version_id=None, config=None):
        """
        Delete Object

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None
        :return:
            **HttpResponse Class**
        """
        query_params = {}
        if version_id is not None:
            version_id = compat.convert_to_bytes(version_id)
            query_params={b'versionId': version_id}
        key = compat.convert_to_bytes(key)
        return self._send_request(http_methods.DELETE, bucket_name, key,
            params=query_params, config=config)

    @required(bucket_name=(bytes, str), key_list=list)
    def delete_multiple_objects(self, bucket_name, key_list, config=None):
        """
        Delete Multiple Objects

        :type bucket: string
        :param bucket: None

        :type key_list: string list
        :param key_list: None
        :return:
            **HttpResponse Class**
        """
        key_list_json = [{'key': compat.convert_to_string(k)} for k in key_list]
        return self._send_request(http_methods.POST, 
                                  bucket_name, 
                                  body=json.dumps({'objects': key_list_json}),
                                  params={b'delete': b''},
                                  config=config)

    @required(source_bucket=(bytes, str),
              target_bucket=(bytes, str),
              target_prefix=(bytes, str))
    def put_bucket_logging(self,
                          source_bucket,
                          target_bucket,
                          target_prefix=None,
                          config=None):
        """
        Put Bucket Logging

        :type source_bucket: string
        :param source_bucket: None

        :type target_bucket: string
        :param target_bucket: None
        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.PUT, 
                                  source_bucket, 
                                  params={b'logging': b''}, 
                                  body=json.dumps({'targetBucket': target_bucket, 
                                                  'targetPrefix': target_prefix}),
                                  config=config)

    @required(bucket_name=(bytes, str))
    def get_bucket_logging(self, bucket_name, config=None):
        """
        Get Bucket Logging

        :type bucket_name: string
        :param bucket_name: None
        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.GET,
                                  bucket_name,
                                  params={b'logging': b''},
                                  config=config)

    @required(bucket_name=(bytes, str))
    def delete_bucket_logging(self, bucket_name, config=None):
        """
        Delete Bucket Logging

        :type bucket_name: string
        :param bucket_name: None
        :return:
            **HttpResponse Class**
        """
        return self._send_request(http_methods.DELETE,
                                  bucket_name,
                                  params={b'logging': b''},
                                  config=config)

    @required(bucket_name=(bytes, str), key=(bytes, str))
    def initiate_multipart_upload(self,
                                  bucket_name,
                                  key,
                                  content_type=None,
                                  storage_class=None,
                                  user_headers=None,
                                  config=None):
        """
        Initialize multi_upload_file.

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None
        :return:
            **HttpResponse**
        """
        key = compat.convert_to_bytes(key)
        headers = {}
        if storage_class is not None:
            headers[http_headers.BOS_STORAGE_CLASS] = storage_class

        if content_type is not None:
            headers[http_headers.CONTENT_TYPE] = utils.convert_to_standard_string(content_type)
        else:
            headers[http_headers.CONTENT_TYPE] = http_content_types.OCTET_STREAM

        if user_headers is not None:
            try:
                headers = BosClient._get_user_header(headers, user_headers, False)
            except Exception as e:
                raise e

        return self._send_request(
            http_methods.POST,
            bucket_name,
            key,
            headers=headers,
            params={b'uploads': b''},
            config=config)

    @required(bucket_name=(bytes, str),
              key=(bytes, str),
              upload_id=(bytes, str),
              part_number=int,
              part_size=compat.integer_types,
              part_fp=object)
    def upload_part(self, bucket_name, key, upload_id,
                    part_number, part_size, part_fp, part_md5=None,
                    progress_callback=None, traffic_limit=None, config=None):
        """
        Upload a part.

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None

        :type upload_id: string
        :param upload_id: None

        :type part_number: int
        :param part_number: None

        :type part_size: int or long
        :param part_size: None

        :type part_fp: file pointer
        :param part_fp: not None

        :type part_md5: str
        :param part_md5: None

        :type config: dict
        :param config: None

        :return:
               **HttpResponse**
        """
        key = compat.convert_to_bytes(key)
        if part_number < bos.MIN_PART_NUMBER or part_number > bos.MAX_PART_NUMBER:
            raise ValueError('Invalid part_number %d. The valid range is from %d to %d.' % (
                part_number, bos.MIN_PART_NUMBER, bos.MAX_PART_NUMBER))

        if part_size > bos.MAX_PUT_OBJECT_LENGTH:
            raise ValueError('Single part length should be less than %d. '
                             % bos.MAX_PUT_OBJECT_LENGTH)

        headers = {http_headers.CONTENT_LENGTH: part_size,
                   http_headers.CONTENT_TYPE: http_content_types.OCTET_STREAM}
        if part_md5 is not None:
            headers[http_headers.CONTENT_MD5] = part_md5

        if progress_callback:
            part_fp = utils.make_progress_adapter(part_fp, progress_callback, part_size)
        
        if traffic_limit is not None:
            headers[http_headers.BOS_TRAFFIC_LIMIT] = traffic_limit
        return self._send_request(
            http_methods.PUT,
            bucket_name,
            key,
            body=part_fp,
            headers=headers,
            params={b'partNumber': part_number, b'uploadId': upload_id},
            config=config)

    @required(source_bucket_name=(bytes, str),
              source_key=(bytes, str),
              target_bucket_name=(bytes, str),
              target_key=(bytes, str),
              upload_id=(bytes, str),
              part_number=int,
              part_size=compat.integer_types,
              offset=compat.integer_types)
    def upload_part_copy(self, 
                         source_bucket_name, source_key, 
                         target_bucket_name, target_key,
                         upload_id, part_number, part_size, offset,
                         etag=None,
                         content_type=None,
                         user_metadata=None,
                         traffic_limit=None,
                         config=None):
        """
        Copy part.

        :type source_bucket_name: string
        :param source_bucket_name: None

        :type source_key: string
        :param source_key: None

        :type target_bucket_name: string
        :param target_bucket_name: None

        :type target_key: string
        :param target_key: None

        :type upload_id: string
        :param upload_id: None

        :return:
            **HttpResponse**
        """
        source_key = compat.convert_to_bytes(source_key)
        target_key = compat.convert_to_bytes(target_key)
        headers = self._prepare_object_headers(
                         content_type=content_type,
                         user_metadata=user_metadata,
                         traffic_limit=traffic_limit)
        headers[http_headers.BCE_COPY_SOURCE] = utils.normalize_string(
                         b"/%s/%s" % (compat.convert_to_bytes(source_bucket_name),
                         source_key), False)
        range = b"""bytes=%d-%d""" % (offset, offset + part_size - 1)
        headers[http_headers.BCE_COPY_SOURCE_RANGE] = range
        if etag is not None:
            headers[http_headers.BCE_COPY_SOURCE_IF_MATCH] = etag

        return self._send_request(
            http_methods.PUT,
            target_bucket_name,
            target_key,
            headers=headers,
            params={b'partNumber': part_number, b'uploadId': upload_id},
            config=config)

    @required(bucket_name=(bytes, str),
              key=(bytes, str),
              upload_id=(bytes, str),
              part_number=int,
              part_size=compat.integer_types,
              file_name=(bytes, str),
              offset=compat.integer_types)
    def upload_part_from_file(self, bucket_name, key, upload_id,
                              part_number, part_size, file_name, offset, part_md5=None,
                              progress_callback=None, traffic_limit=None, config=None):
        """

        :param bucket_name:
        :param key:
        :param upload_id:
        :param part_number:
        :param part_size:
        :param file_name:
        :param offset:
        :param part_md5:
        :param config:
        :return:
        """
        key = compat.convert_to_bytes(key)
        f = open(file_name, 'rb')
        try:
            f.seek(offset)
            return self.upload_part(bucket_name, key, upload_id, part_number, part_size, f,
                                    part_md5=part_md5, progress_callback=progress_callback,
                                    traffic_limit=traffic_limit, config=config)
        finally:
            f.close()

    @required(bucket_name=(bytes, str),
              key=(bytes, str),
              upload_id=(bytes, str),
              part_list=list)
    def complete_multipart_upload(self, bucket_name, key,
                                  upload_id, part_list,
                                  user_headers=None,
                                  user_metadata=None,
                                  cond_read_write=None,
                                  config=None):
        """
        After finish all the task, complete multi_upload_file.

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None

        :type upload_id: string
        :param upload_id: None

        :type part_list: list
        :param part_list: None

        :return:
            **HttpResponse**
        """
        key = compat.convert_to_bytes(key)
        headers = self._prepare_object_headers(
            content_type=http_content_types.JSON,
            user_metadata=user_metadata,
            user_headers=user_headers)

        if cond_read_write is not None:
            headers = self._get_cond_read_write_headers(http_methods.POST, headers, cond_read_write)

        return self._send_request(
            http_methods.POST,
            bucket_name,
            key,
            body=json.dumps({'parts': part_list}),
            headers=headers,
            params={b'uploadId': upload_id})

    @required(bucket_name=(bytes, str), key=(bytes, str), upload_id=(bytes, str))
    def abort_multipart_upload(self, bucket_name, key, upload_id, config=None):
        """
        Abort upload a part which is being uploading.

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None

        :type upload_id: string
        :param upload_id: None
        :return:
            **HttpResponse**
        """
        key = compat.convert_to_bytes(key)
        return self._send_request(http_methods.DELETE, bucket_name, key,
                                  params={b'uploadId': upload_id})

    @required(bucket_name=(bytes, str), key=(bytes, str), upload_id=(bytes, str))
    def list_parts(self, bucket_name, key, upload_id,
                   max_parts=None, part_number_marker=None,
                   config=None):
        """
        List all the parts that have been upload success.

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None

        :type upload_id: string
        :param upload_id: None

        :type max_parts: int
        :param max_parts: None

        :type part_number_marker: string
        :param part_number_marker: None
        :return:
            **_ListPartsResponse Class**
        """
        key = compat.convert_to_bytes(key)
        params = {b'uploadId': upload_id}
        if max_parts is not None:
            params[b'maxParts'] = max_parts
        if part_number_marker is not None:
            params[b'partNumberMarker'] = part_number_marker

        return self._send_request(http_methods.GET, bucket_name, key, params=params, config=config)

    @required(bucket_name=(bytes, str), key=(bytes, str), upload_id=(bytes, str))
    def list_all_parts(self, bucket_name, key, upload_id, config=None):
        """

        :param bucket_name:
        :param key:
        :param upload_id:
        :param config:
        :return:
        """
        key = compat.convert_to_bytes(key)
        part_number_marker = None
        while True:
            response = self.list_parts(bucket_name, key, upload_id,
                                      part_number_marker=part_number_marker, config=config)
            for item in response.parts:
                yield item
            if not response.is_truncated:
                break
            part_number_marker = response.next_part_number_marker

    @required(bucket_name=(bytes, str))
    def list_multipart_uploads(self, bucket_name, max_uploads=None, key_marker=None,
                               prefix=None, delimiter=None,
                               config=None):
        """
        List all Multipart upload task which haven't been ended.(Completed Init_MultiPartUpload
        but not completed Complete_MultiPartUpload or Abort_MultiPartUpload)

        :type bucket: string
        :param bucket: None

        :type delimiter: string
        :param delimiter: None

        :type max_uploads: int
        :param max_uploads: <=1000

        :type key_marker: string
        :param key_marker: None

        :type prefix: string
        :param prefix: None

        :type upload_id_marker: string
        :param upload_id_marker:
        :return:
            **_ListMultipartUploadResponse Class**
        """
        params = {b'uploads': b''}
        if delimiter is not None:
            params[b'delimiter'] = delimiter
        if max_uploads is not None:
            params[b'maxUploads'] = max_uploads
        if key_marker is not None:
            params[b'keyMarker'] = key_marker
        if prefix is not None:
            params[b'prefix'] = prefix

        return self._send_request(http_methods.GET, bucket_name, params=params, config=config)

    @required(bucket_name=(bytes, str))
    def list_all_multipart_uploads(self, bucket_name, prefix=None, delimiter=None, config=None):
        """

        :param bucket_name:
        :param prefix:
        :param delimiter:
        :param config:
        :return:
        """
        key_marker = None
        while True:
            response = self.list_multipart_uploads(bucket_name,
                                                   key_marker=key_marker,
                                                   prefix=prefix,
                                                   delimiter=delimiter,
                                                   config=config)
            for item in response.uploads:
                yield item
            if not response.is_truncated:
                break
            if response.next_key_marker is not None:
                key_marker = response.next_key_marker
            elif len(response.uploads) != 0:
                key_marker = response.uploads[-1].key
            else:
                break

    def _upload_task(self, bucket_name, object_key, upload_id,
        part_number, part_size, file_name, offset, part_list, uploadTaskHandle,
        progress_callback=None, traffic_limit=None):
        if uploadTaskHandle.is_cancel():
            _logger.debug("upload task canceled with partNumber={}!".format(part_number))
            return
        try:
            response = self.upload_part_from_file(bucket_name, object_key, upload_id,
                part_number, part_size, file_name, offset, progress_callback=progress_callback
                , traffic_limit=traffic_limit)
            part_list.append({
                "partNumber": part_number,
                "eTag": response.metadata.etag
            })
            _logger.debug("upload task success with partNumber={}!".format(part_number))
        except Exception as e:
            _logger.debug("upload task failed with partNumber={}!".format(part_number))
            raise e
            #_logger.debug(e)

    @required(bucket_name=(bytes, str), key=(bytes, str), file_name=(bytes, str))
    def put_super_obejct_from_file(self, bucket_name, key, file_name, chunk_size=5,
            thread_num=None,
            uploadTaskHandle=None,
            content_type=None,
            storage_class=None,
            user_headers=None,
            progress_callback=None,
            traffic_limit=None,
            cond_read_write=None,
            config=None):
        """
        Multipart Upload file to bos

        param chunk_size: part size , default part size is 5MB
        """
        # check params
        if chunk_size > 5 * 1024 or chunk_size <= 0:
           raise BceClientError("chunk size is valid, it should be more than 0 and not nore than 5120!")
        left_size = os.path.getsize(file_name)
        # if file size more than 48.8TB, reject
        if left_size > 50000 * 1024 * 1024 * 1024:
           raise BceClientError("File size must not be more than 48.8TB!")
        if thread_num is None or thread_num <= 1:
           thread_num = multiprocessing.cpu_count()
        part_size = chunk_size * 1024 * 1024
        total_part = left_size // part_size
        if left_size % part_size != 0:
            total_part += 1
        if uploadTaskHandle is None:
            uploadTaskHandle = UploadTaskHandle()
        # initial
        upload_id = self.initiate_multipart_upload(bucket_name, key,
                content_type=content_type,
                storage_class=storage_class,
                user_headers=user_headers).upload_id

        executor = ThreadPoolExecutor(thread_num)
        all_tasks = []
        offset = 0
        part_number = 1
        part_list = []

        while left_size > 0:
            if left_size < part_size:
                part_size = left_size
            temp_task= executor.submit(self._upload_task, bucket_name, key, upload_id, part_number, part_size,
                file_name, offset, part_list, uploadTaskHandle, progress_callback, traffic_limit)
            all_tasks.append(temp_task)
            left_size -= part_size
            offset += part_size
            part_number += 1
        # wait all upload task to exit
        wait(all_tasks, return_when=ALL_COMPLETED)
        if uploadTaskHandle.is_cancel():
            _logger.debug("putting super object is canceled!")
            self.abort_multipart_upload(bucket_name, key, upload_id = upload_id)
            return False
        elif len(part_list) != total_part:
            _logger.debug("putting super object failed!")
            self.abort_multipart_upload(bucket_name, key, upload_id = upload_id)
            return False
        # sort
        part_list.sort(key=lambda x: x["partNumber"])
        # complete_multipart_upload
        self.complete_multipart_upload(bucket_name, key, upload_id, part_list, cond_read_write)
        return True

    @required(bucket_name=(bytes, str), key=(bytes, str), acl=(list, dict))
    def set_object_acl(self, bucket_name, key, acl, config=None):
        """
        Set Access Control Level of object

        :type bucket: string
        :param bucket: None

        :type acl: list of grant
        :param acl: None
        :return:
            **HttpResponse Class**
        """
        key = compat.convert_to_bytes(key)
        self._send_request(http_methods.PUT,
                           bucket_name,
                           key,
                           body=json.dumps({'accessControlList': acl},
                                           default=BosClient._dump_acl_object),
                           headers={http_headers.CONTENT_TYPE: http_content_types.JSON},
                           params={b'acl': b''},
                           config=config)

    @required(bucket_name=(bytes, str), key=(bytes, str))
    def set_object_canned_acl(self, bucket_name, key,
            canned_acl=None,
            grant_read=None,
            grant_full_control=None,
            config=None):
        """

        :type bucket_name: string
        :param bucket_name: None

        :type key: string
        :param key: None

        :type canned_acl: string
        :param canned_acl: for header 'x-bce-acl', it's value only is
        canned_acl.PRIVATE or canned_acl.PRIVATE_READ

        :type grant_read: string
        :param grant_read: Object id of getting READ right permission.
        for exapmle,grant_read = 'id="6c47...4c94",id="8c42...4c94"'

        :type grant_full_control: string
        :param grant_full_control: Object id of getting READ right permission.
        for exapmle,grant_full_control = 'id="6c47...4c94",id="8c42...4c94"'

        :param config:
        :return:
            **HttpResponse Class**
        """
        key = compat.convert_to_bytes(key)
        headers = None
        num_args = 0
        if canned_acl is not None:
            headers =  {http_headers.BCE_ACL: compat.convert_to_bytes(canned_acl)}
            num_args += 1
        if grant_read is not None:
            headers = {http_headers.BOS_GRANT_READ: compat.convert_to_bytes(grant_read)}
            num_args += 1
        if grant_full_control is not None:
            headers = {http_headers.BOS_GRANT_FULL_CONTROL: compat.convert_to_bytes(grant_full_control)}
            num_args += 1

        if num_args == 0:
            raise ValueError("donn't give any object canned acl arguments!")
        elif num_args >= 2:
            raise ValueError("cann't get more than one object canned acl arguments!")

        self._send_request(http_methods.PUT,
                           bucket_name,
                           key,
                           headers=headers,
                           params={b'acl': b''},
                           config=config)

    @required(bucket_name=(bytes, str), key=(bytes, str))
    def get_object_acl(self, bucket_name, key, config=None):
        """
        Get Access Control Level of object

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None

        :return:
            **HttpResponse Class**
        """
        key = compat.convert_to_bytes(key)
        return self._send_request(
                http_methods.GET,
                bucket_name,
                key,
                params={b'acl': b''},
                config=config)

    @required(bucket_name=(bytes, str), key=(bytes, str))
    def delete_object_acl(self, bucket_name, key, config=None):
        """
        Get Access Control Level of  object

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: None

        :return:
            **HttpResponse Class**
        """
        key = compat.convert_to_bytes(key)
        return self._send_request(
                http_methods.DELETE,
                bucket_name,
                key,
                params={b'acl': b''},
                config=config)

    @required(bucket_name=(bytes, str), key=(bytes, str), url=(bytes, str))
    def fetch_object(self, bucket_name, key, url,
            fetch_mode=None,
            storage_class=None,
            config=None):
        """
        fetch object with given url and save to Baidu object storage

        :type bucket: string
        :param bucket: None

        :type key: string
        :param key: object name to be saved

        :type url:string
        :param url: url of resource to be fetched

        :type fetch_mode:string
        :param fetch_mode: fetch mode for get resource, valid value only is
        'sync' and 'async'

        :return:
            **HttpResponse Class**
        """
        key = compat.convert_to_bytes(key)
        headers = {}
        headers[http_headers.BOS_FETCH_SOURCE] = compat.convert_to_bytes(url)
        if fetch_mode is not None:
            headers[http_headers.BOS_FETCH_MODE] = fetch_mode
        if storage_class is not None:
            headers[http_headers.BOS_STORAGE_CLASS] = storage_class
        return self._send_request(
                http_methods.POST,
                bucket_name,
                key,
                headers=headers,
                params={b'fetch': b''},
                config=config)

    @required(bucket_name=(bytes, str), target_key=(bytes, str), symlink=(bytes, str), forbid_overwrite=(bool))
    def put_object_symlink(self, bucket_name, target_key, symlink, forbid_overwrite=None, 
            user_metadata=None, storage_class=None, target_bucket=None, content_type=None, config=None):
        """
        put object symlink

        :type bucket: string
        :param bucket: None

        :type key: string
        :type key: object name

        :type symlink: string
        :type symlink_key: symlink name

        :return:
            **HttpResponse Class**
        """
        target_key = compat.convert_to_bytes(target_key)
        symlink = compat.convert_to_bytes(symlink)
        if content_type is None:
                content_type = utils.guess_content_type_by_file_name(symlink)
        headers = self._prepare_object_headers(user_metadata=user_metadata,
                content_type=content_type,
                storage_class=storage_class)
        headers[http_headers.BOS_SYMLINK_TARGET] = target_key
        if forbid_overwrite is not None:
            if forbid_overwrite:
                headers[http_headers.BOS_FORBID_OVERWRITE] = b'true'
            else:
                headers[http_headers.BOS_FORBID_OVERWRITE] = b'false'
        if target_bucket is not None:
            headers[http_headers.BOS_SYMLINK_BUCKET] = compat.convert_to_bytes(target_bucket)
        return self._send_request(http_methods.PUT,
                           bucket_name,
                           symlink,
                           headers=headers,
                           params={b'symlink': b''},
                           config=config)


    @required(bucket_name=(bytes, str), symlink=(bytes, str))
    def get_object_symlink(self, bucket_name, symlink, config=None):
        """
        Get symlink info

        :type bucket: string
        :param bucket: None

        :type symlink: string
        :param symlink: symlink

        :return:
            **HttpResponse Class**
        """
        key = compat.convert_to_bytes(symlink)
        return self._send_request(
                http_methods.GET,
                bucket_name,
                key,
                params={b'symlink': b''},
                config=config)

    @required(bucket_name=(bytes, str), key=(bytes, str), select_object_args=(dict, ))
    def select_object(self, bucket_name, key, select_object_args, headers=None, config=None):
        """

        :type bucket_name: string
        :param bucket_name: bucket name

        :type key: string
        :param key: object name

        :type select_object_args: dict
        :param select_object_args: requesta parameters for select object api

        :param config:
        :return:
        """
        key = compat.convert_to_bytes(key)
        headers = headers or {}
        if "inputSerialization" in select_object_args and "json" in select_object_args["inputSerialization"]:
            select_type = b"json"
        elif "inputSerialization" in select_object_args and "csv" in select_object_args["inputSerialization"]:
            select_type = b"csv"
        else:
            select_type = b"parquet"
        select_response = SelectResponse()
        self._send_request(
            http_methods.POST,
            bucket_name,
            key,
            body=json.dumps({'selectRequest': select_object_args}, default=BosClient._dump_acl_object),
            headers=headers,
            params={b'select': b'', b'type': select_type},
            config=config,
            body_parser=lambda http_response, response: BosClient._parse_select_message(
                http_response, response, select_response)
            )
        return select_response

    def get_user_quota(self, config=None):
        """
        get user quota

        :param config:
        :return:
        """
        return self._send_request(
            http_methods.GET, params={b'userQuota': b''}, config=config,)
    
    @required(max_bucket_count=(int), max_capacity_mega_bytes=(int))
    def put_user_quota(self, max_bucket_count, max_capacity_mega_bytes, config=None):
        """
        put user quota

        :type max_bucket_count: int
        :param max_bucket_count: max bucket count

        :type max_capacity_mega_bytes: long
        :param max_capacity_mega_bytes: max capacity mega bytes

        :param config:
        :return:
        """
        return self._send_request(
            http_methods.PUT,
            body=json.dumps({'maxBucketCount': max_bucket_count, 
                            'maxCapacityMegaBytes': max_capacity_mega_bytes}),
            params={b'userQuota': b''}, config=config)

    def delete_user_quota(self, config=None):
        """
        delete user quota

        :param config:
        :return:
        """
        return self._send_request(
            http_methods.DELETE, params={b'userQuota': b''}, config=config)


    @required(bucket_name=(bytes, str))
    def get_notification(self, bucket_name, config=None):
        """
        get notification

        :type bucket_name: string
        :param bucket_name: bucket name

        :param config:
        :return:
        """
        return self._send_request(
            http_methods.GET, bucket_name=bucket_name, 
            params={b'notification': b''}, config=config,)
    

    @required(bucket_name=(bytes, str), notifications=(list, ))
    def put_notification(self, bucket_name, notifications, config=None):
        """
        put user quota

        :type bucket_name: string
        :param bucket_name: bucket

        :type notifications: list of dict
        :param notifications: notifacation param

        :param config:
        :return:
        """
        return self._send_request(
            http_methods.PUT, bucket_name=bucket_name,
            body=json.dumps({'notifications': notifications}),
            params={b'notification': b''}, config=config,)
    @required(bucket_name=(bytes, str))
    def delete_notification(self, bucket_name, config=None):
        """
        delete notification

        :type bucket_name: string
        :param bucket_name: bucket name

        :param config:
        :return:
        """
        return self._send_request(
            http_methods.DELETE, bucket_name=bucket_name, params={b'notification': b''},
            config=config,)

    @required(bucket_name=(bytes, str), mirror_args=(list, ))
    def put_bucket_mirroring(self, bucket_name, mirror_args, config=None):
        """
        put bucket mirroring

        :type bucket_name: string
        :param bucket_name: bucket name

        :param mirror_args: mirror conf
        :return:
        """
        return self._send_request(
            http_methods.PUT,
            bucket_name=bucket_name,
            body=json.dumps({'bucketMirroringConfiguration': mirror_args}, default=BosClient._dump_acl_object),
            params={b'mirroring': b''},
            config=config,
            )

    @required(bucket_name=(bytes, str))
    def get_bucket_mirroring(self, bucket_name, config=None):
        """
        get bucket mirroring

        :type bucket_name: string
        :param bucket_name: bucket name

        :return:
        """
        return self._send_request(
            http_methods.GET,
            bucket_name=bucket_name,
            params={b'mirroring': b''},
            config=config,
            )
    @required(bucket_name=(bytes, str))
    def delete_bucket_mirroring(self, bucket_name, config=None):
        """
        delete bucket mirroring

        :type bucket_name: string
        :param bucket_name: bucket name

        :return:
        """
        return self._send_request(
            http_methods.DELETE,
            bucket_name=bucket_name,
            params={b'mirroring': b''},
            config=config,
            )

    def put_object_tagging(self, bucket_name, key, obj_tag_args, config=None):
        """
        put object tagging

        :type bucket_name: string
        :param bucket_name: bucket name

        :type key: string
        :param key: object name

        :type obj_tag_args: dict
        :param obj_tag_args: object tagging args

        :return:
        """
        return self._send_request(
            http_methods.PUT,
            bucket_name=bucket_name,
            key=key,
            body=json.dumps(obj_tag_args, default=BosClient._dump_acl_object),
            params={b'tagging': b''},
            config=config,)

    def put_object_tagging_canned(self, bucket_name, key, tag_header, config=None):
        """
        put object tagging

        :type bucket_name: string
        :param bucket_name: bucket name

        :type key: string
        :param key: object name

        :type obj_tag_args: dict
        :param obj_tag_args: object tagging args

        :return:
        """
        headers = {}
        headers[http_headers.BOS_TAGGING] = compat.convert_to_bytes(tag_header)
        return self._send_request(
            http_methods.PUT,
            bucket_name=bucket_name,
            key=key,
            headers=headers,
            params={b'tagging': b''},
            config=config,)

    def get_object_tagging(self, bucket_name, key, config=None):
        """
        put object tagging

        :type bucket_name: string
        :param bucket_name: bucket name

        :type key: string
        :param key: object name

        :return:
        """
        return self._send_request(
            http_methods.GET,
            bucket_name=bucket_name,
            key=key,
            params={b'tagging': b''},
            config=config,)
    
    def put_bucket_versioning(self, bucket_name, status, config=None):
        """
        put bucket versioning

        :type bucket_name: string
        :param bucket_name: bucket name

        :type status: string
        :param key: version status:disable/enabled/suspended

        :return:
        """
        return self._send_request(
            http_methods.PUT,
            bucket_name=bucket_name,
            body=json.dumps({'status': status}, default=BosClient._dump_acl_object),
            params={b'versioning': b''},
            config=config,)
    
    def get_bucket_versioning(self, bucket_name, config=None):
        """
        get bucket versioning

        :type bucket_name: string
        :param bucket_name: bucket name

        :return:
        """
        return self._send_request(
            http_methods.GET,
            bucket_name=bucket_name,
            params={b'versioning': b''},
            config=config,)
    
    @required(bucket_name=(bytes, str))        
    def list_objects_versions(self, bucket_name,
                     max_keys=1000, prefix=None, marker=None, version_marker=None, 
                     delimiter=None, config=None):
        """
        Get Object Information of bucket

        :type bucket: string
        :param bucket: None

        :type delimiter: string
        :param delimiter: None

        :type marker: string
        :param marker: None

        :type max_keys: int
        :param max_keys: value <= 1000

        :type prefix: string
        :param prefix: None

        :return:
            **_ListObjectsResponse Class**
        """
        params = {b'versions': b''}
        if max_keys is not None:
            params[b'maxKeys'] = max_keys
        if prefix is not None:
            params[b'prefix'] = prefix
        if marker is not None:
            params[b'marker'] = marker
        if delimiter is not None:
            params[b'delimiter'] = delimiter
        if version_marker is not None:
            params[b'versionIdMarker'] = version_marker

        return self._send_request(http_methods.GET, bucket_name, params=params, config=config)
    
    @required(bucket_name=(bytes, str), cors_configuration=list)
    def init_bucket_object_lock(self, 
                        bucket_name,
                        retention_days,
                        config=None):
        """
        init bucket object lock
        :type bucket: string
        :param bucket: None

        :type retention_days: int
        :param retention_days: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.POST,
                                  bucket_name,
                                  params={b'objectlock': b''},
                                  body=json.dumps({'retentionDays': retention_days}),
                                  config=config)

    @required(bucket_name=(bytes, str))
    def get_bucket_object_lock(self, bucket_name, config=None):
        """
        get bucket object lock

        :type bucket: string
        :param bucket: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.GET,
                                  bucket_name,
                                  params={b'objectlock': b''},
                                  config=config)

    @required(bucket_name=(bytes, str))
    def delete_bucket_object_lock(self, bucket_name, config=None):
        """
        Delete Bucket Object Lock

        :type bucket: string
        :param bucket: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.DELETE,
                                  bucket_name,
                                  params={b'objectlock': b''},
                                  config=config)

    @required(bucket_name=(bytes, str), cors_configuration=list)
    def complete_bucket_object_lock(self, 
                        bucket_name,
                        config=None):
        """
        complete bucket object lock
        :type bucket: string
        :param bucket: None

        :type retention_days: int
        :param retention_days: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.POST,
                                  bucket_name,
                                  params={b'completeobjectlock': b''},
                                  config=config)
    
    @required(bucket_name=(bytes, str), cors_configuration=list)
    def extend_bucket_object_lock(self, 
                        bucket_name,
                        extend_retent_days,
                        config=None):
        """
        extend bucket object lock
        :type bucket: string
        :param bucket: None

        :type extend_retent_days: int
        :param extend_retent_days: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.POST,
                                  bucket_name,
                                  params={b'extendobjectlock': b''},
                                  body=json.dumps({'extendRetentionDays': extend_retent_days}),
                                  config=config)

    @required(bucket_name=(bytes, str), cors_configuration=list)
    def get_bucket_quota(self, 
                        bucket_name,
                        config=None):
        """
        get bucket quota
        :type bucket: string
        :param bucket: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.GET,
                                  bucket_name,
                                  params={b'quota': b''},
                                  config=config)
    
    @required(bucket_name=(bytes, str), cors_configuration=list)
    def put_bucket_quota(self, 
                        bucket_name,
                        quota_conf,
                        config=None):
        """
        put quota conf of bucket
        :type bucket: string
        :param bucket: None

        :type retention_days: int
        :param retention_days: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.PUT,
                                  bucket_name,
                                  body=json.dumps(quota_conf, default=BosClient._dump_acl_object),
                                  params={b'quota': b''},
                                  config=config)
    
    @required(bucket_name=(bytes, str), cors_configuration=list)
    def delete_bucket_quota(self, 
                        bucket_name,
                        config=None):
        """
        get bucket quota
        :type bucket: string
        :param bucket: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.DELETE,
                                  bucket_name,
                                  params={b'quota': b''},
                                  config=config)
    
    @required(bucket_name=(bytes, str), cors_configuration=list)
    def get_bucket_tagging(self, 
                        bucket_name,
                        config=None):
        """
        get bucket tagging
        :type bucket: string
        :param bucket: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.GET,
                                  bucket_name,
                                  params={b'tagging': b''},
                                  config=config)
    
    @required(bucket_name=(bytes, str), cors_configuration=list)
    def put_bucket_tagging(self, 
                        bucket_name,
                        tag_conf,
                        config=None):
        """
        put tagging conf of bucket
        :type bucket: string
        :param bucket: None

        :type retention_days: int
        :param retention_days: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.PUT,
                                  bucket_name,
                                  body=json.dumps(tag_conf, default=BosClient._dump_acl_object),
                                  params={b'tagging': b''},
                                  config=config)
    
    @required(bucket_name=(bytes, str), cors_configuration=list)
    def delete_bucket_tagging(self, 
                        bucket_name,
                        config=None):
        """
        delete bucket tagging
        :type bucket: string
        :param bucket: None

        :return:**Http Response**
        """
        return self._send_request(http_methods.DELETE,
                                  bucket_name,
                                  params={b'tagging': b''},
                                  config=config)

    @staticmethod
    def _prepare_object_headers(
            content_length=None,
            content_md5=None,
            content_type=None,
            content_sha256=None,
            etag=None,
            user_metadata=None,
            storage_class=None,
            user_headers=None,
            encryption=None,
            customer_key=None,
            customer_key_md5=None,
            traffic_limit=None,
            object_tagging=None,):
        headers = {}

        if content_length is not None:
            if content_length and content_length < 0:
                raise ValueError('content_length should not be negative.')
            headers[http_headers.CONTENT_LENGTH] = compat.convert_to_bytes(content_length)

        if content_md5 is not None:
            headers[http_headers.CONTENT_MD5] = utils.convert_to_standard_string(content_md5)

        if content_type is not None:
            headers[http_headers.CONTENT_TYPE] = utils.convert_to_standard_string(content_type)
        else:
            headers[http_headers.CONTENT_TYPE] = http_content_types.OCTET_STREAM

        if content_sha256 is not None:
            headers[http_headers.BCE_CONTENT_SHA256] = content_sha256

        if etag is not None:
            headers[http_headers.ETAG] = b'"%s"' % utils.convert_to_standard_string(etag)

        if user_metadata is not None:
            meta_size = 0
            meta_data_set = set()
            if not isinstance(user_metadata, dict):
                raise TypeError('user_metadata should be of type dict.')
            for k, v in iteritems(user_metadata):
                meta_data_set.add(k.lower())
                k = utils.convert_to_standard_string(k)
                v = utils.convert_to_standard_string(v)
                normalized_key = http_headers.BCE_USER_METADATA_PREFIX + k
                headers[normalized_key] = v
                meta_size += len(normalized_key)
                meta_size += len(v)
            if meta_size > bos.MAX_USER_METADATA_SIZE:
                raise ValueError(
                    'Metadata size should not be greater than %d.' % bos.MAX_USER_METADATA_SIZE)
            if len(meta_data_set) != len(user_metadata):
                raise ValueError('user_metadata has duplicate elements.')

        if storage_class is not None:
            headers[http_headers.BOS_STORAGE_CLASS] = storage_class

        if encryption is not None:
            headers[http_headers.BOS_SERVER_SIDE_ENCRYPTION] = utils.convert_to_standard_string(encryption)

        if customer_key is not None:
            headers[http_headers.BOS_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY] = \
                utils.convert_to_standard_string(customer_key)

        if customer_key_md5 is not None:
            headers[http_headers.BOS_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5] = \
                utils.convert_to_standard_string(customer_key_md5)

        if user_headers is not None:
            try:
                headers = BosClient._get_user_header(headers, user_headers, False)
            except Exception as e:
                raise e
        
        if traffic_limit is not None:
            headers[http_headers.BOS_TRAFFIC_LIMIT] = traffic_limit

        if object_tagging is not None:
            headers[http_headers.BOS_TAGGING] = compat.convert_to_bytes(object_tagging)

        return headers


    @staticmethod
    def _get_cond_read_write_headers(http_method, headers, cond_read_write):
        """
        get if condition headers
        :type http_method: string
        :param http_method: GET, HEAD, PUT, POST

        :type cond_read_write: string
        :param cond_read_write: put_object, complete_multipart_upload, get_object, get_object_meta_data

        :return: headers
        """
        cond_read_write_set = http_headers.BOS_COND_READ_WRITE_HEADERS
        
        if http_method == http_methods.GET or http_method == http_methods.HEAD:
            cond_read_write_set = cond_read_write_set.union(set(
                [http_headers.BOS_IF_MODIFIED_SINCE,
                 http_headers.BOS_IF_UNMODIFIED_SINCE,
                 http_headers.BOS_IF_MATCH,
                 http_headers.BOS_IF_NONE_MATCH]))
        
        for k, v in iteritems(cond_read_write):
            k = utils.convert_to_standard_string(k)
            if k in cond_read_write_set:
                headers[k] = v
            else:
                raise ValueError('%s is not valid in %s' % (k, http_method))
        return headers


    @staticmethod
    def _get_user_header(headers, user_headers, is_copy=False):
        if not isinstance(user_headers, dict):
            raise TypeError('user_headers should be of type dict.')
        
        bos_headers = http_headers.BOS_BASE_ALLOW_HEADERS

        if not is_copy:
            user_headers_set = bos_headers.union(set([http_headers.CACHE_CONTROL,
                                    http_headers.CONTENT_ENCODING,
                                    http_headers.CONTENT_DISPOSITION,
                                    http_headers.EXPIRES,
                                    http_headers.BOS_PROCESS]))
        else:
            user_headers_set = bos_headers.union(set([http_headers.BCE_COPY_SOURCE_IF_NONE_MATCH,
                                    http_headers.BCE_COPY_SOURCE_IF_UNMODIFIED_SINCE,
                                    http_headers.BCE_COPY_SOURCE_IF_MODIFIED_SINCE]))

        for k, v in iteritems(user_headers):
            k = utils.convert_to_standard_string(k)
            if k != http_headers.BOS_OBJECT_EXPIRES:
                v = utils.convert_to_standard_string(v)
            if k in user_headers_set:
                headers[k] = v
        return headers

    def _get_config_parameter(self, config, attr):
        result = None
        if config is not None:
            result = getattr(config, attr)
        if result is not None:
            return result
        return getattr(self.config, attr)


    @staticmethod
    def _get_path(config, bucket_name=None, key=None, use_backup_endpoint=False):
        host = config.endpoint
        if use_backup_endpoint:
            host = config.backup_endpoint
        endpoint_protocol, host_name, endpoint_port = \
            utils.parse_host_port(config.endpoint, config.protocol)
        if config.cname_enabled or utils.is_cname_like_host(host_name) or utils.is_custom_host(host_name, bucket_name):
            return utils.append_uri(bos.URL_PREFIX, key)

        return utils.append_uri(bos.URL_PREFIX, bucket_name, key)
    

    def _merge_config(self, config, bucket_name):
        new_config = copy.copy(self.config)
        if config is not None:
            new_config.merge_non_none_values(config)
        
        endpoint = self._change_user_endpoint(new_config, bucket_name)
        new_config.endpoint = endpoint
        return new_config
        

    def _change_user_endpoint(self, config, bucket_name):
        endpoint_protocol, user_host_name, endpoint_port = \
            utils.parse_host_port(config.endpoint, config.protocol)
        user_endpoint_split = compat.convert_to_bytes(user_host_name).split(b'.')
        user_endpoint = config.endpoint
        is_bos_path_style_host = utils.is_bos_suffixed_host(user_host_name) and len(user_endpoint_split) == 3
        # 1. check ipv4 or path style
        if utils.check_ipv4(user_host_name):
            return config.endpoint
        
        if  config.path_style_enable:
            return config.endpoint
        
        # 2. check cname domain
        if config.cname_enabled or utils.is_cname_like_host(user_host_name):
            # cname domain
            if is_bos_path_style_host:
                raise ValueError(
                    'endpoint is not cname domain, please set cname_enabled=False')
            else:
                return config.endpoint
        
        # default use virtual-hosted endpoint
        if bucket_name is not None:
            if is_bos_path_style_host:
                # split http head
                if user_endpoint.startswith(HTTP_PROTOCOL_HEAD):
                    http_head_split = user_endpoint.split(b'//') 
                    if len(http_head_split) < 2:
                        return config.endpoint
                    bucket_endpoint = http_head_split[0] + b'//' + compat.convert_to_bytes(bucket_name) +\
                        b'.' + http_head_split[1]
                    return compat.convert_to_bytes(bucket_endpoint)
                else:
                    return compat.convert_to_bytes(bucket_name)+b'.'+\
                        compat.convert_to_bytes(user_endpoint)
        
        # check virtual-hosted endpoint's bucket_name is not query bucket_name
        if len(user_endpoint_split) == 4 and bucket_name is not None:
            if  user_endpoint_split[0] != compat.convert_to_bytes(bucket_name):
                raise ValueError('your endpoint\'s bucket_name is not equal your query bucket_name!')

        return config.endpoint

    @staticmethod
    def _need_retry_for_bos(config, error):
        
        if not isinstance(error, BceServerError):
            return False

        # if you need add more retry condition, please add it here
        if error.status_code == http.client.FORBIDDEN:
            _logger.debug('BOS retry condition matched: 403 Forbidden')
            return True

        return False

    @staticmethod
    def _need_retry_backup_endpoint(error):
        # always retry on IOError
        if isinstance(error, IOError):
            return True

        # Only retry on a subset of service exceptions
        if isinstance(error, BceServerError):
            if error.status_code == http.client.INTERNAL_SERVER_ERROR:
                return True
            if error.status_code == http.client.SERVICE_UNAVAILABLE:
                return True
            if error.code == BceServerError.REQUEST_EXPIRED:
                return True
            if error.status_code == http.client.FORBIDDEN:
                return True
        return False

    def _send_request(
            self, http_method, bucket_name=None, key=None,
            body=None, headers=None, params=None,
            config=None,
            body_parser=None):
        config = self._merge_config(config, bucket_name)

        path = BosClient._get_path(config, bucket_name, key)
        if body_parser is None:
            body_parser = handler.parse_json

        if config.security_token is not None:
            headers = headers or {}
            headers[http_headers.STS_SECURITY_TOKEN] = config.security_token

        last_exception = None
        e = None
        try:
            return bce_http_client.send_request(
                config, bce_v1_signer.sign, [handler.parse_error, body_parser],
                http_method, path, body, headers, params)
        except BceHttpClientError as ex:
            last_exception = ex
            e = ex

        # retry backup endpoint
        if e is not None and config.backup_endpoint is not None and BosClient._need_retry_backup_endpoint(e):
            try:
                _logger.debug(b'Retry for backup endpoint error code: %d.', e.status_code)
                path = BosClient._get_path(config, bucket_name, key, True)
                return bce_http_client.send_request(
                    config, bce_v1_signer.sign, [handler.parse_error, body_parser],
                    http_method, path, body, headers, params, True)
            except BceHttpClientError as ex:
                last_exception = ex
                e = ex

        # retry for bos error
        if e is not None and BosClient._need_retry_for_bos(config, e.last_error):
            try:
                _logger.debug(b'Retry for BOS error code: %d.', e.status_code)
                return bce_http_client.send_request(
                    config, bce_v1_signer.sign, [handler.parse_error, body_parser],
                    http_method, path, body, headers, params)
            except BceHttpClientError as ex:
                last_exception = ex
                e = ex

        if last_exception is None:
            raise
        raise last_exception

class SelectMessage(object):
    """
    returned message from select object api
    """
    def set_record_message(self, headers, payload, crc):
        """
        Initialize for record message
        """
        self.type = "Records"
        self.headers = headers
        self.payload = payload
        self.crc = crc

    def set_cont_message(self, headers, bytes_scanned, bytes_returned, crc):
        """
        Initialize for continue message
        """
        self.type = "Cont"
        self.headers = headers
        self.bytes_scanned = bytes_scanned
        self.bytes_returned = bytes_returned
        self.crc = crc

    def set_end_message(self, headers, crc):
        """
        Initialize for end message
        """
        self.type = "End"
        self.headers = headers
        self.crc = crc

    def __str__(self):
        if self.type == "Records":
            return '{}\n{}'.format(self.headers, self.payload)
        elif self.type == "Cont":
            return '{}\nbytes_scanned/bytes_returned={}/{}'.format(self.headers, self.bytes_scanned,
                                                               self.bytes_returned)
        else:
            return '{}'.format(self.headers)

class SelectResponse(object):
    """
    deal with message of select object api
    """
    def __init__(self):
        self.finish = False

    def init_from_http_response(self, http_response, response):
        """
        get HttpResponse and BceResponse
        """
        self.http_response = http_response
        self.response = response

    def result(self):
        """
        generator for SelectMessage
        """
        f  = self.http_response
        try:
            while not self.finish:
                prelude = f.read(8)
                if not prelude:
                    raise StopIteration
                    return
                total_len = struct.unpack('>I', prelude[0:4])[0]
                headers_len = struct.unpack('>I', prelude[4:8])[0]
                headers = f.read(headers_len)
                headers_map = self._parse_select_headers(headers)
                msg = SelectMessage()
                if headers_map['message-type'] == 'Records':
                    payload_len = total_len - headers_len - 12
                    payload = f.read(payload_len)
                    crc = struct.unpack('>I', f.read(4))[0]
                    msg.set_record_message(headers_map, compat.convert_to_string(payload), crc)
                    yield msg
                elif headers_map['message-type'] == 'Cont':
                    bytes_scanned = f.read(8)
                    bytes_returned = f.read(8)
                    crc = struct.unpack('>I', f.read(4))[0]
                    bytes_scanned = struct.unpack('>Q', bytes_scanned)[0]
                    bytes_returned = struct.unpack('>Q', bytes_returned)[0]
                    msg.set_cont_message(headers_map, bytes_scanned, bytes_returned, crc)
                    yield msg
                elif headers_map['message-type'] == 'End':
                    crc = struct.unpack('>I', f.read(4))[0]
                    if headers_map["error-code"] != "success":
                        raise BceServerError(headers_map['error-message'], code=headers_map['error-code'],
                                request_id=self.response.metadata.bce_request_id)
                        return
                    msg.set_end_message(headers_map, crc)
                    self.finish = True
                    yield msg
            return
        finally:
            self.http_response.close()

    @staticmethod
    def _parse_select_headers(headers):
        """
        parse SELECT headers
        :param headers: <str>
        :return: <dict>
        """
        hm = {}
        index = 0
        while index < len(headers):
            # headers key length
            key_len = struct.unpack('B', headers[index: index + 1])[0]
            index += 1
            # headers key
            key = headers[index: index + key_len]
            index += key_len
            # headers value length
            value_len = struct.unpack('>H', headers[index: index + 2])[0]
            index += 2
            # headers value
            value = headers[index: index + value_len]
            index += value_len
            hm[compat.convert_to_string(key)] = compat.convert_to_string(value)
        return hm