#NEWLINE# def upload( self , container , objects , options = None ) : #NEWLINE# #INDENT# "\n        Upload a list of objects to a given container.\n\n        :param container: The container (or pseudo-folder path) to put the\n                          uploads into.\n        :param objects: A list of file/directory names (strings) or\n                        SwiftUploadObject instances containing a source for the\n                        created object, an object name, and an options dict\n                        (can be None) to override the options for that\n                        individual upload operation::\n\n                            [\n                                '/path/to/file',\n                                SwiftUploadObject('/path', object_name='obj1'),\n                                ...\n                            ]\n\n                        The options dict is as described below.\n\n                        The SwiftUploadObject source may be one of:\n\n                            * A file-like object (with a read method)\n                            * A string containing the path to a local\n                              file or directory\n                            * None, to indicate that we want an empty object\n\n        :param options: A dictionary containing options to override the global\n                        options specified during the service object creation.\n                        These options are applied to all upload operations\n                        performed by this call, unless overridden on a per\n                        object basis. Possible options are given below::\n\n                            {\n                                'meta': [],\n                                'header': [],\n                                'segment_size': None,\n                                'use_slo': False,\n                                'segment_container': None,\n                                'leave_segments': False,\n                                'changed': None,\n                                'skip_identical': False,\n                                'fail_fast': False,\n                                'dir_marker': False  # Only for None sources\n                            }\n\n        :returns: A generator for returning the results of the uploads.\n\n        :raises: SwiftError\n        :raises: ClientException\n        " #NEWLINE# if ( options is not None ) : #NEWLINE# #INDENT# options = dict ( self . _options , ** options ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# options = self . _options #NEWLINE# #UNINDENT# try : #NEWLINE# #INDENT# segment_size = int ( ( 0 if ( options [ 'segment_size' ] is None ) else options [ 'segment_size' ] ) ) #NEWLINE# #UNINDENT# except ValueError : #NEWLINE# #INDENT# raise SwiftError ( 'Segment size should be an integer value' ) #NEWLINE# #UNINDENT# ( container , _sep , pseudo_folder ) = container . partition ( '/' ) #NEWLINE# policy_header = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# _header = split_headers ( options [ 'header' ] ) #NEWLINE# if ( POLICY in _header ) : #NEWLINE# #INDENT# policy_header [ POLICY ] = _header [ POLICY ] #NEWLINE# #UNINDENT# create_containers = [ self . thread_manager . container_pool . submit ( self . _create_container_job , container , headers = policy_header ) ] #NEWLINE# for r in interruptable_as_completed ( create_containers ) : #NEWLINE# #INDENT# res = r . result ( ) #NEWLINE# ( yield res ) #NEWLINE# #UNINDENT# if segment_size : #NEWLINE# #INDENT# seg_container = ( container + '_segments' ) #NEWLINE# if options [ 'segment_container' ] : #NEWLINE# #INDENT# seg_container = options [ 'segment_container' ] #NEWLINE# #UNINDENT# if ( seg_container != container ) : #NEWLINE# #INDENT# if ( not policy_header ) : #NEWLINE# #INDENT# create_containers = [ self . thread_manager . container_pool . submit ( self . _create_container_job , seg_container , policy_source = container ) ] #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# create_containers = [ self . thread_manager . container_pool . submit ( self . _create_container_job , seg_container , headers = policy_header ) ] #NEWLINE# #UNINDENT# for r in interruptable_as_completed ( create_containers ) : #NEWLINE# #INDENT# res = r . result ( ) #NEWLINE# ( yield res ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# rq = Queue ( ) #NEWLINE# file_jobs = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# upload_objects = self . _make_upload_objects ( objects , pseudo_folder ) #NEWLINE# for upload_object in upload_objects : #NEWLINE# #INDENT# s = upload_object . source #NEWLINE# o = upload_object . object_name #NEWLINE# o_opts = upload_object . options #NEWLINE# details = { #NEWLINE# #INDENT# 'action' : 'upload' , #NEWLINE# 'container' : container , #NEWLINE# #UNINDENT# } #NEWLINE# if ( o_opts is not None ) : #NEWLINE# #INDENT# object_options = deepcopy ( options ) #NEWLINE# object_options . update ( o_opts ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# object_options = options #NEWLINE# #UNINDENT# if hasattr ( s , 'read' ) : #NEWLINE# #INDENT# file_future = self . thread_manager . object_uu_pool . submit ( self . _upload_object_job , container , s , o , object_options ) #NEWLINE# details [ 'file' ] = s #NEWLINE# details [ 'object' ] = o #NEWLINE# file_jobs [ file_future ] = details #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( s is not None ) : #NEWLINE# #INDENT# details [ 'path' ] = s #NEWLINE# details [ 'object' ] = o #NEWLINE# if isdir ( s ) : #NEWLINE# #INDENT# dir_future = self . thread_manager . object_uu_pool . submit ( self . _create_dir_marker_job , container , o , object_options , path = s ) #NEWLINE# file_jobs [ dir_future ] = details #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# stat ( s ) #NEWLINE# file_future = self . thread_manager . object_uu_pool . submit ( self . _upload_object_job , container , s , o , object_options , results_queue = rq ) #NEWLINE# file_jobs [ file_future ] = details #NEWLINE# #UNINDENT# except OSError as err : #NEWLINE# #INDENT# ( traceback , err_time ) = report_traceback ( ) #NEWLINE# logger . exception ( err ) #NEWLINE# res = { #NEWLINE# #INDENT# 'action' : 'upload_object' , #NEWLINE# 'container' : container , #NEWLINE# 'object' : o , #NEWLINE# 'success' : False , #NEWLINE# 'error' : err , #NEWLINE# 'traceback' : traceback , #NEWLINE# 'error_timestamp' : err_time , #NEWLINE# 'path' : s , #NEWLINE# #UNINDENT# } #NEWLINE# rq . put ( res ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# details [ 'file' ] = None #NEWLINE# details [ 'object' ] = o #NEWLINE# if object_options [ 'dir_marker' ] : #NEWLINE# #INDENT# dir_future = self . thread_manager . object_uu_pool . submit ( self . _create_dir_marker_job , container , o , object_options ) #NEWLINE# file_jobs [ dir_future ] = details #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# file_future = self . thread_manager . object_uu_pool . submit ( self . _upload_object_job , container , StringIO ( ) , o , object_options ) #NEWLINE# file_jobs [ file_future ] = details #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# Thread ( target = self . _watch_futures , args = ( file_jobs , rq ) ) . start ( ) #NEWLINE# res = get_from_queue ( rq ) #NEWLINE# cancelled = False #NEWLINE# while ( res is not None ) : #NEWLINE# #INDENT# ( yield res ) #NEWLINE# if ( not res [ 'success' ] ) : #NEWLINE# #INDENT# if ( ( not cancelled ) and options [ 'fail_fast' ] ) : #NEWLINE# #INDENT# cancelled = True #NEWLINE# for f in file_jobs : #NEWLINE# #INDENT# f . cancel ( ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# res = get_from_queue ( rq )