libcloud.storage.drivers.dummy.DummyStorageDriver : API documentation

Method	__init__	No summary
Method	get_meta_data	No summary
Method	iterate_containers	No summary
Method	list_container_objects	`Return a list of objects for the given container.`
Method	get_container	No summary
Method	get_container_cdn_url	No summary
Method	get_object	No summary
Method	get_object_cdn_url	No summary
Method	create_container	No summary
Method	delete_container	No summary
Method	download_object	`Download an object to the specified destination path.`
Method	download_object_as_stream	No summary
Method	upload_object	No summary
Method	upload_object_via_stream	No summary
Method	delete_object	No summary
Method	_add_object	Undocumented

Inherited from StorageDriver:

Method	list_containers	`Return a list of containers.`
Method	iterate_container_objects	`Return a generator of objects for the given container.`
Method	enable_container_cdn	`Enable container CDN.`
Method	enable_object_cdn	`Enable object CDN.`
Method	_get_object	`Call passed callback and start transfer of the object'`
Method	_save_object	`Save object to the provided path.`
Method	_upload_object	`Helper function for setting common request headers and calling the passed in callback which uploads an object.`
Method	_upload_data	`Upload data stored in a string.`
Method	_stream_data	`Stream a data over an http connection.`
Method	_upload_file	`Upload a file to the server.`
Method	_get_hash_function	`Return instantiated hash function for the hash type supported by the provider.`

Inherited from BaseDriver (via StorageDriver):

Method _ex_connection_class_kwargs Return extra connection keyword arguments which are passed to the Connection class constructor.

def __init__(self, api_key, api_secret): (source)

overrides libcloud.storage.base.StorageDriver.__init__

@param    api_key:    API key or username to used (required)
@type     api_key:    C{str}
@param    api_secret: Secret password to be used (required)
@type     api_secret: C{str}
@rtype: C{None}

def get_meta_data(self): (source)

>>> driver = DummyStorageDriver('key', 'secret')
>>> driver.get_meta_data()['object_count']
0
>>> driver.get_meta_data()['container_count']
0
>>> driver.get_meta_data()['bytes_used']
0
>>> container = driver.create_container(container_name='test container 1')
>>> container = driver.create_container(container_name='test container 2')
>>> obj = container.upload_object_via_stream(
...  object_name='test object', iterator=DummyFileObject(5, 10), extra={})
>>> driver.get_meta_data()['object_count']
1
>>> driver.get_meta_data()['container_count']
2
>>> driver.get_meta_data()['bytes_used']
50

@rtype: C{dict}

def iterate_containers(self): (source)

overrides libcloud.storage.base.StorageDriver.iterate_containers

>>> driver = DummyStorageDriver('key', 'secret')
>>> list(driver.iterate_containers())
[]
>>> container = driver.create_container(container_name='test container 1')
>>> container
<Container: name=test container 1, provider=Dummy Storage Provider>
>>> container.name
'test container 1'
>>> container = driver.create_container(container_name='test container 2')
>>> container
<Container: name=test container 2, provider=Dummy Storage Provider>
>>> container = driver.create_container(
...  container_name='test container 2') #doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
ContainerAlreadyExistsError:
>>> container_list=list(driver.iterate_containers())
>>> sorted([container.name for container in container_list])
['test container 1', 'test container 2']

@inherits: L{StorageDriver.iterate_containers}

def list_container_objects(self, container): (source)

overrides libcloud.storage.base.StorageDriver.list_container_objects

Return a list of objects for the given container.

@param container: Container instance
@type container: L{Container}

@return: A list of Object instances.
@rtype: C{list} of L{Object}

def get_container(self, container_name): (source)

overrides libcloud.storage.base.StorageDriver.get_container

>>> driver = DummyStorageDriver('key', 'secret')
>>> driver.get_container('unknown') #doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
ContainerDoesNotExistError:
>>> container = driver.create_container(container_name='test container 1')
>>> container
<Container: name=test container 1, provider=Dummy Storage Provider>
>>> container.name
'test container 1'
>>> driver.get_container('test container 1')
<Container: name=test container 1, provider=Dummy Storage Provider>

@inherits: L{StorageDriver.get_container}

def get_container_cdn_url(self, container): (source)

overrides libcloud.storage.base.StorageDriver.get_container_cdn_url

>>> driver = DummyStorageDriver('key', 'secret')
>>> driver.get_container('unknown') #doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
ContainerDoesNotExistError:
>>> container = driver.create_container(container_name='test container 1')
>>> container
<Container: name=test container 1, provider=Dummy Storage Provider>
>>> container.name
'test container 1'
>>> container.get_cdn_url()
'http://www.test.com/container/test_container_1'

@inherits: L{StorageDriver.get_container_cdn_url}

def get_object(self, container_name, object_name): (source)

overrides libcloud.storage.base.StorageDriver.get_object

>>> driver = DummyStorageDriver('key', 'secret')
>>> driver.get_object('unknown', 'unknown') #doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
ContainerDoesNotExistError:
>>> container = driver.create_container(container_name='test container 1')
>>> container
<Container: name=test container 1, provider=Dummy Storage Provider>
>>> driver.get_object(
...  'test container 1', 'unknown') #doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
ObjectDoesNotExistError:
>>> obj = container.upload_object_via_stream(object_name='test object',
...      iterator=DummyFileObject(5, 10), extra={})
>>> obj
<Object: name=test object, size=50, hash=None, provider=Dummy Storage Provider ...>

@inherits: L{StorageDriver.get_object}

def get_object_cdn_url(self, obj): (source)

