# 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

from typing import Dict
from typing import Iterator
from typing import List
from typing import Optional
from typing import Type

import os.path  # pylint: disable-msg=W0404
import hashlib
import warnings
import errno
from os.path import join as pjoin

from libcloud.utils.py3 import httplib
from libcloud.utils.py3 import b

import libcloud.utils.files
from libcloud.common.types import LibcloudError
from libcloud.common.base import Connection
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,  # type: str
        size,  # type: int
        hash,  # type: str
        extra,  # type: dict
        meta_data,  # type: dict
        container,  # type: Container
        driver,  # type: StorageDriver
    ):
        """
        :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:`libcloud.storage.base.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:`libcloud.storage.base.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):
        # type: () -> str
        return self.driver.get_object_cdn_url(obj=self)

    def enable_cdn(self):
        # type: () -> bool
        return self.driver.enable_object_cdn(obj=self)

    def download(
        self, destination_path, overwrite_existing=False, delete_on_failure=True
    ):
        # type: (str, bool, bool) -> bool
        return self.driver.download_object(
            obj=self,
            destination_path=destination_path,
            overwrite_existing=overwrite_existing,
            delete_on_failure=delete_on_failure,
        )

    def as_stream(self, chunk_size=None):
        # type: (Optional[int]) -> Iterator[bytes]
        return self.driver.download_object_as_stream(obj=self, chunk_size=chunk_size)

    def download_range(
        self,
        destination_path,
        start_bytes,
        end_bytes=None,
        overwrite_existing=False,
        delete_on_failure=True,
    ):
        # type: (str, int, Optional[int], bool, bool) -> bool
        return self.driver.download_object_range(
            obj=self,
            destination_path=destination_path,
            start_bytes=start_bytes,
            end_bytes=end_bytes,
            overwrite_existing=overwrite_existing,
            delete_on_failure=delete_on_failure,
        )

    def range_as_stream(self, start_bytes, end_bytes=None, chunk_size=None):
        # type: (int, Optional[int], Optional[int]) -> Iterator[bytes]
        return self.driver.download_object_range_as_stream(
            obj=self,
            start_bytes=start_bytes,
            end_bytes=end_bytes,
            chunk_size=chunk_size,
        )

    def delete(self):
        # type: () -> bool
        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,  # type: str
        extra,  # type: dict
        driver,  # type: StorageDriver
    ):
        """
        :param name: Container name (must be unique).
        :type name: ``str``

        :param extra: Extra attributes.
        :type extra: ``dict``

        :param driver: StorageDriver instance.
        :type driver: :class:`libcloud.storage.base.StorageDriver`
        """

        self.name = name
        self.extra = extra or {}
        self.driver = driver

    def iterate_objects(self, prefix=None, ex_prefix=None):
        # type: (Optional[str], Optional[str]) -> Iterator[Object]
        return self.driver.iterate_container_objects(
            container=self, prefix=prefix, ex_prefix=ex_prefix
        )

    def list_objects(self, prefix=None, ex_prefix=None):
        # type: (Optional[str], Optional[str]) -> List[Object]
        return self.driver.list_container_objects(
            container=self, prefix=prefix, ex_prefix=ex_prefix
        )

    def get_cdn_url(self):
        # type: () -> str
        return self.driver.get_container_cdn_url(container=self)

    def enable_cdn(self):
        # type: () -> bool
        return self.driver.enable_container_cdn(container=self)

    def get_object(self, object_name):
        # type: (str) -> Object
        return self.driver.get_object(container_name=self.name, object_name=object_name)

    def upload_object(
        self, file_path, object_name, extra=None, verify_hash=True, headers=None
    ):
        # type: (str, str, Optional[dict], bool, Optional[Dict[str, str]]) -> Object  # noqa: E501
        return self.driver.upload_object(
            file_path,
            self,
            object_name,
            extra=extra,
            verify_hash=verify_hash,
            headers=headers,
        )

    def upload_object_via_stream(self, iterator, object_name, extra=None, headers=None):
        # type: (Iterator[bytes], str, Optional[dict], Optional[Dict[str, str]]) -> Object  # noqa: E501
        return self.driver.upload_object_via_stream(
            iterator, self, object_name, extra=extra, headers=headers
        )

    def download_object(
        self, obj, destination_path, overwrite_existing=False, delete_on_failure=True
    ):
        # type: (Object, str, bool, bool) -> bool
        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):
        # type: (Object, Optional[int]) -> Iterator[bytes]
        return self.driver.download_object_as_stream(obj, chunk_size)

    def download_object_range(
        self,
        obj,
        destination_path,
        start_bytes,
        end_bytes=None,
        overwrite_existing=False,
        delete_on_failure=True,
    ):
        # type: (Object, str, int, Optional[int], bool, bool) -> bool
        return self.driver.download_object_range(
            obj=obj,
            destination_path=destination_path,
            start_bytes=start_bytes,
            end_bytes=end_bytes,
            overwrite_existing=overwrite_existing,
            delete_on_failure=delete_on_failure,
        )

    def download_object_range_as_stream(
        self, obj, start_bytes, end_bytes=None, chunk_size=None
    ):
        # type: (Object, int, Optional[int], Optional[int]) -> Iterator[bytes]
        return self.driver.download_object_range_as_stream(
            obj=obj, start_bytes=start_bytes, end_bytes=end_bytes, chunk_size=chunk_size
        )

    def delete_object(self, obj):
        # type: (Object) -> bool
        return self.driver.delete_object(obj)

    def delete(self):
        # type: () -> bool
        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  # type: Type[Connection]
    name = None  # type: str
    hash_type = "md5"  # type: str
    supports_chunked_encoding = False  # type: bool

    # 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  # type: bool

    def iterate_containers(self):
        # type: () -> Iterator[Container]
        """
        Return a iterator of containers for the given account

        :return: A iterator of Container instances.
        :rtype: ``iterator`` of :class:`libcloud.storage.base.Container`
        """
        raise NotImplementedError("iterate_containers not implemented for this driver")

    def list_containers(self):
        # type: () -> List[Container]
        """
        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, prefix=None, ex_prefix=None):
        # type: (Container, Optional[str], Optional[str]) -> Iterator[Object]
        """
        Return a iterator of objects for the given container.

        :param container: Container instance
        :type container: :class:`libcloud.storage.base.Container`

        :param prefix: Filter objects starting with a prefix.
        :type  prefix: ``str``

        :param ex_prefix: (Deprecated.) Filter objects starting with a prefix.
        :type  ex_prefix: ``str``

        :return: A iterator of Object instances.
        :rtype: ``iterator`` of :class:`libcloud.storage.base.Object`
        """
        raise NotImplementedError(
            "iterate_container_objects not implemented for this driver"
        )

    def list_container_objects(self, container, prefix=None, ex_prefix=None):
        # type: (Container, Optional[str], Optional[str]) -> List[Object]
        """
        Return a list of objects for the given container.

        :param container: Container instance.
        :type container: :class:`libcloud.storage.base.Container`

        :param prefix: Filter objects starting with a prefix.
        :type  prefix: ``str``

        :param ex_prefix: (Deprecated.) Filter objects starting with a prefix.
        :type  ex_prefix: ``str``

        :return: A list of Object instances.
        :rtype: ``list`` of :class:`libcloud.storage.base.Object`
        """
        return list(
            self.iterate_container_objects(
                container, prefix=prefix, ex_prefix=ex_prefix
            )
        )

    def _normalize_prefix_argument(self, prefix, ex_prefix):
        if ex_prefix:
            warnings.warn(
                "The ``ex_prefix`` argument is deprecated - "
                "please update code to use ``prefix``",
                DeprecationWarning,
            )
            return ex_prefix

        return prefix

    def _filter_listed_container_objects(self, objects, prefix):
        if prefix is not None:
            warnings.warn(
                "Driver %s does not implement native object "
                "filtering; falling back to filtering the full "
                "object stream." % self.__class__.__name__
            )

        for obj in objects:
            if prefix is None or obj.name.startswith(prefix):
                yield obj

    def get_container(self, container_name):
        # type: (str) -> Container
        """
        Return a container instance.

        :param container_name: Container name.
        :type container_name: ``str``

        :return: :class:`Container` instance.
        :rtype: :class:`libcloud.storage.base.Container`
        """
        raise NotImplementedError("get_object not implemented for this driver")

    def get_container_cdn_url(self, container):
        # type: (Container) -> str
        """
        Return a container CDN URL.

        :param container: Container instance
        :type  container: :class:`libcloud.storage.base.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):
        # type: (str, str) -> Object
        """
        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:`libcloud.storage.base.Object`
        """
        raise NotImplementedError("get_object not implemented for this driver")

    def get_object_cdn_url(self, obj):
        # type: (Object) -> str
        """
        Return an object CDN URL.

        :param obj: Object instance
        :type  obj: :class:`libcloud.storage.base.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):
        # type: (Container) -> bool
        """
        Enable container CDN.

        :param container: Container instance
        :type  container: :class:`libcloud.storage.base.Container`

        :rtype: ``bool``
        """
        raise NotImplementedError(
            "enable_container_cdn not implemented for this driver"
        )

    def enable_object_cdn(self, obj):
        # type: (Object) -> bool
        """
        Enable object CDN.

        :param obj: Object instance
        :type  obj: :class:`libcloud.storage.base.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
    ):
        # type: (Object, str, bool, bool) -> bool
        """
        Download an object to the specified destination path.

        :param obj: Object instance.
        :type obj: :class:`libcloud.storage.base.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):
        # type: (Object, Optional[int]) -> Iterator[bytes]
        """
        Return a iterator which yields object data.

        :param obj: Object instance
        :type obj: :class:`libcloud.storage.base.Object`

        :param chunk_size: Optional chunk size (in bytes).
        :type chunk_size: ``int``

        :rtype: ``iterator`` of ``bytes``
        """
        raise NotImplementedError(
            "download_object_as_stream not implemented for this driver"
        )

    def download_object_range(
        self,
        obj,
        destination_path,
        start_bytes,
        end_bytes=None,
        overwrite_existing=False,
        delete_on_failure=True,
    ):
        # type: (Object, str, int, Optional[int], bool, bool) -> bool
        """
        Download part of an object.

        :param obj: Object instance.
        :type obj: :class:`libcloud.storage.base.Object`

        :param destination_path: Full path to a file or a directory where the
                                 incoming file will be saved.
        :type destination_path: ``str``

        :param start_bytes: Start byte offset (inclusive) for the range
                            download. Offset is 0 index based so the first
                            byte in file file is "0".
        :type start_bytes: ``int``

        :param end_bytes: End byte offset (non-inclusive) for the range
                          download. If not provided, it will default to the
                          end of the file.
        :type end_bytes: ``int``

        :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_range not implemented for this driver"
        )

    def download_object_range_as_stream(
        self, obj, start_bytes, end_bytes=None, chunk_size=None
    ):
        # type: (Object, int, Optional[int], Optional[int]) -> Iterator[bytes]
        """
        Return a iterator which yields range / part of the object data.

        :param obj: Object instance
        :type obj: :class:`libcloud.storage.base.Object`

        :param start_bytes: Start byte offset (inclusive) for the range
                            download. Offset is 0 index based so the first
                            byte in file file is "0".
        :type start_bytes: ``int``

        :param end_bytes: End byte offset (non-inclusive) for the range
                          download. If not provided, it will default to the
                          end of the file.
        :type end_bytes: ``int``

        :param chunk_size: Optional chunk size (in bytes).
        :type chunk_size: ``int``

        :rtype: ``iterator`` of ``bytes``
        """
        raise NotImplementedError(
            "download_object_range_as_stream not implemented for this driver"
        )

    def upload_object(
        self,
        file_path,
        container,
        object_name,
        extra=None,
        verify_hash=True,
        headers=None,
    ):
        # type: (str, Container, str, Optional[dict], bool, Optional[Dict[str, str]]) -> Object  # noqa: E501
        """
        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:`libcloud.storage.base.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:`libcloud.storage.base.Object`
        """
        raise NotImplementedError("upload_object not implemented for this driver")

    def upload_object_via_stream(
        self, iterator, container, object_name, extra=None, headers=None
    ):
        # type: (Iterator[bytes], Container, str, Optional[dict], Optional[Dict[str, str]]) -> Object  # noqa: E501
        """
        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:`libcloud.storage.base.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: ``libcloud.storage.base.Object``
        """
        raise NotImplementedError(
            "upload_object_via_stream not implemented for this driver"
        )

    def delete_object(self, obj):
        # type: (Object) -> bool
        """
        Delete an object.

        :param obj: Object instance.
        :type obj: :class:`libcloud.storage.base.Object`

        :return: ``bool`` True on success.
        :rtype: ``bool``
        """
        raise NotImplementedError("delete_object not implemented for this driver")

    def create_container(self, container_name):
        # type: (str) -> Container
        """
        Create a new container.

        :param container_name: Container name.
        :type container_name: ``str``

        :return: Container instance on success.
        :rtype: :class:`libcloud.storage.base.Container`
        """
        raise NotImplementedError("create_container not implemented for this driver")

    def delete_container(self, container):
        # type: (Container) -> bool
        """
        Delete a container.

        :param container: Container instance
        :type container: :class:`libcloud.storage.base.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 not isinstance(success_status_code, (list, tuple)):
            success_status_codes = [success_status_code]
        else:
            success_status_codes = success_status_code

        if response.status in success_status_codes:
            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,
        partial_download=False,
    ):
        """
        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``

        :param partial_download: True if this is a range (partial) save,
                                 False otherwise.
        :type partial_download: ``bool``

        :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,
            )

        bytes_transferred = 0

        with open(file_path, "wb") as file_handle:
            for chunk in response._response.iter_content(chunk_size):
                file_handle.write(b(chunk))
                bytes_transferred += len(chunk)

        if not partial_download and int(obj.size) != int(bytes_transferred):
            # Transfer failed, support retry?
            # NOTE: We only perform this check if this is a regular and not a
            # partial / range download
            if delete_on_failure:
                try:
                    os.unlink(file_path)
                except Exception:
                    pass

            return False

        return True

    def _upload_object(
        self,
        object_name,
        content_type,
        request_path,
        request_method="PUT",
        headers=None,
        file_path=None,
        stream=None,
        chunked=False,
        multipart=False,
    ):
        """
        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 (
            stream is not None
            and not hasattr(stream, "next")
            and not hasattr(stream, "__next__")
        ):
            raise AttributeError("iterator object must implement next() " + "method.")

        headers["Content-Type"] = self._determine_content_type(
            content_type, object_name, file_path=file_path
        )

        if stream:
            response = self.connection.request(
                request_path,
                method=request_method,
                data=stream,
                headers=headers,
                raw=True,
            )
            stream_hash, stream_length = self._hash_buffered_stream(
                stream, self._get_hash_function()
            )
        else:
            with open(file_path, "rb") as file_stream:
                response = self.connection.request(
                    request_path,
                    method=request_method,
                    data=file_stream,
                    headers=headers,
                    raw=True,
                )
            with open(file_path, "rb") as file_stream:
                stream_hash, stream_length = self._hash_buffered_stream(
                    file_stream, self._get_hash_function()
                )

        if not response.success():
            response.parse_error()

        return {
            "response": response,
            "bytes_transferred": stream_length,
            "data_hash": stream_hash,
        }

    def _determine_content_type(self, content_type, object_name, file_path=None):
        if content_type:
            return content_type

        name = file_path or object_name
        content_type, _ = libcloud.utils.files.guess_file_mime_type(name)

        if self.strict_mode and not content_type:
            raise AttributeError(
                "File content-type could not be guessed for "
                '"%s" and no content_type value is provided' % name
            )

        return content_type or DEFAULT_CONTENT_TYPE

    def _hash_buffered_stream(self, stream, hasher, blocksize=65536):
        total_len = 0

        if hasattr(stream, "__next__") or hasattr(stream, "next"):
            # Ensure we start from the begining of a stream in case stream is
            # not at the beginning
            if hasattr(stream, "seek"):
                try:
                    stream.seek(0)
                except OSError as e:
                    if e.errno != errno.ESPIPE:
                        # This represents "OSError: [Errno 29] Illegal seek"
                        # error. This could either mean that the underlying
                        # handle doesn't support seek operation (e.g. pipe) or
                        # that the invalid seek position is provided. Sadly
                        # there is no good robust way to distinghuish that so
                        # we simply ignore all the "Illeal seek" errors so
                        # this function works correctly with pipes.
                        # See https://github.com/apache/libcloud/pull/1427 for
                        # details
                        raise e

            for chunk in libcloud.utils.files.read_in_chunks(iterator=stream):
                hasher.update(b(chunk))
                total_len += len(chunk)

            return (hasher.hexdigest(), total_len)

        if not hasattr(stream, "__exit__"):
            for s in stream:
                hasher.update(s)
                total_len = total_len + len(s)
            return (hasher.hexdigest(), total_len)

        with stream:
            buf = stream.read(blocksize)
            while len(buf) > 0:
                total_len = total_len + len(buf)
                hasher.update(buf)
                buf = stream.read(blocksize)

        return (hasher.hexdigest(), total_len)

    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

    def _validate_start_and_end_bytes(self, start_bytes, end_bytes=None):
        # type: (int, Optional[int]) -> bool
        """
        Method which validates that start_bytes and end_bytes arguments contain
        valid values.
        """
        if start_bytes < 0:
            raise ValueError("start_bytes must be greater than 0")

        if end_bytes is not None:
            if start_bytes > end_bytes:
                raise ValueError("start_bytes must be smaller than end_bytes")
            elif start_bytes == end_bytes:
                raise ValueError(
                    "start_bytes and end_bytes can't be the "
                    "same. end_bytes is non-inclusive"
                )

        return True

    def _get_standard_range_str(
        self, start_bytes, end_bytes=None, end_bytes_inclusive=False
    ):
        # type: (int, Optional[int], bool) -> str
        """
        Return range string which is used as a Range header value for range
        requests for drivers which follow standard Range header notation

        This returns range string in the following format:
        bytes=<start_bytes>-<end bytes>.

        For example:

        bytes=1-10
        bytes=0-2
        bytes=5-
        bytes=100-5000

        :param end_bytes_inclusive: True if "end_bytes" offset should be
        inclusive (aka opposite from the Python indexing behavior where the end
        index is not inclusive).
        """
        range_str = "bytes=%s-" % (start_bytes)

        if end_bytes is not None:
            if end_bytes_inclusive:
                range_str += str(end_bytes)
            else:
                range_str += str(end_bytes - 1)

        return range_str