You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@libcloud.apache.org by an...@apache.org on 2016/11/14 23:50:53 UTC
[06/56] [abbrv] libcloud git commit: Removed sdist
http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/storage/base.py
----------------------------------------------------------------------
diff --git a/apache-libcloud-1.0.0rc2/libcloud/storage/base.py b/apache-libcloud-1.0.0rc2/libcloud/storage/base.py
deleted file mode 100644
index f13dd0a..0000000
--- a/apache-libcloud-1.0.0rc2/libcloud/storage/base.py
+++ /dev/null
@@ -1,831 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You 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.
-
-"""
-Provides base classes for working with storage
-"""
-
-# Backward compatibility for Python 2.5
-from __future__ import with_statement
-
-import os.path # pylint: disable-msg=W0404
-import hashlib
-from os.path import join as pjoin
-
-from libcloud.utils.py3 import httplib
-from libcloud.utils.py3 import next
-from libcloud.utils.py3 import b
-
-import libcloud.utils.files
-from libcloud.common.types import LibcloudError
-from libcloud.common.base import ConnectionUserAndKey, BaseDriver
-from libcloud.storage.types import ObjectDoesNotExistError
-
-__all__ = [
- 'Object',
- 'Container',
- 'StorageDriver',
-
- 'CHUNK_SIZE',
- 'DEFAULT_CONTENT_TYPE'
-]
-
-CHUNK_SIZE = 8096
-
-# Default Content-Type which is sent when uploading an object if one is not
-# supplied and can't be detected when using non-strict mode.
-DEFAULT_CONTENT_TYPE = 'application/octet-stream'
-
-
-class Object(object):
- """
- Represents an object (BLOB).
- """
-
- def __init__(self, name, size, hash, extra, meta_data, container,
- driver):
- """
- :param name: Object name (must be unique per container).
- :type name: ``str``
-
- :param size: Object size in bytes.
- :type size: ``int``
-
- :param hash: Object hash.
- :type hash: ``str``
-
- :param container: Object container.
- :type container: :class:`Container`
-
- :param extra: Extra attributes.
- :type extra: ``dict``
-
- :param meta_data: Optional object meta data.
- :type meta_data: ``dict``
-
- :param driver: StorageDriver instance.
- :type driver: :class:`StorageDriver`
- """
-
- self.name = name
- self.size = size
- self.hash = hash
- self.container = container
- self.extra = extra or {}
- self.meta_data = meta_data or {}
- self.driver = driver
-
- def get_cdn_url(self):
- return self.driver.get_object_cdn_url(obj=self)
-
- def enable_cdn(self, **kwargs):
- return self.driver.enable_object_cdn(obj=self, **kwargs)
-
- def download(self, destination_path, overwrite_existing=False,
- delete_on_failure=True):
- return self.driver.download_object(self, destination_path,
- overwrite_existing,
- delete_on_failure)
-
- def as_stream(self, chunk_size=None):
- return self.driver.download_object_as_stream(self, chunk_size)
-
- def delete(self):
- return self.driver.delete_object(self)
-
- def __repr__(self):
- return ('<Object: name=%s, size=%s, hash=%s, provider=%s ...>' %
- (self.name, self.size, self.hash, self.driver.name))
-
-
-class Container(object):
- """
- Represents a container (bucket) which can hold multiple objects.
- """
-
- def __init__(self, name, extra, driver):
- """
- :param name: Container name (must be unique).
- :type name: ``str``
-
- :param extra: Extra attributes.
- :type extra: ``dict``
-
- :param driver: StorageDriver instance.
- :type driver: :class:`StorageDriver`
- """
-
- self.name = name
- self.extra = extra or {}
- self.driver = driver
-
- def iterate_objects(self):
- return self.driver.iterate_container_objects(container=self)
-
- def list_objects(self):
- return self.driver.list_container_objects(container=self)
-
- def get_cdn_url(self):
- return self.driver.get_container_cdn_url(container=self)
-
- def enable_cdn(self, **kwargs):
- return self.driver.enable_container_cdn(container=self, **kwargs)
-
- def get_object(self, object_name):
- return self.driver.get_object(container_name=self.name,
- object_name=object_name)
-
- def upload_object(self, file_path, object_name, extra=None, **kwargs):
- return self.driver.upload_object(
- file_path, self, object_name, extra=extra, **kwargs)
-
- def upload_object_via_stream(self, iterator, object_name, extra=None,
- **kwargs):
- return self.driver.upload_object_via_stream(
- iterator, self, object_name, extra=extra, **kwargs)
-
- def download_object(self, obj, destination_path, overwrite_existing=False,
- delete_on_failure=True):
- return self.driver.download_object(
- obj, destination_path, overwrite_existing=overwrite_existing,
- delete_on_failure=delete_on_failure)
-
- def download_object_as_stream(self, obj, chunk_size=None):
- return self.driver.download_object_as_stream(obj, chunk_size)
-
- def delete_object(self, obj):
- return self.driver.delete_object(obj)
-
- def delete(self):
- return self.driver.delete_container(self)
-
- def __repr__(self):
- return ('<Container: name=%s, provider=%s>'
- % (self.name, self.driver.name))
-
-
-class StorageDriver(BaseDriver):
- """
- A base StorageDriver to derive from.
- """
-
- connectionCls = ConnectionUserAndKey
- name = None
- hash_type = 'md5'
- supports_chunked_encoding = False
-
- # When strict mode is used, exception will be thrown if no content type is
- # provided and none can be detected when uploading an object
- strict_mode = False
-
- def iterate_containers(self):
- """
- Return a generator of containers for the given account
-
- :return: A generator of Container instances.
- :rtype: ``generator`` of :class:`Container`
- """
- raise NotImplementedError(
- 'iterate_containers not implemented for this driver')
-
- def list_containers(self):
- """
- Return a list of containers.
-
- :return: A list of Container instances.
- :rtype: ``list`` of :class:`Container`
- """
- return list(self.iterate_containers())
-
- def iterate_container_objects(self, container):
- """
- Return a generator of objects for the given container.
-
- :param container: Container instance
- :type container: :class:`Container`
-
- :return: A generator of Object instances.
- :rtype: ``generator`` of :class:`Object`
- """
- raise NotImplementedError(
- 'iterate_container_objects not implemented for this driver')
-
- def list_container_objects(self, container):
- """
- Return a list of objects for the given container.
-
- :param container: Container instance.
- :type container: :class:`Container`
-
- :return: A list of Object instances.
- :rtype: ``list`` of :class:`Object`
- """
- return list(self.iterate_container_objects(container))
-
- def get_container(self, container_name):
- """
- Return a container instance.
-
- :param container_name: Container name.
- :type container_name: ``str``
-
- :return: :class:`Container` instance.
- :rtype: :class:`Container`
- """
- raise NotImplementedError(
- 'get_object not implemented for this driver')
-
- def get_container_cdn_url(self, container):
- """
- Return a container CDN URL.
-
- :param container: Container instance
- :type container: :class:`Container`
-
- :return: A CDN URL for this container.
- :rtype: ``str``
- """
- raise NotImplementedError(
- 'get_container_cdn_url not implemented for this driver')
-
- def get_object(self, container_name, object_name):
- """
- Return an object instance.
-
- :param container_name: Container name.
- :type container_name: ``str``
-
- :param object_name: Object name.
- :type object_name: ``str``
-
- :return: :class:`Object` instance.
- :rtype: :class:`Object`
- """
- raise NotImplementedError(
- 'get_object not implemented for this driver')
-
- def get_object_cdn_url(self, obj):
- """
- Return an object CDN URL.
-
- :param obj: Object instance
- :type obj: :class:`Object`
-
- :return: A CDN URL for this object.
- :rtype: ``str``
- """
- raise NotImplementedError(
- 'get_object_cdn_url not implemented for this driver')
-
- def enable_container_cdn(self, container):
- """
- Enable container CDN.
-
- :param container: Container instance
- :type container: :class:`Container`
-
- :rtype: ``bool``
- """
- raise NotImplementedError(
- 'enable_container_cdn not implemented for this driver')
-
- def enable_object_cdn(self, obj):
- """
- Enable object CDN.
-
- :param obj: Object instance
- :type obj: :class:`Object`
-
- :rtype: ``bool``
- """
- raise NotImplementedError(
- 'enable_object_cdn not implemented for this driver')
-
- def download_object(self, obj, destination_path, overwrite_existing=False,
- delete_on_failure=True):
- """
- Download an object to the specified destination path.
-
- :param obj: Object instance.
- :type obj: :class:`Object`
-
- :param destination_path: Full path to a file or a directory where the
- incoming file will be saved.
- :type destination_path: ``str``
-
- :param overwrite_existing: True to overwrite an existing file,
- defaults to False.
- :type overwrite_existing: ``bool``
-
- :param delete_on_failure: True to delete a partially downloaded file if
- the download was not successful (hash
- mismatch / file size).
- :type delete_on_failure: ``bool``
-
- :return: True if an object has been successfully downloaded, False
- otherwise.
- :rtype: ``bool``
- """
- raise NotImplementedError(
- 'download_object not implemented for this driver')
-
- def download_object_as_stream(self, obj, chunk_size=None):
- """
- Return a generator which yields object data.
-
- :param obj: Object instance
- :type obj: :class:`Object`
-
- :param chunk_size: Optional chunk size (in bytes).
- :type chunk_size: ``int``
- """
- raise NotImplementedError(
- 'download_object_as_stream not implemented for this driver')
-
- def upload_object(self, file_path, container, object_name, extra=None,
- verify_hash=True, headers=None):
- """
- Upload an object currently located on a disk.
-
- :param file_path: Path to the object on disk.
- :type file_path: ``str``
-
- :param container: Destination container.
- :type container: :class:`Container`
-
- :param object_name: Object name.
- :type object_name: ``str``
-
- :param verify_hash: Verify hash
- :type verify_hash: ``bool``
-
- :param extra: Extra attributes (driver specific). (optional)
- :type extra: ``dict``
-
- :param headers: (optional) Additional request headers,
- such as CORS headers. For example:
- headers = {'Access-Control-Allow-Origin': 'http://mozilla.com'}
- :type headers: ``dict``
-
- :rtype: :class:`Object`
- """
- raise NotImplementedError(
- 'upload_object not implemented for this driver')
-
- def upload_object_via_stream(self, iterator, container,
- object_name,
- extra=None,
- headers=None):
- """
- Upload an object using an iterator.
-
- If a provider supports it, chunked transfer encoding is used and you
- don't need to know in advance the amount of data to be uploaded.
-
- Otherwise if a provider doesn't support it, iterator will be exhausted
- so a total size for data to be uploaded can be determined.
-
- Note: Exhausting the iterator means that the whole data must be
- buffered in memory which might result in memory exhausting when
- uploading a very large object.
-
- If a file is located on a disk you are advised to use upload_object
- function which uses fs.stat function to determine the file size and it
- doesn't need to buffer whole object in the memory.
-
- :param iterator: An object which implements the iterator interface.
- :type iterator: :class:`object`
-
- :param container: Destination container.
- :type container: :class:`Container`
-
- :param object_name: Object name.
- :type object_name: ``str``
-
- :param extra: (optional) Extra attributes (driver specific). Note:
- This dictionary must contain a 'content_type' key which represents
- a content type of the stored object.
- :type extra: ``dict``
-
- :param headers: (optional) Additional request headers,
- such as CORS headers. For example:
- headers = {'Access-Control-Allow-Origin': 'http://mozilla.com'}
- :type headers: ``dict``
-
- :rtype: ``object``
- """
- raise NotImplementedError(
- 'upload_object_via_stream not implemented for this driver')
-
- def delete_object(self, obj):
- """
- Delete an object.
-
- :param obj: Object instance.
- :type obj: :class:`Object`
-
- :return: ``bool`` True on success.
- :rtype: ``bool``
- """
- raise NotImplementedError(
- 'delete_object not implemented for this driver')
-
- def create_container(self, container_name):
- """
- Create a new container.
-
- :param container_name: Container name.
- :type container_name: ``str``
-
- :return: Container instance on success.
- :rtype: :class:`Container`
- """
- raise NotImplementedError(
- 'create_container not implemented for this driver')
-
- def delete_container(self, container):
- """
- Delete a container.
-
- :param container: Container instance
- :type container: :class:`Container`
-
- :return: ``True`` on success, ``False`` otherwise.
- :rtype: ``bool``
- """
- raise NotImplementedError(
- 'delete_container not implemented for this driver')
-
- def _get_object(self, obj, callback, callback_kwargs, response,
- success_status_code=None):
- """
- Call passed callback and start transfer of the object'
-
- :param obj: Object instance.
- :type obj: :class:`Object`
-
- :param callback: Function which is called with the passed
- callback_kwargs
- :type callback: :class:`function`
-
- :param callback_kwargs: Keyword arguments which are passed to the
- callback.
- :type callback_kwargs: ``dict``
-
- :param response: Response instance.
- :type response: :class:`Response`
-
- :param success_status_code: Status code which represents a successful
- transfer (defaults to httplib.OK)
- :type success_status_code: ``int``
-
- :return: ``True`` on success, ``False`` otherwise.
- :rtype: ``bool``
- """
- success_status_code = success_status_code or httplib.OK
-
- if response.status == success_status_code:
- return callback(**callback_kwargs)
- elif response.status == httplib.NOT_FOUND:
- raise ObjectDoesNotExistError(object_name=obj.name,
- value='', driver=self)
-
- raise LibcloudError(value='Unexpected status code: %s' %
- (response.status),
- driver=self)
-
- def _save_object(self, response, obj, destination_path,
- overwrite_existing=False, delete_on_failure=True,
- chunk_size=None):
- """
- Save object to the provided path.
-
- :param response: RawResponse instance.
- :type response: :class:`RawResponse`
-
- :param obj: Object instance.
- :type obj: :class:`Object`
-
- :param destination_path: Destination directory.
- :type destination_path: ``str``
-
- :param delete_on_failure: True to delete partially downloaded object if
- the download fails.
- :type delete_on_failure: ``bool``
-
- :param overwrite_existing: True to overwrite a local path if it already
- exists.
- :type overwrite_existing: ``bool``
-
- :param chunk_size: Optional chunk size
- (defaults to ``libcloud.storage.base.CHUNK_SIZE``, 8kb)
- :type chunk_size: ``int``
-
- :return: ``True`` on success, ``False`` otherwise.
- :rtype: ``bool``
- """
-
- chunk_size = chunk_size or CHUNK_SIZE
-
- base_name = os.path.basename(destination_path)
-
- if not base_name and not os.path.exists(destination_path):
- raise LibcloudError(
- value='Path %s does not exist' % (destination_path),
- driver=self)
-
- if not base_name:
- file_path = pjoin(destination_path, obj.name)
- else:
- file_path = destination_path
-
- if os.path.exists(file_path) and not overwrite_existing:
- raise LibcloudError(
- value='File %s already exists, but ' % (file_path) +
- 'overwrite_existing=False',
- driver=self)
-
- stream = libcloud.utils.files.read_in_chunks(response, chunk_size)
-
- try:
- data_read = next(stream)
- except StopIteration:
- # Empty response?
- return False
-
- bytes_transferred = 0
-
- with open(file_path, 'wb') as file_handle:
- while len(data_read) > 0:
- file_handle.write(b(data_read))
- bytes_transferred += len(data_read)
-
- try:
- data_read = next(stream)
- except StopIteration:
- data_read = ''
-
- if int(obj.size) != int(bytes_transferred):
- # Transfer failed, support retry?
- if delete_on_failure:
- try:
- os.unlink(file_path)
- except Exception:
- pass
-
- return False
-
- return True
-
- def _upload_object(self, object_name, content_type, upload_func,
- upload_func_kwargs, request_path, request_method='PUT',
- headers=None, file_path=None, iterator=None):
- """
- Helper function for setting common request headers and calling the
- passed in callback which uploads an object.
- """
- headers = headers or {}
-
- if file_path and not os.path.exists(file_path):
- raise OSError('File %s does not exist' % (file_path))
-
- if iterator is not None and not hasattr(iterator, 'next') and not \
- hasattr(iterator, '__next__'):
- raise AttributeError('iterator object must implement next() ' +
- 'method.')
-
- if not content_type:
- if file_path:
- name = file_path
- else:
- name = object_name
- content_type, _ = libcloud.utils.files.guess_file_mime_type(name)
-
- if not content_type:
- if self.strict_mode:
- raise AttributeError('File content-type could not be '
- 'guessed and no content_type value '
- 'is provided')
- else:
- # Fallback to a content-type
- content_type = DEFAULT_CONTENT_TYPE
-
- file_size = None
-
- if iterator:
- if self.supports_chunked_encoding:
- headers['Transfer-Encoding'] = 'chunked'
- upload_func_kwargs['chunked'] = True
- else:
- # Chunked transfer encoding is not supported. Need to buffer
- # all the data in memory so we can determine file size.
- iterator = libcloud.utils.files.read_in_chunks(
- iterator=iterator)
- data = libcloud.utils.files.exhaust_iterator(iterator=iterator)
-
- file_size = len(data)
- upload_func_kwargs['data'] = data
- else:
- file_size = os.path.getsize(file_path)
- upload_func_kwargs['chunked'] = False
-
- if file_size is not None and 'Content-Length' not in headers:
- headers['Content-Length'] = file_size
-
- headers['Content-Type'] = content_type
- response = self.connection.request(request_path,
- method=request_method, data=None,
- headers=headers, raw=True)
-
- upload_func_kwargs['response'] = response
- success, data_hash, bytes_transferred = upload_func(
- **upload_func_kwargs)
-
- if not success:
- raise LibcloudError(
- value='Object upload failed, Perhaps a timeout?', driver=self)
-
- result_dict = {'response': response, 'data_hash': data_hash,
- 'bytes_transferred': bytes_transferred}
- return result_dict
-
- def _upload_data(self, response, data, calculate_hash=True):
- """
- Upload data stored in a string.
-
- :param response: RawResponse object.
- :type response: :class:`RawResponse`
-
- :param data: Data to upload.
- :type data: ``str``
-
- :param calculate_hash: True to calculate hash of the transferred data.
- (defaults to True).
- :type calculate_hash: ``bool``
-
- :return: First item is a boolean indicator of success, second
- one is the uploaded data MD5 hash and the third one
- is the number of transferred bytes.
- :rtype: ``tuple``
- """
- bytes_transferred = 0
- data_hash = None
-
- if calculate_hash:
- data_hash = self._get_hash_function()
- data_hash.update(b(data))
-
- try:
- response.connection.connection.send(b(data))
- except Exception:
- # TODO: let this exception propagate
- # Timeout, etc.
- return False, None, bytes_transferred
-
- bytes_transferred = len(data)
-
- if calculate_hash:
- data_hash = data_hash.hexdigest()
-
- return True, data_hash, bytes_transferred
-
- def _stream_data(self, response, iterator, chunked=False,
- calculate_hash=True, chunk_size=None, data=None):
- """
- Stream a data over an http connection.
-
- :param response: RawResponse object.
- :type response: :class:`RawResponse`
-
- :param response: An object which implements an iterator interface
- or a File like object with read method.
- :type iterator: :class:`object`
-
- :param chunked: True if the chunked transfer encoding should be used
- (defaults to False).
- :type chunked: ``bool``
-
- :param calculate_hash: True to calculate hash of the transferred data.
- (defaults to True).
- :type calculate_hash: ``bool``
-
- :param chunk_size: Optional chunk size (defaults to ``CHUNK_SIZE``)
- :type chunk_size: ``int``
-
- :rtype: ``tuple``
- :return: First item is a boolean indicator of success, second
- one is the uploaded data MD5 hash and the third one
- is the number of transferred bytes.
- """
-
- chunk_size = chunk_size or CHUNK_SIZE
-
- data_hash = None
- if calculate_hash:
- data_hash = self._get_hash_function()
-
- generator = libcloud.utils.files.read_in_chunks(iterator, chunk_size,
- fill_size=True)
-
- bytes_transferred = 0
- try:
- chunk = next(generator)
- except StopIteration:
- # Special case when StopIteration is thrown on the first iteration
- # create a 0-byte long object
- chunk = ''
- if chunked:
- response.connection.connection.send(b('%X\r\n' %
- (len(chunk))))
- response.connection.connection.send(chunk)
- response.connection.connection.send(b('\r\n'))
- response.connection.connection.send(b('0\r\n\r\n'))
- else:
- response.connection.connection.send(chunk)
- return True, data_hash.hexdigest(), bytes_transferred
-
- while len(chunk) > 0:
- try:
- if chunked:
- response.connection.connection.send(b('%X\r\n' %
- (len(chunk))))
- response.connection.connection.send(b(chunk))
- response.connection.connection.send(b('\r\n'))
- else:
- response.connection.connection.send(b(chunk))
- except Exception:
- # TODO: let this exception propagate
- # Timeout, etc.
- return False, None, bytes_transferred
-
- bytes_transferred += len(chunk)
- if calculate_hash:
- data_hash.update(b(chunk))
-
- try:
- chunk = next(generator)
- except StopIteration:
- chunk = ''
-
- if chunked:
- response.connection.connection.send(b('0\r\n\r\n'))
-
- if calculate_hash:
- data_hash = data_hash.hexdigest()
-
- return True, data_hash, bytes_transferred
-
- def _upload_file(self, response, file_path, chunked=False,
- calculate_hash=True):
- """
- Upload a file to the server.
-
- :type response: :class:`RawResponse`
- :param response: RawResponse object.
-
- :type file_path: ``str``
- :param file_path: Path to a local file.
-
- :type iterator: :class:`object`
- :param response: An object which implements an iterator interface (File
- object, etc.)
-
- :rtype: ``tuple``
- :return: First item is a boolean indicator of success, second
- one is the uploaded data MD5 hash and the third one
- is the number of transferred bytes.
- """
- with open(file_path, 'rb') as file_handle:
- success, data_hash, bytes_transferred = (
- self._stream_data(
- response=response,
- iterator=iter(file_handle),
- chunked=chunked,
- calculate_hash=calculate_hash))
-
- return success, data_hash, bytes_transferred
-
- def _get_hash_function(self):
- """
- Return instantiated hash function for the hash type supported by
- the provider.
- """
- try:
- func = getattr(hashlib, self.hash_type)()
- except AttributeError:
- raise RuntimeError('Invalid or unsupported hash type: %s' %
- (self.hash_type))
-
- return func
http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/__init__.py
----------------------------------------------------------------------
diff --git a/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/__init__.py b/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/__init__.py
deleted file mode 100644
index fe8b04f..0000000
--- a/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/__init__.py
+++ /dev/null
@@ -1,23 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You 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.
-
-"""
-Drivers for working with different providers
-"""
-
-__all__ = [
- 'dummy',
- 'cloudfiles'
-]
http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/atmos.py
----------------------------------------------------------------------
diff --git a/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/atmos.py b/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/atmos.py
deleted file mode 100644
index c52be03..0000000
--- a/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/atmos.py
+++ /dev/null
@@ -1,472 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You 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.
-
-import sys
-import base64
-import hashlib
-import hmac
-import time
-
-from libcloud.utils.py3 import PY3
-from libcloud.utils.py3 import b
-from libcloud.utils.py3 import httplib
-from libcloud.utils.py3 import next
-from libcloud.utils.py3 import urlparse
-from libcloud.utils.py3 import urlencode
-from libcloud.utils.py3 import urlquote
-from libcloud.utils.py3 import urlunquote
-
-if PY3:
- from io import FileIO as file
-
-from libcloud.utils.files import read_in_chunks, guess_file_mime_type
-from libcloud.common.base import ConnectionUserAndKey, XmlResponse
-from libcloud.common.types import LibcloudError
-
-from libcloud.storage.base import Object, Container, StorageDriver, CHUNK_SIZE
-from libcloud.storage.types import ContainerAlreadyExistsError, \
- ContainerDoesNotExistError, ContainerIsNotEmptyError, \
- ObjectDoesNotExistError
-
-
-def collapse(s):
- return ' '.join([x for x in s.split(' ') if x])
-
-
-class AtmosError(LibcloudError):
- def __init__(self, code, message, driver=None):
- super(AtmosError, self).__init__(value=message, driver=driver)
- self.code = code
-
-
-class AtmosResponse(XmlResponse):
- def success(self):
- return self.status in (httplib.OK, httplib.CREATED, httplib.NO_CONTENT,
- httplib.PARTIAL_CONTENT)
-
- def parse_error(self):
- tree = self.parse_body()
-
- if tree is None:
- return None
-
- code = int(tree.find('Code').text)
- message = tree.find('Message').text
- raise AtmosError(code=code, message=message,
- driver=self.connection.driver)
-
-
-class AtmosConnection(ConnectionUserAndKey):
- responseCls = AtmosResponse
-
- def add_default_headers(self, headers):
- headers['x-emc-uid'] = self.user_id
- headers['Date'] = time.strftime('%a, %d %b %Y %H:%M:%S GMT',
- time.gmtime())
- headers['x-emc-date'] = headers['Date']
-
- if 'Content-Type' not in headers:
- headers['Content-Type'] = 'application/octet-stream'
- if 'Accept' not in headers:
- headers['Accept'] = '*/*'
-
- return headers
-
- def pre_connect_hook(self, params, headers):
- headers['x-emc-signature'] = self._calculate_signature(params, headers)
-
- return params, headers
-
- def _calculate_signature(self, params, headers):
- pathstring = urlunquote(self.action)
- if pathstring.startswith(self.driver.path):
- pathstring = pathstring[len(self.driver.path):]
- if params:
- if type(params) is dict:
- params = list(params.items())
- pathstring += '?' + urlencode(params)
- pathstring = pathstring.lower()
-
- xhdrs = [(k, v) for k, v in list(headers.items()) if
- k.startswith('x-emc-')]
- xhdrs.sort(key=lambda x: x[0])
-
- signature = [
- self.method,
- headers.get('Content-Type', ''),
- headers.get('Range', ''),
- headers.get('Date', ''),
- pathstring,
- ]
- signature.extend([k + ':' + collapse(v) for k, v in xhdrs])
- signature = '\n'.join(signature)
- key = base64.b64decode(self.key)
- signature = hmac.new(b(key), b(signature), hashlib.sha1).digest()
- return base64.b64encode(b(signature)).decode('utf-8')
-
-
-class AtmosDriver(StorageDriver):
- connectionCls = AtmosConnection
-
- host = None
- path = None
- api_name = 'atmos'
- supports_chunked_encoding = True
- website = 'http://atmosonline.com/'
- name = 'atmos'
-
- DEFAULT_CDN_TTL = 60 * 60 * 24 * 7 # 1 week
-
- def __init__(self, key, secret=None, secure=True, host=None, port=None):
- host = host or self.host
- super(AtmosDriver, self).__init__(key, secret, secure, host, port)
-
- def iterate_containers(self):
- result = self.connection.request(self._namespace_path(''))
- entries = self._list_objects(result.object, object_type='directory')
- for entry in entries:
- extra = {
- 'object_id': entry['id']
- }
- yield Container(entry['name'], extra, self)
-
- def get_container(self, container_name):
- path = self._namespace_path(container_name) + '/?metadata/system'
- try:
- result = self.connection.request(path)
- except AtmosError:
- e = sys.exc_info()[1]
- if e.code != 1003:
- raise
- raise ContainerDoesNotExistError(e, self, container_name)
- meta = self._emc_meta(result)
- extra = {
- 'object_id': meta['objectid']
- }
- return Container(container_name, extra, self)
-
- def create_container(self, container_name):
- path = self._namespace_path(container_name) + '/'
- try:
- self.connection.request(path, method='POST')
- except AtmosError:
- e = sys.exc_info()[1]
- if e.code != 1016:
- raise
- raise ContainerAlreadyExistsError(e, self, container_name)
- return self.get_container(container_name)
-
- def delete_container(self, container):
- try:
- self.connection.request(self._namespace_path(container.name) + '/',
- method='DELETE')
- except AtmosError:
- e = sys.exc_info()[1]
- if e.code == 1003:
- raise ContainerDoesNotExistError(e, self, container.name)
- elif e.code == 1023:
- raise ContainerIsNotEmptyError(e, self, container.name)
- return True
-
- def get_object(self, container_name, object_name):
- container = self.get_container(container_name)
- object_name_cleaned = self._clean_object_name(object_name)
- path = self._namespace_path(container_name) + '/' + object_name_cleaned
-
- try:
- result = self.connection.request(path + '?metadata/system')
- system_meta = self._emc_meta(result)
-
- result = self.connection.request(path + '?metadata/user')
- user_meta = self._emc_meta(result)
- except AtmosError:
- e = sys.exc_info()[1]
- if e.code != 1003:
- raise
- raise ObjectDoesNotExistError(e, self, object_name)
-
- last_modified = time.strptime(system_meta['mtime'],
- '%Y-%m-%dT%H:%M:%SZ')
- last_modified = time.strftime('%a, %d %b %Y %H:%M:%S GMT',
- last_modified)
- extra = {
- 'object_id': system_meta['objectid'],
- 'last_modified': last_modified
- }
- data_hash = user_meta.pop('md5', '')
- return Object(object_name, int(system_meta['size']), data_hash, extra,
- user_meta, container, self)
-
- def upload_object(self, file_path, container, object_name, extra=None,
- verify_hash=True):
- upload_func = self._upload_file
- upload_func_kwargs = {'file_path': file_path}
- method = 'PUT'
-
- extra = extra or {}
- object_name_cleaned = self._clean_object_name(object_name)
- request_path = self._namespace_path(container.name) + '/' +\
- object_name_cleaned
- content_type = extra.get('content_type', None)
-
- try:
- self.connection.request(request_path + '?metadata/system')
- except AtmosError:
- e = sys.exc_info()[1]
- if e.code != 1003:
- raise
- method = 'POST'
-
- result_dict = self._upload_object(
- object_name=object_name,
- content_type=content_type,
- upload_func=upload_func,
- upload_func_kwargs=upload_func_kwargs,
- request_path=request_path,
- request_method=method,
- headers={}, file_path=file_path)
-
- bytes_transferred = result_dict['bytes_transferred']
-
- if extra is None:
- meta_data = {}
- else:
- meta_data = extra.get('meta_data', {})
- meta_data['md5'] = result_dict['data_hash']
- user_meta = ', '.join([k + '=' + str(v) for k, v in
- list(meta_data.items())])
- self.connection.request(request_path + '?metadata/user', method='POST',
- headers={'x-emc-meta': user_meta})
- result = self.connection.request(request_path + '?metadata/system')
- meta = self._emc_meta(result)
- del meta_data['md5']
- extra = {
- 'object_id': meta['objectid'],
- 'meta_data': meta_data,
- }
-
- return Object(object_name, bytes_transferred, result_dict['data_hash'],
- extra, meta_data, container, self)
-
- def upload_object_via_stream(self, iterator, container, object_name,
- extra=None):
- if isinstance(iterator, file):
- iterator = iter(iterator)
-
- data_hash = hashlib.md5()
- generator = read_in_chunks(iterator, CHUNK_SIZE, True)
- bytes_transferred = 0
- try:
- chunk = next(generator)
- except StopIteration:
- chunk = ''
-
- path = self._namespace_path(container.name + '/' + object_name)
- method = 'PUT'
-
- if extra is not None:
- content_type = extra.get('content_type', None)
- else:
- content_type = None
- if not content_type:
- content_type, _ = guess_file_mime_type(object_name)
-
- if not content_type:
- raise AttributeError(
- 'File content-type could not be guessed and' +
- ' no content_type value provided')
-
- try:
- self.connection.request(path + '?metadata/system')
- except AtmosError:
- e = sys.exc_info()[1]
- if e.code != 1003:
- raise
- method = 'POST'
-
- while True:
- end = bytes_transferred + len(chunk) - 1
- data_hash.update(b(chunk))
- headers = {
- 'x-emc-meta': 'md5=' + data_hash.hexdigest(),
- 'Content-Type': content_type,
- }
-
- if len(chunk) > 0 and bytes_transferred > 0:
- headers['Range'] = 'Bytes=%d-%d' % (bytes_transferred, end)
- method = 'PUT'
-
- result = self.connection.request(path, method=method, data=chunk,
- headers=headers)
- bytes_transferred += len(chunk)
-
- try:
- chunk = next(generator)
- except StopIteration:
- break
- if len(chunk) == 0:
- break
-
- data_hash = data_hash.hexdigest()
-
- if extra is None:
- meta_data = {}
- else:
- meta_data = extra.get('meta_data', {})
- meta_data['md5'] = data_hash
- user_meta = ', '.join([k + '=' + str(v) for k, v in
- list(meta_data.items())])
- self.connection.request(path + '?metadata/user', method='POST',
- headers={'x-emc-meta': user_meta})
-
- result = self.connection.request(path + '?metadata/system')
-
- meta = self._emc_meta(result)
- extra = {
- 'object_id': meta['objectid'],
- 'meta_data': meta_data,
- }
-
- return Object(object_name, bytes_transferred, data_hash, extra,
- meta_data, container, self)
-
- def download_object(self, obj, destination_path, overwrite_existing=False,
- delete_on_failure=True):
- path = self._namespace_path(obj.container.name + '/' + obj.name)
- response = self.connection.request(path, method='GET', raw=True)
-
- return self._get_object(obj=obj, callback=self._save_object,
- response=response,
- callback_kwargs={
- 'obj': obj,
- 'response': response.response,
- 'destination_path': destination_path,
- 'overwrite_existing': overwrite_existing,
- 'delete_on_failure': delete_on_failure
- },
- success_status_code=httplib.OK)
-
- def download_object_as_stream(self, obj, chunk_size=None):
- path = self._namespace_path(obj.container.name + '/' + obj.name)
- response = self.connection.request(path, method='GET', raw=True)
-
- return self._get_object(obj=obj, callback=read_in_chunks,
- response=response,
- callback_kwargs={
- 'iterator': response.response,
- 'chunk_size': chunk_size
- },
- success_status_code=httplib.OK)
-
- def delete_object(self, obj):
- path = self._namespace_path(obj.container.name) + '/' +\
- self._clean_object_name(obj.name)
- try:
- self.connection.request(path, method='DELETE')
- except AtmosError:
- e = sys.exc_info()[1]
- if e.code != 1003:
- raise
- raise ObjectDoesNotExistError(e, self, obj.name)
- return True
-
- def enable_object_cdn(self, obj):
- return True
-
- def get_object_cdn_url(self, obj, expiry=None, use_object=False):
- """
- Return an object CDN URL.
-
- :param obj: Object instance
- :type obj: :class:`Object`
-
- :param expiry: Expiry
- :type expiry: ``str``
-
- :param use_object: Use object
- :type use_object: ``bool``
-
- :rtype: ``str``
- """
- if use_object:
- path = '/rest/objects' + obj.meta_data['object_id']
- else:
- path = '/rest/namespace/' + obj.container.name + '/' + obj.name
-
- if self.secure:
- protocol = 'https'
- else:
- protocol = 'http'
-
- expiry = str(expiry or int(time.time()) + self.DEFAULT_CDN_TTL)
- params = [
- ('uid', self.key),
- ('expires', expiry),
- ]
- params.append(('signature', self._cdn_signature(path, params, expiry)))
-
- params = urlencode(params)
- path = self.path + path
- return urlparse.urlunparse((protocol, self.host, path, '', params, ''))
-
- def _cdn_signature(self, path, params, expiry):
- key = base64.b64decode(self.secret)
- signature = '\n'.join(['GET', path.lower(), self.key, expiry])
- signature = hmac.new(key, signature, hashlib.sha1).digest()
-
- return base64.b64encode(signature)
-
- def _list_objects(self, tree, object_type=None):
- listing = tree.find(self._emc_tag('DirectoryList'))
- entries = []
- for entry in listing.findall(self._emc_tag('DirectoryEntry')):
- file_type = entry.find(self._emc_tag('FileType')).text
- if object_type is not None and object_type != file_type:
- continue
- entries.append({
- 'id': entry.find(self._emc_tag('ObjectID')).text,
- 'type': file_type,
- 'name': entry.find(self._emc_tag('Filename')).text
- })
- return entries
-
- def _clean_object_name(self, name):
- return urlquote(name.encode('ascii'))
-
- def _namespace_path(self, path):
- return self.path + '/rest/namespace/' + urlquote(path.encode('ascii'))
-
- def _object_path(self, object_id):
- return self.path + '/rest/objects/' + object_id.encode('ascii')
-
- @staticmethod
- def _emc_tag(tag):
- return '{http://www.emc.com/cos/}' + tag
-
- def _emc_meta(self, response):
- meta = response.headers.get('x-emc-meta', '')
- if len(meta) == 0:
- return {}
- meta = meta.split(', ')
- return dict([x.split('=', 1) for x in meta])
-
- def iterate_container_objects(self, container):
- headers = {'x-emc-include-meta': '1'}
- path = self._namespace_path(container.name) + '/'
- result = self.connection.request(path, headers=headers)
- entries = self._list_objects(result.object, object_type='regular')
- for entry in entries:
- metadata = {'object_id': entry['id']}
- yield Object(entry['name'], 0, '', {}, metadata, container, self)
http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/auroraobjects.py
----------------------------------------------------------------------
diff --git a/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/auroraobjects.py b/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/auroraobjects.py
deleted file mode 100644
index 96e7313..0000000
--- a/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/auroraobjects.py
+++ /dev/null
@@ -1,52 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You 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.
-
-from libcloud.common.types import LibcloudError
-from libcloud.storage.providers import Provider
-from libcloud.storage.drivers.s3 import BaseS3StorageDriver, BaseS3Connection
-
-__all__ = [
- 'AuroraObjectsStorageDriver'
-]
-
-AURORA_OBJECTS_EU_HOST = 'o.auroraobjects.eu'
-
-NO_CDN_SUPPORT_ERROR = 'CDN is not supported by AuroraObjects'
-
-
-class BaseAuroraObjectsConnection(BaseS3Connection):
- host = AURORA_OBJECTS_EU_HOST
-
-
-class BaseAuroraObjectsStorageDriver(BaseS3StorageDriver):
- type = Provider.AURORAOBJECTS
- name = 'PCextreme AuroraObjects'
- website = 'https://www.pcextreme.com/aurora/objects'
-
-
-class AuroraObjectsStorageDriver(BaseAuroraObjectsStorageDriver):
- connectionCls = BaseAuroraObjectsConnection
-
- def enable_container_cdn(self, *argv):
- raise LibcloudError(NO_CDN_SUPPORT_ERROR, driver=self)
-
- def enable_object_cdn(self, *argv):
- raise LibcloudError(NO_CDN_SUPPORT_ERROR, driver=self)
-
- def get_container_cdn_url(self, *argv):
- raise LibcloudError(NO_CDN_SUPPORT_ERROR, driver=self)
-
- def get_object_cdn_url(self, *argv):
- raise LibcloudError(NO_CDN_SUPPORT_ERROR, driver=self)
http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/azure_blobs.py
----------------------------------------------------------------------
diff --git a/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/azure_blobs.py b/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/azure_blobs.py
deleted file mode 100644
index 13d42f6..0000000
--- a/apache-libcloud-1.0.0rc2/libcloud/storage/drivers/azure_blobs.py
+++ /dev/null
@@ -1,986 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You 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.
-
-from __future__ import with_statement
-
-import base64
-import os
-import binascii
-
-from xml.etree.ElementTree import Element, SubElement
-
-from libcloud.utils.py3 import PY3
-from libcloud.utils.py3 import httplib
-from libcloud.utils.py3 import urlquote
-from libcloud.utils.py3 import tostring
-from libcloud.utils.py3 import b
-
-from libcloud.utils.xml import fixxpath
-from libcloud.utils.files import read_in_chunks
-from libcloud.common.types import LibcloudError
-from libcloud.common.azure import AzureConnection
-
-from libcloud.storage.base import Object, Container, StorageDriver
-from libcloud.storage.types import ContainerIsNotEmptyError
-from libcloud.storage.types import ContainerAlreadyExistsError
-from libcloud.storage.types import InvalidContainerNameError
-from libcloud.storage.types import ContainerDoesNotExistError
-from libcloud.storage.types import ObjectDoesNotExistError
-from libcloud.storage.types import ObjectHashMismatchError
-
-if PY3:
- from io import FileIO as file
-
-# Desired number of items in each response inside a paginated request
-RESPONSES_PER_REQUEST = 100
-
-# As per the Azure documentation, if the upload file size is less than
-# 64MB, we can upload it in a single request. However, in real life azure
-# servers seem to disconnect randomly after around 5 MB or 200s of upload.
-# So, it is better that for file sizes greater than 4MB, we upload it in
-# chunks.
-# Also, with large sizes, if we use a lease, the lease will timeout after
-# 60 seconds, but the upload might still be in progress. This can be
-# handled in code, but if we use chunked uploads, the lease renewal will
-# happen automatically.
-AZURE_BLOCK_MAX_SIZE = 4 * 1024 * 1024
-
-# Azure block blocks must be maximum 4MB
-# Azure page blobs must be aligned in 512 byte boundaries (4MB fits that)
-AZURE_CHUNK_SIZE = 4 * 1024 * 1024
-
-# Azure page blob must be aligned in 512 byte boundaries
-AZURE_PAGE_CHUNK_SIZE = 512
-
-# The time period (in seconds) for which a lease must be obtained.
-# If set as -1, we get an infinite lease, but that is a bad idea. If
-# after getting an infinite lease, there was an issue in releasing the
-# lease, the object will remain 'locked' forever, unless the lease is
-# released using the lease_id (which is not exposed to the user)
-AZURE_LEASE_PERIOD = 60
-
-AZURE_STORAGE_HOST_SUFFIX = 'blob.core.windows.net'
-
-
-class AzureBlobLease(object):
- """
- A class to help in leasing an azure blob and renewing the lease
- """
- def __init__(self, driver, object_path, use_lease):
- """
- :param driver: The Azure storage driver that is being used
- :type driver: :class:`AzureStorageDriver`
-
- :param object_path: The path of the object we need to lease
- :type object_path: ``str``
-
- :param use_lease: Indicates if we must take a lease or not
- :type use_lease: ``bool``
- """
- self.object_path = object_path
- self.driver = driver
- self.use_lease = use_lease
- self.lease_id = None
- self.params = {'comp': 'lease'}
-
- def renew(self):
- """
- Renew the lease if it is older than a predefined time period
- """
- if self.lease_id is None:
- return
-
- headers = {'x-ms-lease-action': 'renew',
- 'x-ms-lease-id': self.lease_id,
- 'x-ms-lease-duration': '60'}
-
- response = self.driver.connection.request(self.object_path,
- headers=headers,
- params=self.params,
- method='PUT')
-
- if response.status != httplib.OK:
- raise LibcloudError('Unable to obtain lease', driver=self)
-
- def update_headers(self, headers):
- """
- Update the lease id in the headers
- """
- if self.lease_id:
- headers['x-ms-lease-id'] = self.lease_id
-
- def __enter__(self):
- if not self.use_lease:
- return self
-
- headers = {'x-ms-lease-action': 'acquire',
- 'x-ms-lease-duration': '60'}
-
- response = self.driver.connection.request(self.object_path,
- headers=headers,
- params=self.params,
- method='PUT')
-
- if response.status == httplib.NOT_FOUND:
- return self
- elif response.status != httplib.CREATED:
- raise LibcloudError('Unable to obtain lease', driver=self)
-
- self.lease_id = response.headers['x-ms-lease-id']
- return self
-
- def __exit__(self, type, value, traceback):
- if self.lease_id is None:
- return
-
- headers = {'x-ms-lease-action': 'release',
- 'x-ms-lease-id': self.lease_id}
- response = self.driver.connection.request(self.object_path,
- headers=headers,
- params=self.params,
- method='PUT')
-
- if response.status != httplib.OK:
- raise LibcloudError('Unable to release lease', driver=self)
-
-
-class AzureBlobsConnection(AzureConnection):
- """
- Represents a single connection to Azure Blobs
- """
-
-
-class AzureBlobsStorageDriver(StorageDriver):
- name = 'Microsoft Azure (blobs)'
- website = 'http://windows.azure.com/'
- connectionCls = AzureBlobsConnection
- hash_type = 'md5'
- supports_chunked_encoding = False
- ex_blob_type = 'BlockBlob'
-
- def __init__(self, key, secret=None, secure=True, host=None, port=None,
- **kwargs):
- self._host_argument_set = bool(host)
-
- # B64decode() this key and keep it, so that we don't have to do
- # so for every request. Minor performance improvement
- secret = base64.b64decode(b(secret))
-
- super(AzureBlobsStorageDriver, self).__init__(key=key, secret=secret,
- secure=secure, host=host,
- port=port, **kwargs)
-
- def _ex_connection_class_kwargs(self):
- result = {}
-
- # host argument has precedence
- if not self._host_argument_set:
- result['host'] = '%s.%s' % (self.key, AZURE_STORAGE_HOST_SUFFIX)
-
- return result
-
- def _xml_to_container(self, node):
- """
- Converts a container XML node to a container instance
-
- :param node: XML info of the container
- :type node: :class:`xml.etree.ElementTree.Element`
-
- :return: A container instance
- :rtype: :class:`Container`
- """
-
- name = node.findtext(fixxpath(xpath='Name'))
- props = node.find(fixxpath(xpath='Properties'))
- metadata = node.find(fixxpath(xpath='Metadata'))
-
- extra = {
- 'url': node.findtext(fixxpath(xpath='Url')),
- 'last_modified': node.findtext(fixxpath(xpath='Last-Modified')),
- 'etag': props.findtext(fixxpath(xpath='Etag')),
- 'lease': {
- 'status': props.findtext(fixxpath(xpath='LeaseStatus')),
- 'state': props.findtext(fixxpath(xpath='LeaseState')),
- 'duration': props.findtext(fixxpath(xpath='LeaseDuration')),
- },
- 'meta_data': {}
- }
-
- for meta in metadata.getchildren():
- extra['meta_data'][meta.tag] = meta.text
-
- return Container(name=name, extra=extra, driver=self)
-
- def _response_to_container(self, container_name, response):
- """
- Converts a HTTP response to a container instance
-
- :param container_name: Name of the container
- :type container_name: ``str``
-
- :param response: HTTP Response
- :type node: L{}
-
- :return: A container instance
- :rtype: :class:`Container`
- """
-
- headers = response.headers
- extra = {
- 'url': 'http://%s%s' % (response.connection.host,
- response.connection.action),
- 'etag': headers['etag'],
- 'last_modified': headers['last-modified'],
- 'lease': {
- 'status': headers.get('x-ms-lease-status', None),
- 'state': headers.get('x-ms-lease-state', None),
- 'duration': headers.get('x-ms-lease-duration', None),
- },
- 'meta_data': {}
- }
-
- for key, value in response.headers.items():
- if key.startswith('x-ms-meta-'):
- key = key.split('x-ms-meta-')[1]
- extra['meta_data'][key] = value
-
- return Container(name=container_name, extra=extra, driver=self)
-
- def _xml_to_object(self, container, blob):
- """
- Converts a BLOB XML node to an object instance
-
- :param container: Instance of the container holding the blob
- :type: :class:`Container`
-
- :param blob: XML info of the blob
- :type blob: L{}
-
- :return: An object instance
- :rtype: :class:`Object`
- """
-
- name = blob.findtext(fixxpath(xpath='Name'))
- props = blob.find(fixxpath(xpath='Properties'))
- metadata = blob.find(fixxpath(xpath='Metadata'))
- etag = props.findtext(fixxpath(xpath='Etag'))
- size = int(props.findtext(fixxpath(xpath='Content-Length')))
-
- extra = {
- 'content_type': props.findtext(fixxpath(xpath='Content-Type')),
- 'etag': etag,
- 'md5_hash': props.findtext(fixxpath(xpath='Content-MD5')),
- 'last_modified': props.findtext(fixxpath(xpath='Last-Modified')),
- 'url': blob.findtext(fixxpath(xpath='Url')),
- 'hash': props.findtext(fixxpath(xpath='Etag')),
- 'lease': {
- 'status': props.findtext(fixxpath(xpath='LeaseStatus')),
- 'state': props.findtext(fixxpath(xpath='LeaseState')),
- 'duration': props.findtext(fixxpath(xpath='LeaseDuration')),
- },
- 'content_encoding': props.findtext(fixxpath(
- xpath='Content-Encoding')),
- 'content_language': props.findtext(fixxpath(
- xpath='Content-Language')),
- 'blob_type': props.findtext(fixxpath(xpath='BlobType'))
- }
-
- if extra['md5_hash']:
- value = binascii.hexlify(base64.b64decode(b(extra['md5_hash'])))
- value = value.decode('ascii')
- extra['md5_hash'] = value
-
- meta_data = {}
- for meta in metadata.getchildren():
- meta_data[meta.tag] = meta.text
-
- return Object(name=name, size=size, hash=etag, meta_data=meta_data,
- extra=extra, container=container, driver=self)
-
- def _response_to_object(self, object_name, container, response):
- """
- Converts a HTTP response to an object (from headers)
-
- :param object_name: Name of the object
- :type object_name: ``str``
-
- :param container: Instance of the container holding the blob
- :type: :class:`Container`
-
- :param response: HTTP Response
- :type node: L{}
-
- :return: An object instance
- :rtype: :class:`Object`
- """
-
- headers = response.headers
- size = int(headers['content-length'])
- etag = headers['etag']
-
- extra = {
- 'url': 'http://%s%s' % (response.connection.host,
- response.connection.action),
- 'etag': etag,
- 'md5_hash': headers.get('content-md5', None),
- 'content_type': headers.get('content-type', None),
- 'content_language': headers.get('content-language', None),
- 'content_encoding': headers.get('content-encoding', None),
- 'last_modified': headers['last-modified'],
- 'lease': {
- 'status': headers.get('x-ms-lease-status', None),
- 'state': headers.get('x-ms-lease-state', None),
- 'duration': headers.get('x-ms-lease-duration', None),
- },
- 'blob_type': headers['x-ms-blob-type']
- }
-
- if extra['md5_hash']:
- value = binascii.hexlify(base64.b64decode(b(extra['md5_hash'])))
- value = value.decode('ascii')
- extra['md5_hash'] = value
-
- meta_data = {}
- for key, value in response.headers.items():
- if key.startswith('x-ms-meta-'):
- key = key.split('x-ms-meta-')[1]
- meta_data[key] = value
-
- return Object(name=object_name, size=size, hash=etag, extra=extra,
- meta_data=meta_data, container=container, driver=self)
-
- def iterate_containers(self):
- """
- @inherits: :class:`StorageDriver.iterate_containers`
- """
- params = {'comp': 'list',
- 'maxresults': RESPONSES_PER_REQUEST,
- 'include': 'metadata'}
-
- while True:
- response = self.connection.request('/', params)
- if response.status != httplib.OK:
- raise LibcloudError('Unexpected status code: %s' %
- (response.status), driver=self)
-
- body = response.parse_body()
- containers = body.find(fixxpath(xpath='Containers'))
- containers = containers.findall(fixxpath(xpath='Container'))
-
- for container in containers:
- yield self._xml_to_container(container)
-
- params['marker'] = body.findtext('NextMarker')
- if not params['marker']:
- break
-
- def iterate_container_objects(self, container):
- """
- @inherits: :class:`StorageDriver.iterate_container_objects`
- """
- params = {'restype': 'container',
- 'comp': 'list',
- 'maxresults': RESPONSES_PER_REQUEST,
- 'include': 'metadata'}
-
- container_path = self._get_container_path(container)
-
- while True:
- response = self.connection.request(container_path,
- params=params)
-
- if response.status == httplib.NOT_FOUND:
- raise ContainerDoesNotExistError(value=None,
- driver=self,
- container_name=container.name)
-
- elif response.status != httplib.OK:
- raise LibcloudError('Unexpected status code: %s' %
- (response.status), driver=self)
-
- body = response.parse_body()
- blobs = body.find(fixxpath(xpath='Blobs'))
- blobs = blobs.findall(fixxpath(xpath='Blob'))
-
- for blob in blobs:
- yield self._xml_to_object(container, blob)
-
- params['marker'] = body.findtext('NextMarker')
- if not params['marker']:
- break
-
- def get_container(self, container_name):
- """
- @inherits: :class:`StorageDriver.get_container`
- """
- params = {'restype': 'container'}
-
- container_path = '/%s' % (container_name)
-
- response = self.connection.request(container_path, params=params,
- method='HEAD')
-
- if response.status == httplib.NOT_FOUND:
- raise ContainerDoesNotExistError('Container %s does not exist' %
- (container_name), driver=self,
- container_name=container_name)
- elif response.status != httplib.OK:
- raise LibcloudError('Unexpected status code: %s' %
- (response.status), driver=self)
-
- return self._response_to_container(container_name, response)
-
- def get_object(self, container_name, object_name):
- """
- @inherits: :class:`StorageDriver.get_object`
- """
-
- container = self.get_container(container_name=container_name)
- object_path = self._get_object_path(container, object_name)
-
- response = self.connection.request(object_path, method='HEAD')
-
- if response.status == httplib.OK:
- obj = self._response_to_object(object_name, container, response)
- return obj
-
- raise ObjectDoesNotExistError(value=None, driver=self,
- object_name=object_name)
-
- def _get_container_path(self, container):
- """
- Return a container path
-
- :param container: Container instance
- :type container: :class:`Container`
-
- :return: A path for this container.
- :rtype: ``str``
- """
- return '/%s' % (container.name)
-
- def _get_object_path(self, container, object_name):
- """
- Return an object's CDN path.
-
- :param container: Container instance
- :type container: :class:`Container`
-
- :param object_name: Object name
- :type object_name: :class:`str`
-
- :return: A path for this object.
- :rtype: ``str``
- """
- container_url = self._get_container_path(container)
- object_name_cleaned = urlquote(object_name)
- object_path = '%s/%s' % (container_url, object_name_cleaned)
- return object_path
-
- def create_container(self, container_name):
- """
- @inherits: :class:`StorageDriver.create_container`
- """
- params = {'restype': 'container'}
-
- container_path = '/%s' % (container_name)
- response = self.connection.request(container_path, params=params,
- method='PUT')
-
- if response.status == httplib.CREATED:
- return self._response_to_container(container_name, response)
- elif response.status == httplib.CONFLICT:
- raise ContainerAlreadyExistsError(
- value='Container with this name already exists. The name must '
- 'be unique among all the containers in the system',
- container_name=container_name, driver=self)
- elif response.status == httplib.BAD_REQUEST:
- raise InvalidContainerNameError(value='Container name contains ' +
- 'invalid characters.',
- container_name=container_name,
- driver=self)
-
- raise LibcloudError('Unexpected status code: %s' % (response.status),
- driver=self)
-
- def delete_container(self, container):
- """
- @inherits: :class:`StorageDriver.delete_container`
- """
- # Azure does not check if the container is empty. So, we will do
- # a check to ensure that the behaviour is similar to other drivers
- for obj in container.iterate_objects():
- raise ContainerIsNotEmptyError(
- value='Container must be empty before it can be deleted.',
- container_name=container.name, driver=self)
-
- params = {'restype': 'container'}
- container_path = self._get_container_path(container)
-
- # Note: All the objects in the container must be deleted first
- response = self.connection.request(container_path, params=params,
- method='DELETE')
-
- if response.status == httplib.ACCEPTED:
- return True
- elif response.status == httplib.NOT_FOUND:
- raise ContainerDoesNotExistError(value=None,
- driver=self,
- container_name=container.name)
-
- return False
-
- def download_object(self, obj, destination_path, overwrite_existing=False,
- delete_on_failure=True):
- """
- @inherits: :class:`StorageDriver.download_object`
- """
- obj_path = self._get_object_path(obj.container, obj.name)
- response = self.connection.request(obj_path, raw=True, data=None)
-
- return self._get_object(obj=obj, callback=self._save_object,
- response=response,
- callback_kwargs={
- 'obj': obj,
- 'response': response.response,
- 'destination_path': destination_path,
- 'overwrite_existing': overwrite_existing,
- 'delete_on_failure': delete_on_failure},
- success_status_code=httplib.OK)
-
- def download_object_as_stream(self, obj, chunk_size=None):
- """
- @inherits: :class:`StorageDriver.download_object_as_stream`
- """
- obj_path = self._get_object_path(obj.container, obj.name)
- response = self.connection.request(obj_path, raw=True, data=None)
-
- return self._get_object(obj=obj, callback=read_in_chunks,
- response=response,
- callback_kwargs={'iterator': response.response,
- 'chunk_size': chunk_size},
- success_status_code=httplib.OK)
-
- def _upload_in_chunks(self, response, data, iterator, object_path,
- blob_type, lease, calculate_hash=True):
- """
- Uploads data from an interator in fixed sized chunks to S3
-
- :param response: Response object from the initial POST request
- :type response: :class:`RawResponse`
-
- :param data: Any data from the initial POST request
- :type data: ``str``
-
- :param iterator: The generator for fetching the upload data
- :type iterator: ``generator``
-
- :param object_path: The path of the object to which we are uploading
- :type object_name: ``str``
-
- :param blob_type: The blob type being uploaded
- :type blob_type: ``str``
-
- :param lease: The lease object to be used for renewal
- :type lease: :class:`AzureBlobLease`
-
- :keyword calculate_hash: Indicates if we must calculate the data hash
- :type calculate_hash: ``bool``
-
- :return: A tuple of (status, checksum, bytes transferred)
- :rtype: ``tuple``
- """
-
- # Get the upload id from the response xml
- if response.status != httplib.CREATED:
- raise LibcloudError('Error initializing upload. Code: %d' %
- (response.status), driver=self)
-
- data_hash = None
- if calculate_hash:
- data_hash = self._get_hash_function()
-
- bytes_transferred = 0
- count = 1
- chunks = []
- headers = {}
-
- lease.update_headers(headers)
-
- if blob_type == 'BlockBlob':
- params = {'comp': 'block'}
- else:
- params = {'comp': 'page'}
-
- # Read the input data in chunk sizes suitable for AWS
- for data in read_in_chunks(iterator, AZURE_CHUNK_SIZE):
- data = b(data)
- content_length = len(data)
- offset = bytes_transferred
- bytes_transferred += content_length
-
- if calculate_hash:
- data_hash.update(data)
-
- chunk_hash = self._get_hash_function()
- chunk_hash.update(data)
- chunk_hash = base64.b64encode(b(chunk_hash.digest()))
-
- headers['Content-MD5'] = chunk_hash.decode('utf-8')
- headers['Content-Length'] = content_length
-
- if blob_type == 'BlockBlob':
- # Block id can be any unique string that is base64 encoded
- # A 10 digit number can hold the max value of 50000 blocks
- # that are allowed for azure
- block_id = base64.b64encode(b('%10d' % (count)))
- block_id = block_id.decode('utf-8')
- params['blockid'] = block_id
-
- # Keep this data for a later commit
- chunks.append(block_id)
- else:
- headers['x-ms-page-write'] = 'update'
- headers['x-ms-range'] = 'bytes=%d-%d' % \
- (offset, (bytes_transferred - 1))
-
- # Renew lease before updating
- lease.renew()
-
- resp = self.connection.request(object_path, method='PUT',
- data=data, headers=headers,
- params=params)
-
- if resp.status != httplib.CREATED:
- resp.parse_error()
- raise LibcloudError('Error uploading chunk %d. Code: %d' %
- (count, resp.status), driver=self)
-
- count += 1
-
- if calculate_hash:
- data_hash = data_hash.hexdigest()
-
- if blob_type == 'BlockBlob':
- self._commit_blocks(object_path, chunks, lease)
-
- # The Azure service does not return a hash immediately for
- # chunked uploads. It takes some time for the data to get synced
- response.headers['content-md5'] = None
-
- return (True, data_hash, bytes_transferred)
-
- def _commit_blocks(self, object_path, chunks, lease):
- """
- Makes a final commit of the data.
-
- :param object_path: Server side object path.
- :type object_path: ``str``
-
- :param upload_id: A list of (chunk_number, chunk_hash) tuples.
- :type upload_id: ``list``
- """
-
- root = Element('BlockList')
-
- for block_id in chunks:
- part = SubElement(root, 'Uncommitted')
- part.text = str(block_id)
-
- data = tostring(root)
- params = {'comp': 'blocklist'}
- headers = {}
-
- lease.update_headers(headers)
- lease.renew()
-
- response = self.connection.request(object_path, data=data,
- params=params, headers=headers,
- method='PUT')
-
- if response.status != httplib.CREATED:
- raise LibcloudError('Error in blocklist commit', driver=self)
-
- def _check_values(self, blob_type, object_size):
- """
- Checks if extension arguments are valid
-
- :param blob_type: The blob type that is being uploaded
- :type blob_type: ``str``
-
- :param object_size: The (max) size of the object being uploaded
- :type object_size: ``int``
- """
-
- if blob_type not in ['BlockBlob', 'PageBlob']:
- raise LibcloudError('Invalid blob type', driver=self)
-
- if blob_type == 'PageBlob':
- if not object_size:
- raise LibcloudError('Max blob size is mandatory for page blob',
- driver=self)
-
- if object_size % AZURE_PAGE_CHUNK_SIZE:
- raise LibcloudError('Max blob size is not aligned to '
- 'page boundary', driver=self)
-
- def upload_object(self, file_path, container, object_name, extra=None,
- verify_hash=True, ex_blob_type=None, ex_use_lease=False):
- """
- Upload an object currently located on a disk.
-
- @inherits: :class:`StorageDriver.upload_object`
-
- :param ex_blob_type: Storage class
- :type ex_blob_type: ``str``
-
- :param ex_use_lease: Indicates if we must take a lease before upload
- :type ex_use_lease: ``bool``
- """
-
- if ex_blob_type is None:
- ex_blob_type = self.ex_blob_type
-
- # Get the size of the file
- file_size = os.stat(file_path).st_size
-
- # The presumed size of the object
- object_size = file_size
-
- self._check_values(ex_blob_type, file_size)
-
- with file(file_path, 'rb') as file_handle:
- iterator = iter(file_handle)
-
- # If size is greater than 64MB or type is Page, upload in chunks
- if ex_blob_type == 'PageBlob' or file_size > AZURE_BLOCK_MAX_SIZE:
- # For chunked upload of block blobs, the initial size must
- # be 0.
- if ex_blob_type == 'BlockBlob':
- object_size = None
-
- object_path = self._get_object_path(container, object_name)
-
- upload_func = self._upload_in_chunks
- upload_func_kwargs = {'iterator': iterator,
- 'object_path': object_path,
- 'blob_type': ex_blob_type,
- 'lease': None}
- else:
- upload_func = self._stream_data
- upload_func_kwargs = {'iterator': iterator,
- 'chunked': False,
- 'calculate_hash': verify_hash}
-
- return self._put_object(container=container,
- object_name=object_name,
- object_size=object_size,
- upload_func=upload_func,
- upload_func_kwargs=upload_func_kwargs,
- file_path=file_path, extra=extra,
- verify_hash=verify_hash,
- blob_type=ex_blob_type,
- use_lease=ex_use_lease)
-
- def upload_object_via_stream(self, iterator, container, object_name,
- verify_hash=False, extra=None,
- ex_use_lease=False, ex_blob_type=None,
- ex_page_blob_size=None):
- """
- @inherits: :class:`StorageDriver.upload_object_via_stream`
-
- :param ex_blob_type: Storage class
- :type ex_blob_type: ``str``
-
- :param ex_page_blob_size: The maximum size to which the
- page blob can grow to
- :type ex_page_blob_size: ``int``
-
- :param ex_use_lease: Indicates if we must take a lease before upload
- :type ex_use_lease: ``bool``
- """
-
- if ex_blob_type is None:
- ex_blob_type = self.ex_blob_type
-
- self._check_values(ex_blob_type, ex_page_blob_size)
-
- object_path = self._get_object_path(container, object_name)
-
- upload_func = self._upload_in_chunks
- upload_func_kwargs = {'iterator': iterator,
- 'object_path': object_path,
- 'blob_type': ex_blob_type,
- 'lease': None}
-
- return self._put_object(container=container,
- object_name=object_name,
- object_size=ex_page_blob_size,
- upload_func=upload_func,
- upload_func_kwargs=upload_func_kwargs,
- extra=extra, verify_hash=verify_hash,
- blob_type=ex_blob_type,
- use_lease=ex_use_lease)
-
- def delete_object(self, obj):
- """
- @inherits: :class:`StorageDriver.delete_object`
- """
- object_path = self._get_object_path(obj.container, obj.name)
- response = self.connection.request(object_path, method='DELETE')
-
- if response.status == httplib.ACCEPTED:
- return True
- elif response.status == httplib.NOT_FOUND:
- raise ObjectDoesNotExistError(value=None, driver=self,
- object_name=obj.name)
-
- return False
-
- def _update_metadata(self, headers, meta_data):
- """
- Update the given metadata in the headers
-
- :param headers: The headers dictionary to be updated
- :type headers: ``dict``
-
- :param meta_data: Metadata key value pairs
- :type meta_data: ``dict``
- """
- for key, value in list(meta_data.items()):
- key = 'x-ms-meta-%s' % (key)
- headers[key] = value
-
- def _prepare_upload_headers(self, object_name, object_size,
- extra, meta_data, blob_type):
- """
- Prepare headers for uploading an object
-
- :param object_name: The full name of the object being updated
- :type object_name: ``str``
-
- :param object_size: The size of the object. In case of PageBlobs,
- this indicates the maximum size the blob can grow to
- :type object_size: ``int``
-
- :param extra: Extra control data for the upload
- :type extra: ``dict``
-
- :param meta_data: Metadata key value pairs
- :type meta_data: ``dict``
-
- :param blob_type: Page or Block blob type
- :type blob_type: ``str``
- """
- headers = {}
-
- if blob_type is None:
- blob_type = self.ex_blob_type
-
- headers['x-ms-blob-type'] = blob_type
-
- self._update_metadata(headers, meta_data)
-
- if object_size is not None:
- headers['Content-Length'] = object_size
-
- if blob_type == 'PageBlob':
- headers['Content-Length'] = 0
- headers['x-ms-blob-content-length'] = object_size
-
- return headers
-
- def _put_object(self, container, object_name, object_size, upload_func,
- upload_func_kwargs, file_path=None, extra=None,
- verify_hash=True, blob_type=None, use_lease=False):
- """
- Control function that does the real job of uploading data to a blob
- """
- extra = extra or {}
- meta_data = extra.get('meta_data', {})
- content_type = extra.get('content_type', None)
-
- headers = self._prepare_upload_headers(object_name, object_size,
- extra, meta_data, blob_type)
-
- object_path = self._get_object_path(container, object_name)
-
- # Get a lease if required and do the operations
- with AzureBlobLease(self, object_path, use_lease) as lease:
- if 'lease' in upload_func_kwargs:
- upload_func_kwargs['lease'] = lease
-
- lease.update_headers(headers)
-
- iterator = iter('')
- result_dict = self._upload_object(object_name, content_type,
- upload_func, upload_func_kwargs,
- object_path, headers=headers,
- file_path=file_path,
- iterator=iterator)
-
- response = result_dict['response']
- bytes_transferred = result_dict['bytes_transferred']
- data_hash = result_dict['data_hash']
- headers = response.headers
- response = response.response
-
- if response.status != httplib.CREATED:
- raise LibcloudError(
- 'Unexpected status code, status_code=%s' % (response.status),
- driver=self)
-
- server_hash = headers['content-md5']
-
- if server_hash:
- server_hash = binascii.hexlify(base64.b64decode(b(server_hash)))
- server_hash = server_hash.decode('utf-8')
- else:
- # TODO: HACK - We could poll the object for a while and get
- # the hash
- pass
-
- if (verify_hash and server_hash and data_hash != server_hash):
- raise ObjectHashMismatchError(
- value='MD5 hash checksum does not match',
- object_name=object_name, driver=self)
-
- return Object(name=object_name, size=bytes_transferred,
- hash=headers['etag'], extra=None,
- meta_data=meta_data, container=container,
- driver=self)
-
- def ex_set_object_metadata(self, obj, meta_data):
- """
- Set metadata for an object
-
- :param obj: The blob object
- :type obj: :class:`Object`
-
- :param meta_data: Metadata key value pairs
- :type meta_data: ``dict``
- """
- object_path = self._get_object_path(obj.container, obj.name)
- params = {'comp': 'metadata'}
- headers = {}
-
- self._update_metadata(headers, meta_data)
-
- response = self.connection.request(object_path, method='PUT',
- params=params,
- headers=headers)
-
- if response.status != httplib.OK:
- response.parse_error('Setting metadata')