Fixes for transferring files from external services using the http protocol. Automatic file transfers from Pac Bio SMRT Portal are functional for version 1.1.0.

Author: gregvonkuster

external_service_types/pacific_biosciences_smrt_portal.xml

@@ -2,9 +2,6 @@
     <description></description>
     <version>1</version>
     <data_transfer_settings>
-        <!--
-        <data_transfer protocol='scp' automatic_transfer='True' host='host' user_name='user_name' password='password' data_location='data_location' />
-        -->
         <data_transfer protocol='http' automatic_transfer='True' />
     </data_transfer_settings>
     <run_details>

lib/galaxy/jobs/deferred/data_transfer.py

@@ -4,8 +4,6 @@
 import logging, urllib2, re, shutil
 
 from galaxy import eggs
-from galaxy.util import json
-from string import Template
 from sqlalchemy import and_
 
 from galaxy.util.odict import odict
@@ -19,6 +17,8 @@
 
 log = logging.getLogger( __name__ )
 
+__all__ = [ 'DataTransfer' ]
+
 class DataTransfer( object ):
     check_interval = 15
     dataset_name_re = re.compile( '(dataset\d+)_(name)' )
@@ -36,7 +36,10 @@ def run_job( self, job ):
             if job.params[ 'protocol' ] in [ 'http', 'https' ]:
                 results = []
                 for result in job.params[ 'results' ].values():
-                    result[ 'transfer_job' ] = self.app.transfer_manager.new( protocol=job.params[ 'protocol' ], url=result[ 'url' ] )
+                    result[ 'transfer_job' ] = self.app.transfer_manager.new( protocol=job.params[ 'protocol' ],
+                                                                              name=result[ 'name' ],
+                                                                              datatype=result[ 'datatype' ],
+                                                                              url=result[ 'url' ] )
                     results.append( result )
             elif job.params[ 'protocol' ] == 'scp':
                 results = []
@@ -83,9 +86,9 @@ def run_job( self, job ):
             # Update the state of the relevant SampleDataset
             new_status = self.app.model.SampleDataset.transfer_status.ADD_TO_LIBRARY
             if protocol in [ 'http', 'https' ]:
-                result_dict = job.params[ 'results' ]
-                library_dataset_name = job.params[ 'result' ][ 'name' ]
-                extension = job.params[ 'result' ][ 'datatype' ]
+                result_dict = job.params[ 'result' ]
+                library_dataset_name = result_dict[ 'name' ]
+                extension = result_dict[ 'datatype' ]
             elif protocol in [ 'scp' ]:
                 # In this case, job.params will be a dictionary that contains a key named 'result'.  The value
                 # of the result key is a dictionary that looks something like:

lib/galaxy/jobs/deferred/pacific_biosciences_smrt_portal.py

