composer.utils.object_store.object_store#

Abstract class for utilities that upload to and download from object stores.

Classes

ObjectStore

Abstract class for implementing object stores, such as LibcloudObjectStore and S3ObjectStore.

Exceptions

ObjectStoreTransientError

Custom exception class to signify transient errors.

class composer.utils.object_store.object_store.ObjectStore[source]#

Bases: abc.ABC

Abstract class for implementing object stores, such as LibcloudObjectStore and S3ObjectStore.

close()[source]#: Close the object store.

download_object(object_name, filename, overwrite=False, callback=None)[source]#

Download an object to the specified destination path.

Parameters

object_name (str) – The name of the object to download.
filename (str | Path) – Full path to a file or a directory where the incoming file will be saved.
overwrite (bool, optional) – Whether to overwrite an existing file at filename, if it exists. (default: False)
callback ((int) -> None, optional) – If specified, the callback is periodically called with the number of bytes already downloaded and the total size of the object.

Raises

FileNotFoundError – If the file was not found in the object store.
ObjectStoreTransientError – If there was a transient connection issue with downloading the object.

get_object_size(object_name)[source]#

Get the size of an object, in bytes.

Parameters

object_name (str) – The name of the object.

Returns

int – The object size, in bytes.

Raises

FileNotFoundError – If the file was not found in the object store.
ObjectStoreTransientError – If there was a transient connection issue with getting the object size.

get_uri(object_name)[source]#

Returns the URI for object_name.

Note

This function does not check that object_name is in the object store. It computes the URI statically.

Parameters: object_name (str) – The object name.
Returns: str – The URI for object_name in the object store.

upload_object(object_name, filename, callback=None)[source]#

Upload an object currently located on a disk.

Parameters

object_name (str) – Object name (where object will be stored in the container)
filename (str | Path) – Path the the object on disk
callback ((int, int) -> None, optional) – If specified, the callback is periodically called with the number of bytes uploaded and the total size of the object being uploaded.

Raises

ObjectStoreTransientError – If there was a transient connection issue with uploading the object.

exception composer.utils.object_store.object_store.ObjectStoreTransientError[source]#

Bases: RuntimeError

Custom exception class to signify transient errors.

Implementations of the ObjectStore should re-raise any transient exceptions (e.g. too many requests, temporarily unavailable) with this class, so callers can easily detect whether they should attempt to retry any operation.

For example, the S3ObjectStore does the following:

from composer.utils import ObjectStore, ObjectStoreTransientError
import botocore.exceptions

class S3ObjectStore(ObjectStore):

    def upload_object(self, file_path: str, object_name: str):
        try:
            ...
        except botocore.exceptions.ClientError as e:
            if e.response['Error']['Code'] == 'LimitExceededException':
                raise ObjectStoreTransientError(e.response['Error']['Code']) from e
            raise e

Then, callers can automatically handle exceptions:

import time
from composer.utils import ObjectStore, ObjectStoreTransientError

def upload_file(object_store: ObjectStore, max_num_attempts: int = 3):
    for i in range(max_num_attempts):
        try:
            object_store.upload_object(...)
        except ObjectStoreTransientError:
            if i + 1 == max_num_attempts:
                raise
            else:
                # Try again after exponential back-off
                time.sleep(2**i)
        else:
            # upload successful
            return