overrides libcloud.storage.base.StorageDriver.get_object_cdn_url

>>> driver = DummyStorageDriver('key', 'secret')
>>> container = driver.create_container(container_name='test container 1')
>>> container
<Container: name=test container 1, provider=Dummy Storage Provider>
>>> obj = container.upload_object_via_stream(object_name='test object 5',
...      iterator=DummyFileObject(5, 10), extra={})
>>> obj
<Object: name=test object 5, size=50, hash=None, provider=Dummy Storage Provider ...>
>>> obj.get_cdn_url()
'http://www.test.com/object/test_object_5'

@inherits: L{StorageDriver.get_object_cdn_url}

def create_container(self, container_name): (source)

overrides libcloud.storage.base.StorageDriver.create_container

>>> driver = DummyStorageDriver('key', 'secret')
>>> container = driver.create_container(container_name='test container 1')
>>> container
<Container: name=test container 1, provider=Dummy Storage Provider>
>>> container = driver.create_container(
...    container_name='test container 1') #doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
ContainerAlreadyExistsError:

@inherits: L{StorageDriver.create_container}

def delete_container(self, container): (source)

overrides libcloud.storage.base.StorageDriver.delete_container

>>> driver = DummyStorageDriver('key', 'secret')
>>> container = Container(name = 'test container',
...    extra={'object_count': 0}, driver=driver)
>>> driver.delete_container(container=container)#doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
ContainerDoesNotExistError:
>>> container = driver.create_container(
...      container_name='test container 1') #doctest: +IGNORE_EXCEPTION_DETAIL
>>> len(driver._containers)
1
>>> driver.delete_container(container=container)
True
>>> len(driver._containers)
0
>>> container = driver.create_container(
...    container_name='test container 1') #doctest: +IGNORE_EXCEPTION_DETAIL
>>> obj = container.upload_object_via_stream(
...   object_name='test object', iterator=DummyFileObject(5, 10), extra={})
>>> driver.delete_container(container=container)#doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
ContainerIsNotEmptyError:

@inherits: L{StorageDriver.delete_container}

def download_object(self, obj, destination_path, overwrite_existing=False, delete_on_failure=True): (source)

overrides libcloud.storage.base.StorageDriver.download_object

Download an object to the specified destination path.

@param obj: Object instance.
@type obj: L{Object}

@param destination_path: Full path to a file or a directory where the
                        incoming file will be saved.
@type destination_path: C{str}

@param overwrite_existing: True to overwrite an existing file,
    defaults to False.
@type overwrite_existing: C{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: C{bool}

@return: True if an object has been successfully downloaded, False
otherwise.
@rtype: C{bool}

def download_object_as_stream(self, obj, chunk_size=None): (source)

overrides libcloud.storage.base.StorageDriver.download_object_as_stream

>>> driver = DummyStorageDriver('key', 'secret')
>>> container = driver.create_container(
...   container_name='test container 1') #doctest: +IGNORE_EXCEPTION_DETAIL
>>> obj = container.upload_object_via_stream(object_name='test object',
...    iterator=DummyFileObject(5, 10), extra={})
>>> stream = container.download_object_as_stream(obj)
>>> stream #doctest: +ELLIPSIS
<...closed...>

@inherits: L{StorageDriver.download_object_as_stream}

def upload_object(self, file_path, container, object_name, extra=None, file_hash=None): (source)

overrides libcloud.storage.base.StorageDriver.upload_object

>>> driver = DummyStorageDriver('key', 'secret')
>>> container = driver.create_container(container_name='test container 1')
>>> container.upload_object(file_path='/tmp/inexistent.file',
...     object_name='test') #doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
LibcloudError:
>>> file_path = path = os.path.abspath(__file__)
>>> file_size = os.path.getsize(file_path)
>>> obj = container.upload_object(file_path=file_path, object_name='test')
>>> obj #doctest: +ELLIPSIS
<Object: name=test, size=...>
>>> obj.size == file_size
True

@inherits: L{StorageDriver.upload_object}
@param file_hash: File hash
@type file_hash: C{str}

def upload_object_via_stream(self, iterator, container, object_name, extra=None): (source)

overrides libcloud.storage.base.StorageDriver.upload_object_via_stream

>>> driver = DummyStorageDriver('key', 'secret')
>>> container = driver.create_container(
...    container_name='test container 1') #doctest: +IGNORE_EXCEPTION_DETAIL
>>> obj = container.upload_object_via_stream(
...   object_name='test object', iterator=DummyFileObject(5, 10), extra={})
>>> obj #doctest: +ELLIPSIS
<Object: name=test object, size=50, ...>

@inherits: L{StorageDriver.upload_object_via_stream}

def delete_object(self, obj): (source)

overrides libcloud.storage.base.StorageDriver.delete_object

>>> driver = DummyStorageDriver('key', 'secret')
>>> container = driver.create_container(
...   container_name='test container 1') #doctest: +IGNORE_EXCEPTION_DETAIL
>>> obj = container.upload_object_via_stream(object_name='test object',
...   iterator=DummyFileObject(5, 10), extra={})
>>> obj #doctest: +ELLIPSIS
<Object: name=test object, size=50, ...>
>>> container.delete_object(obj=obj)
True
>>> obj = Object(name='test object 2',
...    size=1000, hash=None, extra=None,
...    meta_data=None, container=container,driver=None)
>>> container.delete_object(obj=obj) #doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
ObjectDoesNotExistError:

@inherits: L{StorageDriver.delete_object}

l.s.d.d.DummyStorageDriver(StorageDriver) : class documentation