@@ -3,6 +3,9 @@
 produced by SMRT Portal.
 """
 import logging, urllib2, re, shutil
+from string import Template
+from galaxy.util import json
+
 from data_transfer import *
 
 log = logging.getLogger( __name__ )
@@ -18,9 +21,16 @@ def create_job( self, trans, **kwd ):
             sample = kwd[ 'sample' ]
             smrt_job_id = kwd[ 'secondary_analysis_job_id' ]
             external_service = sample.request.type.get_external_service( 'pacific_biosciences_smrt_portal' )
+            external_service.load_data_transfer_settings( trans )
+            http_configs = external_service.data_transfer[ trans.model.ExternalService.data_transfer_protocol.HTTP ]
+            if not http_configs[ 'automatic_transfer' ]:
+                raise Exception( "Manual data transfer using http is not yet supported." )
             smrt_host = external_service.form_values.content[ 'host' ]
-            external_service_type_id = external_service.external_service_type_id
-            external_service_type = self.app.external_service_types.all_external_service_types[ external_service_type_id ]
+            external_service_type = external_service.get_external_service_type( trans )
+            # TODO: is there a better way to store the protocol?
+            # external_service_type.data_transfer looks somethng like
+            # {'http': <galaxy.sample_tracking.data_transfer.HttpDataTransferFactory object at 0x1064239d0>}
+            protocol = external_service_type.data_transfer.keys()[0]
             results = {}
             for k, v in external_service.form_values.content.items():
                 match = self.dataset_name_re.match( k ) or self.dataset_datatype_re.match( k )
@@ -42,6 +52,7 @@ def create_job( self, trans, **kwd ):
             self.sa_session.add(sample)
             self.sa_session.flush()
             params = { 'type' : 'init_transfer',
+                       'protocol' : protocol,
                        'sample_id' : sample.id,
                        'results' : results,
                        'smrt_host' : smrt_host,
@@ -50,6 +61,7 @@ def create_job( self, trans, **kwd ):
             self._associate_untransferred_datasets_with_sample( sample, external_service, results )
         elif 'transfer_job_id' in kwd:
             params = { 'type' : 'finish_transfer',
+                       'protocol' : kwd[ 'result' ][ 'protocol' ],
                        'sample_id' : kwd[ 'sample_id' ],
                        'result' : kwd[ 'result' ],
                        'transfer_job_id' : kwd[ 'transfer_job_id' ] }

lib/galaxy/jobs/transfer_manager.py

@@ -30,14 +30,13 @@ def new( self, path=None, **kwd ):
         if protocol in [ 'http', 'https' ]:
             if 'url' not in kwd:
                 raise Exception( 'Missing required parameter "url".' )
-            transfer_job = self.app.model.TransferJob( state=self.app.model.TransferJob.states.NEW, params=kwd )
         elif protocol == 'scp':
             # TODO: add more checks here?
             if 'sample_dataset_id' not in kwd:
                 raise Exception( 'Missing required parameter "sample_dataset_id".' )
             if 'file_path' not in kwd:
                 raise Exception( 'Missing required parameter "file_path".' )
-            transfer_job = self.app.model.TransferJob( state=self.app.model.TransferJob.states.NEW, params=kwd )
+        transfer_job = self.app.model.TransferJob( state=self.app.model.TransferJob.states.NEW, params=kwd )
         self.sa_session.add( transfer_job )
         self.sa_session.flush()
         return transfer_job

lib/galaxy/model/__init__.py

@@ -1825,6 +1825,11 @@ def load_data_transfer_settings( self, trans ):
                 scp_configs[ 'data_location' ] = self.form_values.content.get( data_transfer_obj.config.get( 'data_location', '' ), '' )
                 scp_configs[ 'rename_dataset' ] = self.form_values.content.get( data_transfer_obj.config.get( 'rename_dataset', '' ), '' )
                 self.data_transfer[ self.data_transfer_protocol.SCP ] = scp_configs
+            if data_transfer_protocol == self.data_transfer_protocol.HTTP:
+                http_configs = {}
+                automatic_transfer = data_transfer_obj.config.get( 'automatic_transfer', 'false' )
+                http_configs[ 'automatic_transfer' ] = util.string_as_bool( automatic_transfer )
+                self.data_transfer[ self.data_transfer_protocol.HTTP ] = http_configs
     def populate_actions( self, trans, item, param_dict=None ):
         return self.get_external_service_type( trans ).actions.populate( self, item, param_dict=param_dict )
 

lib/galaxy/sample_tracking/data_transfer.py

@@ -27,12 +27,23 @@ def parse( self, config_file, elem ):
         # Validate 
         for name, value in self.config.items():
             assert value, "'%s' attribute missing in 'data_transfer' element of type 'scp' in external_service_type xml config file: '%s'." % ( name, config_file )
-        
+
+class HttpDataTransferFactory( DataTransferFactory ):
+    type = 'http'
+    def __init__( self ):
+        pass
+    def parse( self, config_file, elem ):
+        self.config = {}
+        self.config[ 'automatic_transfer' ] = elem.get( 'automatic_transfer' )
+        # Validate 
+        for name, value in self.config.items():
+            assert value, "'%s' attribute missing in 'data_transfer' element of type 'http' in external_service_type xml config file: '%s'." % ( name, config_file )
+
 class FtpDataTransferFactory( DataTransferFactory ):
     type = 'ftp'
     def __init__( self ):
         pass
     def parse( self, elem ):
         pass
-    
-data_transfer_factories = dict( [ ( data_transfer.type, data_transfer() ) for data_transfer in [ ScpDataTransferFactory, FtpDataTransferFactory ] ] )
+
+data_transfer_factories = dict( [ ( data_transfer.type, data_transfer() ) for data_transfer in [ ScpDataTransferFactory, HttpDataTransferFactory, FtpDataTransferFactory ] ] )

lib/galaxy/sample_tracking/external_service_types.py

@@ -84,12 +84,16 @@ def parse( self, root ):
     def parse_data_transfer_settings( self, root ):
         self.data_transfer = {}
         data_transfer_settings_elem = root.find( 'data_transfer_settings' )
-        # Currently only data transfer using scp is supported.
+        # Currently only data transfer using scp or http is supported.
         for data_transfer_elem in data_transfer_settings_elem.findall( "data_transfer" ):
             if data_transfer_elem.get( 'protocol' ) == model.ExternalService.data_transfer_protocol.SCP:
                 scp_data_transfer = data_transfer_factories[ model.ExternalService.data_transfer_protocol.SCP ]
                 scp_data_transfer.parse( self.config_file, data_transfer_elem  )
                 self.data_transfer[ model.ExternalService.data_transfer_protocol.SCP ] = scp_data_transfer
+            if data_transfer_elem.get( 'protocol' ) == model.ExternalService.data_transfer_protocol.HTTP:
+                http_data_transfer = data_transfer_factories[ model.ExternalService.data_transfer_protocol.HTTP ]
+                http_data_transfer.parse( self.config_file, data_transfer_elem  )
+                self.data_transfer[ model.ExternalService.data_transfer_protocol.HTTP ] = http_data_transfer
     def parse_run_details( self, root ):
         self.run_details = {}
         run_details_elem = root.find( 'run_details' )

lib/galaxy/web/controllers/requests_admin.py

@@ -671,7 +671,7 @@ def initiate_data_transfer( self, trans, sample_id, sample_datasets=[], sample_d
                 if not scp_configs[ 'automatic_transfer' ]:
                     deferred_plugin = 'ManualDataTransferPlugin'
                 else:
-                    raise Exception( "Automatic data transfer using scp is not yet suppored." )
+                    raise Exception( "Automatic data transfer using scp is not yet supported." )
             trans.app.job_manager.deferred_job_queue.plugins[ deferred_plugin ].create_job( trans,
                                                                                             sample=sample,
                                                                                             sample_datasets=sample_datasets,