- galaxy-commits - lists.galaxyproject.org

commit/galaxy-central: martenson: allow JS modal to execute callbacks on hide()
by commits-noreply＠bitbucket.org 07 May '14

07 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/40054ac6f07b/ Changeset: 40054ac6f07b User: martenson Date: 2014-05-07 20:23:15 Summary: allow JS modal to execute callbacks on hide() Affected #: 2 files diff -r a73c02cd503ef2bec4819013517530df7045127e -r 40054ac6f07bce479affd6b39772640e9321e57b static/scripts/mvc/ui/ui-modal.js --- a/static/scripts/mvc/ui/ui-modal.js +++ b/static/scripts/mvc/ui/ui-modal.js @@ -7,12 +7,13 @@ // defaults options optionsDefault: { - title : 'ui-modal', - body : '', - backdrop : true, - height : null, - width : null, - closing_events : false + title : 'ui-modal', + body : '', + backdrop : true, + height : null, + width : null, + closing_events : false, + closing_callback : null }, // button list @@ -58,6 +59,9 @@ hide: function() { this.visible = false; this.$el.fadeOut('fast'); + if (this.options.closing_callback){ + this.options.closing_callback(); + } }, // enable buttons diff -r a73c02cd503ef2bec4819013517530df7045127e -r 40054ac6f07bce479affd6b39772640e9321e57b static/scripts/packed/mvc/ui/ui-modal.js --- a/static/scripts/packed/mvc/ui/ui-modal.js +++ b/static/scripts/packed/mvc/ui/ui-modal.js @@ -1,1 +1,1 @@ -define([],function(){var a=Backbone.View.extend({elMain:"body",optionsDefault:{title:"ui-modal",body:"",backdrop:true,height:null,width:null,closing_events:false},buttonList:{},initialize:function(b){if(b){this._create(b)}},show:function(b){this.initialize(b);if(this.options.height){this.$body.css("height",this.options.height);this.$body.css("overflow","hidden")}else{this.$body.css("max-height",$(window).height()/2)}if(this.options.width){this.$dialog.css("width",this.options.width)}if(this.visible){this.$el.show()}else{this.$el.fadeIn("fast")}this.visible=true},hide:function(){this.visible=false;this.$el.fadeOut("fast")},enableButton:function(b){var c=this.buttonList[b];this.$buttons.find("#"+c).prop("disabled",false)},disableButton:function(b){var c=this.buttonList[b];this.$buttons.find("#"+c).prop("disabled",true)},showButton:function(b){var c=this.buttonList[b];this.$buttons.find("#"+c).show()},hideButton:function(b){var c=this.buttonList[b];this.$buttons.find("#"+c).hide()},getButton:function(b){var c=this.buttonList[b];return this.$buttons.find("#"+c)},scrollTop:function(){return this.$body.scrollTop()},_create:function(d){var c=this;this.options=_.defaults(d,this.optionsDefault);if(this.options.body=="progress"){this.options.body=$('<div class="progress progress-striped active"><div class="progress-bar progress-bar-info" style="width:100%"></div></div>')}if(this.$el){this.$el.remove();$(document).off("keyup.ui-modal")}this.setElement(this._template(this.options.title));this.$dialog=(this.$el).find(".modal-dialog");this.$body=(this.$el).find(".modal-body");this.$footer=(this.$el).find(".modal-footer");this.$buttons=(this.$el).find(".buttons");this.$backdrop=(this.$el).find(".modal-backdrop");this.$body.html(this.options.body);if(!this.options.backdrop){this.$backdrop.removeClass("in")}if(this.options.buttons){this.buttonList={};var b=0;$.each(this.options.buttons,function(e,g){var f="button-"+b++;c.$buttons.append($('<button id="'+f+'"></button>').text(e).click(g)).append(" ");c.buttonList[e]=f})}else{this.$footer.hide()}$(this.elMain).append($(this.el));if(this.options.closing_events){$(document).on("keyup.ui-modal",function(f){if(f.keyCode==27){c.hide()}});this.$el.find(".modal-backdrop").on("click",function(){c.hide()})}},_template:function(b){return'<div class="modal"><div class="modal-backdrop fade in" style="z-index: -1;"></div><div class="modal-dialog"><div class="modal-content"><div class="modal-header"><button type="button" class="close" style="display: none;">×</button><h4 class="title">'+b+'</h4></div><div class="modal-body" style="position: static;"></div><div class="modal-footer"><div class="buttons" style="float: right;"></div></div></div</div></div>'}});return{View:a}}); \ No newline at end of file +define([],function(){var a=Backbone.View.extend({elMain:"body",optionsDefault:{title:"ui-modal",body:"",backdrop:true,height:null,width:null,closing_events:false,closing_callback:null},buttonList:{},initialize:function(b){if(b){this._create(b)}},show:function(b){this.initialize(b);if(this.options.height){this.$body.css("height",this.options.height);this.$body.css("overflow","hidden")}else{this.$body.css("max-height",$(window).height()/2)}if(this.options.width){this.$dialog.css("width",this.options.width)}if(this.visible){this.$el.show()}else{this.$el.fadeIn("fast")}this.visible=true},hide:function(){this.visible=false;this.$el.fadeOut("fast");if(this.options.closing_callback){this.options.closing_callback()}},enableButton:function(b){var c=this.buttonList[b];this.$buttons.find("#"+c).prop("disabled",false)},disableButton:function(b){var c=this.buttonList[b];this.$buttons.find("#"+c).prop("disabled",true)},showButton:function(b){var c=this.buttonList[b];this.$buttons.find("#"+c).show()},hideButton:function(b){var c=this.buttonList[b];this.$buttons.find("#"+c).hide()},getButton:function(b){var c=this.buttonList[b];return this.$buttons.find("#"+c)},scrollTop:function(){return this.$body.scrollTop()},_create:function(d){var c=this;this.options=_.defaults(d,this.optionsDefault);if(this.options.body=="progress"){this.options.body=$('<div class="progress progress-striped active"><div class="progress-bar progress-bar-info" style="width:100%"></div></div>')}if(this.$el){this.$el.remove();$(document).off("keyup.ui-modal")}this.setElement(this._template(this.options.title));this.$dialog=(this.$el).find(".modal-dialog");this.$body=(this.$el).find(".modal-body");this.$footer=(this.$el).find(".modal-footer");this.$buttons=(this.$el).find(".buttons");this.$backdrop=(this.$el).find(".modal-backdrop");this.$body.html(this.options.body);if(!this.options.backdrop){this.$backdrop.removeClass("in")}if(this.options.buttons){this.buttonList={};var b=0;$.each(this.options.buttons,function(e,g){var f="button-"+b++;c.$buttons.append($('<button id="'+f+'"></button>').text(e).click(g)).append(" ");c.buttonList[e]=f})}else{this.$footer.hide()}$(this.elMain).append($(this.el));if(this.options.closing_events){$(document).on("keyup.ui-modal",function(f){if(f.keyCode==27){c.hide()}});this.$el.find(".modal-backdrop").on("click",function(){c.hide()})}},_template:function(b){return'<div class="modal"><div class="modal-backdrop fade in" style="z-index: -1;"></div><div class="modal-dialog"><div class="modal-content"><div class="modal-header"><button type="button" class="close" style="display: none;">×</button><h4 class="title">'+b+'</h4></div><div class="modal-body" style="position: static;"></div><div class="modal-footer"><div class="buttons" style="float: right;"></div></div></div</div></div>'}});return{View:a}}); \ No newline at end of file Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: martenson: data libraries: API pep8, cleanup, repaired yet another encoding bug
by commits-noreply＠bitbucket.org 07 May '14

07 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/a73c02cd503e/ Changeset: a73c02cd503e User: martenson Date: 2014-05-07 20:10:30 Summary: data libraries: API pep8, cleanup, repaired yet another encoding bug Affected #: 2 files diff -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c -r a73c02cd503ef2bec4819013517530df7045127e lib/galaxy/webapps/galaxy/api/folder_contents.py --- a/lib/galaxy/webapps/galaxy/api/folder_contents.py +++ b/lib/galaxy/webapps/galaxy/api/folder_contents.py @@ -13,6 +13,7 @@ import logging log = logging.getLogger( __name__ ) + class FolderContentsController( BaseAPIController, UsesLibraryMixin, UsesLibraryMixinItems, UsesHistoryDatasetAssociationMixin ): """ Class controls retrieval, creation and updating of folder contents. @@ -23,14 +24,14 @@ """ GET /api/folders/{encoded_folder_id}/contents - Displays a collection (list) of a folder's contents + Displays a collection (list) of a folder's contents (files and folders). Encoded folder ID is prepended with 'F' if it is a folder as opposed to a data set which does not have it. Full path is provided in - response as a separate object providing data for + response as a separate object providing data for breadcrumb path building. - :param folder_id: encoded ID of the folder which + :param folder_id: encoded ID of the folder which contents should be library_dataset_dict :type folder_id: encoded string @@ -70,7 +71,7 @@ can_add_library_item = trans.user_is_admin() or trans.app.security_agent.can_add_library_item( current_user_roles, folder ) if not ( trans.user_is_admin() or trans.app.security_agent.can_access_library_item( current_user_roles, folder, trans.user ) ): - if folder.parent_id == None: + if folder.parent_id is None: try: library = trans.sa_session.query( trans.app.model.Library ).filter( trans.app.model.Library.table.c.root_folder_id == decoded_folder_id ).one() except Exception: @@ -82,14 +83,14 @@ log.warning( "SECURITY: User (id: %s) without proper access rights is trying to load folder with ID of %s" % ( trans.user.id, decoded_folder_id ) ) else: log.warning( "SECURITY: Anonymous user without proper access rights is trying to load folder with ID of %s" % ( decoded_folder_id ) ) - raise exceptions.ObjectNotFound( 'Folder with the id provided ( %s ) was not found' % str( folder_id ) ) + raise exceptions.ObjectNotFound( 'Folder with the id provided ( %s ) was not found' % str( folder_id ) ) else: if trans.user: log.warning( "SECURITY: User (id: %s) without proper access rights is trying to load folder with ID of %s" % ( trans.user.id, decoded_folder_id ) ) else: log.warning( "SECURITY: Anonymous user without proper access rights is trying to load folder with ID of %s" % ( decoded_folder_id ) ) raise exceptions.ObjectNotFound( 'Folder with the id provided ( %s ) was not found' % str( folder_id ) ) - + def build_path( folder ): """ Search the path upwards recursively and load the whole route of @@ -111,50 +112,45 @@ upper_folder = trans.sa_session.query( trans.app.model.LibraryFolder ).get( folder.parent_id ) path_to_root.extend( build_path( upper_folder ) ) return path_to_root - + # Return the reversed path so it starts with the library node. full_path = build_path( folder )[::-1] folder_contents = [] update_time = '' create_time = '' - # Go through every accessible item in the folder and include its meta-data. + # Go through every accessible item (folders, datasets) in the folder and include its meta-data. for content_item in self._load_folder_contents( trans, folder, deleted ): - can_access = trans.app.security_agent.can_access_library_item( current_user_roles, content_item, trans.user ) - if ( can_access or ( content_item.api_type == 'folder' and trans.app.security_agent.folder_is_unrestricted( content_item ) ) ): - return_item = {} - encoded_id = trans.security.encode_id( content_item.id ) - update_time = content_item.update_time.strftime( "%Y-%m-%d %I:%M %p" ) - create_time = content_item.create_time.strftime( "%Y-%m-%d %I:%M %p" ) + return_item = {} + encoded_id = trans.security.encode_id( content_item.id ) + update_time = content_item.update_time.strftime( "%Y-%m-%d %I:%M %p" ) + create_time = content_item.create_time.strftime( "%Y-%m-%d %I:%M %p" ) - # For folder return also hierarchy values - if content_item.api_type == 'folder': - encoded_id = 'F' + encoded_id + if content_item.api_type == 'folder': + encoded_id = 'F' + encoded_id - # this is commented for now as it includes the items that might not be accessible - # item_count = content_item.item_count if can_access else 0 - # return_item.update ( dict ( item_count = item_count ) ) + if content_item.api_type == 'file': + library_dataset_dict = content_item.to_dict() + library_dataset_dict[ 'is_unrestricted' ] = trans.app.security_agent.dataset_is_public( content_item.library_dataset_dataset_association.dataset ) + library_dataset_dict[ 'is_private' ] = trans.app.security_agent.dataset_is_private_to_user( trans, content_item ) - if content_item.api_type == 'file': - library_dataset_dict = content_item.to_dict() - library_dataset_dict['data_type'] - library_dataset_dict['file_size'] - library_dataset_dict['date_uploaded'] - return_item.update ( dict ( data_type = library_dataset_dict['data_type'], - file_size = library_dataset_dict['file_size'], - date_uploaded = library_dataset_dict['date_uploaded'] ) ) + return_item.update( dict( data_type=library_dataset_dict[ 'data_type' ], + file_size=library_dataset_dict[ 'file_size' ], + date_uploaded=library_dataset_dict[ 'date_uploaded' ], + is_unrestricted=library_dataset_dict[ 'is_unrestricted' ], + is_private=library_dataset_dict[ 'is_private' ] ) ) - # For every item return also the default meta-data - return_item.update( dict( id = encoded_id, - type = content_item.api_type, - name = content_item.name, - update_time = update_time, - create_time = create_time, - deleted = content_item.deleted - ) ) - folder_contents.append( return_item ) + # For every item include the default meta-data + return_item.update( dict( id=encoded_id, + type=content_item.api_type, + name=content_item.name, + update_time=update_time, + create_time=create_time, + deleted=content_item.deleted + ) ) + folder_contents.append( return_item ) - return { 'metadata' : { 'full_path' : full_path, 'can_add_library_item': can_add_library_item, 'folder_name': folder.name }, 'folder_contents' : folder_contents } + return { 'metadata': { 'full_path': full_path, 'can_add_library_item': can_add_library_item, 'folder_name': folder.name }, 'folder_contents': folder_contents } def _load_folder_contents( self, trans, folder, include_deleted ): """ @@ -166,7 +162,7 @@ :type folder: Galaxy LibraryFolder :param include_deleted: flag, when true the items that are deleted - and can be undeleted by current user are shown + and can be undeleted by current user are shown :type include_deleted: boolean :returns: a list containing the requested items @@ -175,34 +171,42 @@ current_user_roles = trans.get_current_user_roles() is_admin = trans.user_is_admin() content_items = [] - for subfolder in folder.active_folders: + for subfolder in folder.folders: if subfolder.deleted: if include_deleted: if is_admin: + # Admins can see all deleted folders. subfolder.api_type = 'folder' content_items.append( subfolder ) else: + # Users with MODIFY permissions can see deleted folders. can_modify = trans.app.security_agent.can_modify_library_item( current_user_roles, subfolder ) if can_modify: subfolder.api_type = 'folder' content_items.append( subfolder ) else: - if is_admin: - subfolder.api_type = 'folder' - content_items.append( subfolder ) - else: - can_access, folder_ids = trans.app.security_agent.check_folder_contents( trans.user, current_user_roles, subfolder ) - if can_access: - subfolder.api_type = 'folder' - content_items.append( subfolder ) + # Undeleted folders are non-restricted for now. The contents are not. + # TODO decide on restrictions + subfolder.api_type = 'folder' + content_items.append( subfolder ) + # if is_admin: + # subfolder.api_type = 'folder' + # content_items.append( subfolder ) + # else: + # can_access, folder_ids = trans.app.security_agent.check_folder_contents( trans.user, current_user_roles, subfolder ) + # if can_access: + # subfolder.api_type = 'folder' + # content_items.append( subfolder ) for dataset in folder.datasets: if dataset.deleted: if include_deleted: if is_admin: + # Admins can see all deleted datasets. dataset.api_type = 'file' content_items.append( dataset ) else: + # Users with MODIFY permissions on the item can see the deleted item. can_modify = trans.app.security_agent.can_modify_library_item( current_user_roles, dataset ) if can_modify: dataset.api_type = 'file' @@ -230,19 +234,19 @@ :type payload: dict * folder_id: the parent folder of the new item - * from_hda_id: (optional) the id of an accessible HDA to copy + * from_hda_id: (optional) the id of an accessible HDA to copy into the library * ldda_message: (optional) the new message attribute of the LDDA created - * extended_metadata: (optional) dub-dictionary containing any + * extended_metadata: (optional) dub-dictionary containing any extended metadata to associate with the item - :returns: a dictionary containing the id, name, + :returns: a dictionary containing the id, name, and 'show' url of the new item :rtype: dict - :raises: ObjectAttributeInvalidException, - InsufficientPermissionsException, ItemAccessibilityException, + :raises: ObjectAttributeInvalidException, + InsufficientPermissionsException, ItemAccessibilityException, InternalServerError """ encoded_folder_id_16 = self.__decode_library_content_id( trans, encoded_folder_id ) @@ -287,13 +291,13 @@ :param encoded_folder_id: encoded id of Galaxy LibraryFolder :type encoded_folder_id: encoded string - :returns: last 16 chars of the encoded id in case it was Folder + :returns: last 16 chars of the encoded id in case it was Folder (had 'F' prepended) :type: string :raises: MalformedId """ - if ( len( encoded_folder_id ) == 17 and encoded_folder_id.startswith( 'F' )): + if ( len( encoded_folder_id ) == 17 and encoded_folder_id.startswith( 'F' )): return encoded_folder_id[1:] else: raise exceptions.MalformedId( 'Malformed folder id ( %s ) specified, unable to decode.' % str( encoded_folder_id ) ) diff -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c -r a73c02cd503ef2bec4819013517530df7045127e lib/galaxy/webapps/galaxy/api/folders.py --- a/lib/galaxy/webapps/galaxy/api/folders.py +++ b/lib/galaxy/webapps/galaxy/api/folders.py @@ -1,18 +1,24 @@ """ API operations on library folders """ -import os, string, shutil, urllib, re, socket, traceback -from galaxy import datatypes, jobs, web, security +# import os +# import shutil +# import urllib +# import re +# import socket +# import traceback +# import string +from galaxy import web from galaxy import exceptions from galaxy.web import _future_expose_api as expose_api -from galaxy.web import _future_expose_api_anonymous as expose_api_anonymous -from galaxy.web.base.controller import BaseAPIController,UsesLibraryMixin,UsesLibraryMixinItems +from galaxy.web.base.controller import BaseAPIController, UsesLibraryMixin, UsesLibraryMixinItems from sqlalchemy.orm.exc import MultipleResultsFound from sqlalchemy.orm.exc import NoResultFound import logging log = logging.getLogger( __name__ ) + class FoldersController( BaseAPIController, UsesLibraryMixin, UsesLibraryMixinItems ): @web.expose_api @@ -32,18 +38,19 @@ Displays information about a folder. - :param encoded_parent_folder_id: the parent folder's id (required) - :type encoded_parent_folder_id: an encoded id string (should be prefixed by 'F') + :param id: the folder's encoded id (required) + :type id: an encoded id string (has to be prefixed by 'F') + + :returns: dictionary including details of the folder + :rtype: dict """ - # Eliminate any 'F' in front of the folder id. Just take the - # last 16 characters: - if ( len( id ) >= 17 ): - id = id[-16:] - # Retrieve the folder and return its contents encoded. Note that the - # check_ownership=false since we are only displaying it. - content = self.get_library_folder( trans, id, check_ownership=False, - check_accessible=True ) - return self.encode_all_ids( trans, content.to_dict( view='element' ) ) + folder_id_without_prefix = self.__cut_the_prefix( id ) + content = self.get_library_folder( trans, folder_id_without_prefix, check_ownership=False, check_accessible=True ) + return_dict = self.encode_all_ids( trans, content.to_dict( view='element' ) ) + return_dict[ 'id' ] = 'F' + return_dict[ 'id' ] + if return_dict[ 'parent_id' ] is not None: + return_dict[ 'parent_id' ] = 'F' + return_dict[ 'parent_id' ] + return return_dict @expose_api def create( self, trans, encoded_parent_folder_id, **kwd ): @@ -53,13 +60,13 @@ *POST /api/folders/{encoded_parent_folder_id} Create a new folder object underneath the one specified in the parameters. - + :param encoded_parent_folder_id: the parent folder's id (required) - :type encoded_parent_folder_id: an encoded id string (should be prefixed by 'F') + :type encoded_parent_folder_id: an encoded id string (should be prefixed by 'F') :param name: the name of the new folder (required) - :type name: str - + :type name: str + :param description: the description of the new folder :type description: str @@ -70,16 +77,16 @@ """ payload = kwd.get( 'payload', None ) - if payload == None: + if payload is None: raise exceptions.RequestParameterMissingException( "Missing required parameters 'encoded_parent_folder_id' and 'name'." ) name = payload.get( 'name', None ) description = payload.get( 'description', '' ) - if encoded_parent_folder_id == None: + if encoded_parent_folder_id is None: raise exceptions.RequestParameterMissingException( "Missing required parameter 'encoded_parent_folder_id'." ) - elif name == None: + elif name is None: raise exceptions.RequestParameterMissingException( "Missing required parameter 'name'." ) - # encoded_parent_folder_id may be prefixed by 'F' + # encoded_parent_folder_id should be prefixed by 'F' encoded_parent_folder_id = self.__cut_the_prefix( encoded_parent_folder_id ) try: decoded_parent_folder_id = trans.security.decode_id( encoded_parent_folder_id ) @@ -93,7 +100,7 @@ except NoResultFound: raise exceptions.RequestParameterInvalidException( 'No folder found with the id provided.' ) except Exception, e: - raise exceptions.InternalServerError( 'Error loading from the database.' + str(e)) + raise exceptions.InternalServerError( 'Error loading from the database.' + str( e ) ) library = parent_folder.parent_library if library.deleted: @@ -108,29 +115,28 @@ try: folder = trans.sa_session.query( trans.app.model.LibraryFolder ).get( v.id ) except Exception, e: - raise exceptions.InternalServerError( 'Error loading from the database.' + str( e )) + raise exceptions.InternalServerError( 'Error loading from the database.' + str( e )) if folder: - return_dict = folder.to_dict( view='element' ) - return_dict[ 'parent_library_id' ] = trans.security.encode_id( return_dict[ 'parent_library_id' ] ) - return_dict[ 'parent_id' ] = 'F' + trans.security.encode_id( return_dict[ 'parent_id' ] ) - return_dict[ 'id' ] = 'F' + trans.security.encode_id( return_dict[ 'id' ] ) + update_time = folder.update_time.strftime( "%Y-%m-%d %I:%M %p" ) + return_dict = self.encode_all_ids( trans, folder.to_dict( view='element' ) ) + return_dict[ 'update_time' ] = update_time + return_dict[ 'parent_id' ] = 'F' + return_dict[ 'parent_id' ] + return_dict[ 'id' ] = 'F' + return_dict[ 'id' ] return return_dict else: - raise exceptions.InternalServerError( 'Error while creating a folder.' + str(e)) + raise exceptions.InternalServerError( 'Error while creating a folder.' + str( e ) ) @web.expose_api def update( self, trans, id, library_id, payload, **kwd ): """ PUT /api/folders/{encoded_folder_id} - For now this does nothing. There are no semantics for folders that - indicates that an update operation is needed; the existing - library_contents folder does not allow for update, either. + """ - pass + raise exceptions.NotImplemented( 'Updating folder through this endpoint is not implemented yet.' ) def __cut_the_prefix(self, encoded_id): - if ( len( encoded_id ) == 17 and encoded_id.startswith( 'F' ) ): - return encoded_id[-16:] + + if ( ( len( encoded_id ) % 2 == 1 ) and encoded_id.startswith( 'F' ) ): + return encoded_id[ 1: ] else: - return encoded_id - + raise exceptions.MalformedId( 'Malformed folder id ( %s ) specified, unable to decode.' % str( encoded_id ) ) Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: jmchilton: Merged in jmchilton/galaxy-central-fork-1 (pull request #385)
by commits-noreply＠bitbucket.org 07 May '14

07 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/9eb2b42e21fe/ Changeset: 9eb2b42e21fe User: jmchilton Date: 2014-05-07 18:57:21 Summary: Merged in jmchilton/galaxy-central-fork-1 (pull request #385) CLI Job Runner Enhancements Affected #: 14 files diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/cli.py --- a/lib/galaxy/jobs/runners/cli.py +++ b/lib/galaxy/jobs/runners/cli.py @@ -3,18 +3,18 @@ """ import os -import time -import glob import logging from galaxy import model from galaxy.jobs import JobDestination from galaxy.jobs.runners import AsynchronousJobState, AsynchronousJobRunner +from .util.cli import CliInterface, split_params log = logging.getLogger( __name__ ) __all__ = [ 'ShellJobRunner' ] + class ShellJobRunner( AsynchronousJobRunner ): """ Job runner backed by a finite pool of worker threads. FIFO scheduling @@ -25,53 +25,27 @@ """Start the job runner """ super( ShellJobRunner, self ).__init__( app, nworkers ) - self.cli_shells = None - self.cli_job_interfaces = None - self.__load_cli_plugins() - + self.cli_interface = CliInterface() self._init_monitor_thread() self._init_worker_threads() - def __load_cli_plugins(self): - def __load(module_path, d): - for file in glob.glob(os.path.join(os.path.join(os.getcwd(), 'lib', *module_path.split('.')), '*.py')): - if os.path.basename(file).startswith('_'): - continue - module_name = '%s.%s' % (module_path, os.path.basename(file).rsplit('.py', 1)[0]) - module = __import__(module_name) - for comp in module_name.split( "." )[1:]: - module = getattr(module, comp) - for name in module.__all__: - log.debug('Loaded cli plugin %s' % name) - d[name] = getattr(module, name) - - self.cli_shells = {} - self.cli_job_interfaces = {} - __load('galaxy.jobs.runners.cli_shell', self.cli_shells) - __load('galaxy.jobs.runners.cli_job', self.cli_job_interfaces) - def get_cli_plugins( self, shell_params, job_params ): - # load shell plugin - shell = self.cli_shells[shell_params['plugin']](**shell_params) - job_interface = self.cli_job_interfaces[job_params['plugin']](**job_params) - return shell, job_interface + return self.cli_interface.get_plugins( shell_params, job_params ) def url_to_destination( self, url ): params = {} - shell_params, job_params = url.split('/')[2:4] + shell_params, job_params = url.split( '/' )[ 2:4 ] # split 'foo=bar&baz=quux' into { 'foo' : 'bar', 'baz' : 'quux' } - shell_params = dict ( [ ( 'shell_' + k, v ) for k, v in [ kv.split('=', 1) for kv in shell_params.split('&') ] ] ) - job_params = dict ( [ ( 'job_' + k, v ) for k, v in [ kv.split('=', 1) for kv in job_params.split('&') ] ] ) + shell_params = dict( [ ( 'shell_' + k, v ) for k, v in [ kv.split( '=', 1 ) for kv in shell_params.split( '&' ) ] ] ) + job_params = dict( [ ( 'job_' + k, v ) for k, v in [ kv.split( '=', 1 ) for kv in job_params.split( '&' ) ] ] ) params.update( shell_params ) params.update( job_params ) - log.debug("Converted URL '%s' to destination runner=cli, params=%s" % (url, params)) + log.debug( "Converted URL '%s' to destination runner=cli, params=%s" % ( url, params ) ) # Create a dynamic JobDestination - return JobDestination(runner='cli', params=params) + return JobDestination( runner='cli', params=params ) def parse_destination_params( self, params ): - shell_params = dict((k.replace('shell_', '', 1), v) for k, v in params.items() if k.startswith('shell_')) - job_params = dict((k.replace('job_', '', 1), v) for k, v in params.items() if k.startswith('job_')) - return shell_params, job_params + return split_params( params ) def queue_job( self, job_wrapper ): """Create job script and submit it to the DRM""" @@ -93,8 +67,12 @@ # define job attributes ajs = AsynchronousJobState( files_dir=job_wrapper.working_directory, job_wrapper=job_wrapper ) - # fill in the DRM's job run template - script = job_interface.get_job_template(ajs.output_file, ajs.error_file, ajs.job_name, job_wrapper, command_line, ajs.exit_code_file) + job_file_kwargs = job_interface.job_script_kwargs(ajs.output_file, ajs.error_file, ajs.job_name) + script = self.get_job_file( + job_wrapper, + exit_code_path=ajs.exit_code_file, + **job_file_kwargs + ) try: fh = file(ajs.job_file, "w") diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/cli_job/__init__.py --- a/lib/galaxy/jobs/runners/cli_job/__init__.py +++ /dev/null @@ -1,21 +0,0 @@ -""" -Base class for cli job plugins -""" - -class BaseJobExec(object): - def __init__(self, **params): - raise NotImplementedError() - def get_job_template(self, ofile, efile, job_name, job_wrapper, command_line, ecfile): - raise NotImplementedError() - def submit(self, script_file): - raise NotImplementedError() - def delete(self, job_id): - raise NotImplementedError() - def get_status(self, job_ids=None): - raise NotImplementedError() - def get_single_status(self, job_id): - raise NotImplementedError() - def parse_status(self, status, job_ids): - raise NotImplementedError() - def parse_single_status(self, status, job_id): - raise NotImplementedError() diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/cli_job/torque.py --- a/lib/galaxy/jobs/runners/cli_job/torque.py +++ /dev/null @@ -1,135 +0,0 @@ -""" -Command-line interface to TORQUE PBS -""" - -import os -import logging - -from galaxy.model import Job -job_states = Job.states - -from galaxy.jobs.runners.cli_job import BaseJobExec - -log = logging.getLogger( __name__ ) - -__all__ = ('Torque',) - -try: - import xml.etree.cElementTree as et -except: - import xml.etree.ElementTree as et - -job_template = """#!/bin/sh -%s -GALAXY_LIB="%s" -if [ "$GALAXY_LIB" != "None" ]; then - if [ -n "$PYTHONPATH" ]; then - export PYTHONPATH="$GALAXY_LIB:$PYTHONPATH" - else - export PYTHONPATH="$GALAXY_LIB" - fi -fi -%s -cd %s -%s -echo $? > %s -""" - -argmap = { 'destination' : '-q', - 'Execution_Time' : '-a', - 'Account_Name' : '-A', - 'Checkpoint' : '-c', - 'Error_Path' : '-e', - 'Group_List' : '-g', - 'Hold_Types' : '-h', - 'Join_Paths' : '-j', - 'Keep_Files' : '-k', - 'Resource_List' : '-l', - 'Mail_Points' : '-m', - 'Mail_Users' : '-M', - 'Job_Name' : '-N', - 'Output_Path' : '-o', - 'Priority' : '-p', - 'Rerunable' : '-r', - 'Shell_Path_List' : '-S', - 'job_array_request' : '-t', - 'User_List' : '-u', - 'Variable_List' : '-v' } - -class Torque(BaseJobExec): - def __init__(self, **params): - self.params = {} - for k, v in params.items(): - self.params[k] = v - - def get_job_template(self, ofile, efile, job_name, job_wrapper, command_line, ecfile): - pbsargs = { '-o' : ofile, - '-e' : efile, - '-N' : job_name } - for k, v in self.params.items(): - if k == 'plugin': - continue - try: - if not k.startswith('-'): - k = argmap[k] - pbsargs[k] = v - except: - log.warning('Unrecognized long argument passed to Torque CLI plugin: %s' % k) - template_pbsargs = '' - for k, v in pbsargs.items(): - template_pbsargs += '#PBS %s %s\n' % (k, v) - return job_template % (template_pbsargs, - job_wrapper.galaxy_lib_dir, - job_wrapper.get_env_setup_clause(), - os.path.abspath(job_wrapper.working_directory), - command_line, - ecfile) - - def submit(self, script_file): - return 'qsub %s' % script_file - - def delete(self, job_id): - return 'qdel %s' % job_id - - def get_status(self, job_ids=None): - return 'qstat -x' - - def get_single_status(self, job_id): - return 'qstat -f %s' % job_id - - def parse_status(self, status, job_ids): - # in case there's noise in the output, find the big blob 'o xml - tree = None - rval = {} - for line in status.strip().splitlines(): - try: - tree = et.fromstring(line.strip()) - assert tree.tag == 'Data' - break - except Exception, e: - tree = None - if tree is None: - log.warning('No valid qstat XML return from `qstat -x`, got the following: %s' % status) - return None - else: - for job in tree.findall('Job'): - id = job.find('Job_Id').text - if id in job_ids: - state = job.find('job_state').text - # map PBS job states to Galaxy job states. - rval[id] = self.__get_job_state(state) - return rval - - def parse_single_status(self, status, job_id): - for line in status.splitlines(): - line = line.split(' = ') - if line[0] == 'job_state': - return line[1] - # no state found, job has exited - return job_states.OK - - def __get_job_state(self, state): - return { 'E' : job_states.RUNNING, - 'R' : job_states.RUNNING, - 'Q' : job_states.QUEUED, - 'C' : job_states.OK }.get(state, state) diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/cli_shell/__init__.py --- a/lib/galaxy/jobs/runners/cli_shell/__init__.py +++ /dev/null @@ -1,11 +0,0 @@ -""" -Base class for runners which execute commands via a shell -""" - -class BaseShellExec(object): - def __init__(self, *args, **kwargs): - raise NotImplementedError() - def copy(self, rcp_cmd, files, dest): - raise NotImplementedError() - def execute(self, cmd, persist=False, timeout=60): - raise NotImplementedError() diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/cli_shell/rsh.py --- a/lib/galaxy/jobs/runners/cli_shell/rsh.py +++ /dev/null @@ -1,64 +0,0 @@ -""" -Interface for remote shell commands (rsh, rcp) and derivatives that use the same syntax (ssh, scp) -""" - -import logging -import os -import subprocess -import tempfile -import time - -from galaxy.util.bunch import Bunch -from galaxy.jobs.runners.cli_shell import BaseShellExec - -log = logging.getLogger( __name__ ) - -__all__ = ('RemoteShell', 'SecureShell', 'GlobusSecureShell') - -class RemoteShell(BaseShellExec): - def __init__(self, rsh='rsh', rcp='rcp', hostname=None, username=None, **kwargs): - self.rsh = rsh - self.rcp = rcp - self.hostname = hostname - self.username = username - self.sessions = {} - def copy(self, rcp_cmd, files, dest): - pass - def execute(self, cmd, persist=False, timeout=60): - # TODO: implement persistence - if self.username is None: - fullcmd = '%s %s %s' % (self.rsh, self.hostname, cmd) - else: - fullcmd = '%s -l %s %s %s' % (self.rsh, self.username, self.hostname, cmd) - # Read stdout to a tempfile in case it's large (>65K) - outf = tempfile.TemporaryFile() - p = subprocess.Popen(fullcmd, shell=True, stdin=None, stdout=outf, stderr=subprocess.PIPE) - # poll until timeout - for i in range(timeout/3): - r = p.poll() - if r is not None: - break - time.sleep(3) - else: - pid = int(p.pid) - for sig in (15, 9): - try: - os.kill(pid, sig) - time.sleep(3) - except: - log.warning('Killing pid %s (cmd: "%s") with signal %s failed' % (p.pid, fullcmd, sig)) - return Bunch(stdout='', stderr='Execution timed out', returncode=-1) - outf.seek(0) - return Bunch(stdout=outf.read(), stderr=p.stderr.read(), returncode=p.returncode) - - -class SecureShell(RemoteShell): - SSH_NEW_KEY_STRING = 'Are you sure you want to continue connecting' - def __init__(self, rsh='ssh', rcp='scp', **kwargs): - rsh += ' -oStrictHostKeyChecking=yes -oConnectTimeout=60' - rcp += ' -oStrictHostKeyChecking=yes -oConnectTimeout=60' - super(SecureShell, self).__init__(rsh=rsh, rcp=rcp, **kwargs) - -class GlobusSecureShell(SecureShell): - def __init__(self, rsh='gsissh', rcp='gsiscp', **kwargs): - super(SecureShell, self).__init__(rsh=rsh, rcp=rcp, **kwargs) diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/__init__.py --- a/lib/galaxy/jobs/runners/util/__init__.py +++ b/lib/galaxy/jobs/runners/util/__init__.py @@ -3,5 +3,8 @@ processes and interfacing with job managers. This module should contain functionality shared between Galaxy and the LWR. """ +from galaxy.util.bunch import Bunch -from galaxy.util.bunch import Bunch +from .kill import kill_pid + +__all__ = [kill_pid, Bunch] diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/__init__.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/__init__.py @@ -0,0 +1,56 @@ +""" +""" +from glob import glob +from os.path import basename, join +from os import getcwd + +DEFAULT_SHELL_PLUGIN = 'LocalShell' + + +class CliInterface(object): + """ + High-level interface for loading shell and job plugins and matching + them to specified parameters. + """ + + def __init__(self, code_dir='lib'): + """ + """ + def __load(module_path, d): + module_pattern = join(join(getcwd(), code_dir, *module_path.split('.')), '*.py') + for file in glob(module_pattern): + if basename(file).startswith('_'): + continue + module_name = '%s.%s' % (module_path, basename(file).rsplit('.py', 1)[0]) + module = __import__(module_name) + for comp in module_name.split(".")[1:]: + module = getattr(module, comp) + for name in module.__all__: + try: + d[name] = getattr(module, name) + except TypeError: + raise TypeError("Invalid type for name %s" % name) + + self.cli_shells = {} + self.cli_job_interfaces = {} + + module_prefix = self.__module__ + __load('%s.shell' % module_prefix, self.cli_shells) + __load('%s.job' % module_prefix, self.cli_job_interfaces) + + def get_plugins(self, shell_params, job_params): + """ + Return shell and job interface defined by and configured via + specified params. + """ + shell_plugin = shell_params.get('plugin', DEFAULT_SHELL_PLUGIN) + job_plugin = job_params['plugin'] + shell = self.cli_shells[shell_plugin](**shell_params) + job_interface = self.cli_job_interfaces[job_plugin](**job_params) + return shell, job_interface + + +def split_params(params): + shell_params = dict((k.replace('shell_', '', 1), v) for k, v in params.items() if k.startswith('shell_')) + job_params = dict((k.replace('job_', '', 1), v) for k, v in params.items() if k.startswith('job_')) + return shell_params, job_params diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/job/__init__.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/job/__init__.py @@ -0,0 +1,58 @@ +""" +Abstract base class for cli job plugins. +""" +from abc import ABCMeta, abstractmethod + + +class BaseJobExec(object): + __metaclass__ = ABCMeta + + @abstractmethod + def __init__(self, **params): + """ + Constructor for CLI job executor. + """ + + def job_script_kwargs(self, ofile, efile, job_name): + """ Return extra keyword argument for consumption by job script + module. + """ + return {} + + @abstractmethod + def submit(self, script_file): + """ + Given specified script_file path, yield command to submit it + to external job manager. + """ + + @abstractmethod + def delete(self, job_id): + """ + Given job id, return command to stop execution or dequeue specified + job. + """ + + @abstractmethod + def get_status(self, job_ids=None): + """ + Return command to get statuses of specified job ids. + """ + + @abstractmethod + def get_single_status(self, job_id): + """ + Return command to get the status of a single, specified job. + """ + + @abstractmethod + def parse_status(self, status, job_ids): + """ + Parse the statuses of output from get_status command. + """ + + @abstractmethod + def parse_single_status(self, status, job_id): + """ + Parse the status of output from get_single_status command. + """ diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/job/slurm_torque.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/job/slurm_torque.py @@ -0,0 +1,29 @@ +import re +from .torque import Torque + +__all__ = ('SlurmTorque',) + + +class SlurmTorque(Torque): + """ A CLI job executor for Slurm's Torque compatibility mode. This differs + from real torque CLI in that -x command line is not available so job status + needs to be parsed from qstat table instead of XML. + """ + + def get_status(self, job_ids=None): + return 'qstat' + + def parse_status(self, status, job_ids): + rval = {} + for line in status.strip().splitlines(): + if line.startswith("Job ID"): + continue + line_parts = re.compile("\s+").split(line) + if len(line_parts) < 5: + continue + id = line_parts[0] + state = line_parts[4] + if id in job_ids: + # map PBS job states to Galaxy job states. + rval[id] = self._get_job_state(state) + return rval diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/job/torque.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/job/torque.py @@ -0,0 +1,120 @@ +try: + import xml.etree.cElementTree as et +except: + import xml.etree.ElementTree as et + +try: + from galaxy.model import Job + job_states = Job.states +except ImportError: + # Not in Galaxy, map Galaxy job states to LWR ones. + from galaxy.util import enum + job_states = enum(RUNNING='running', OK='complete', QUEUED='queued') + +from ..job import BaseJobExec + +__all__ = ('Torque',) + +from logging import getLogger +log = getLogger(__name__) + +argmap = {'destination': '-q', + 'Execution_Time': '-a', + 'Account_Name': '-A', + 'Checkpoint': '-c', + 'Error_Path': '-e', + 'Group_List': '-g', + 'Hold_Types': '-h', + 'Join_Paths': '-j', + 'Keep_Files': '-k', + 'Resource_List': '-l', + 'Mail_Points': '-m', + 'Mail_Users': '-M', + 'Job_Name': '-N', + 'Output_Path': '-o', + 'Priority': '-p', + 'Rerunable': '-r', + 'Shell_Path_List': '-S', + 'job_array_request': '-t', + 'User_List': '-u', + 'Variable_List': '-v'} + + +class Torque(BaseJobExec): + + def __init__(self, **params): + self.params = {} + for k, v in params.items(): + self.params[k] = v + + def job_script_kwargs(self, ofile, efile, job_name): + pbsargs = {'-o': ofile, + '-e': efile, + '-N': job_name} + for k, v in self.params.items(): + if k == 'plugin': + continue + try: + if not k.startswith('-'): + k = argmap[k] + pbsargs[k] = v + except: + log.warning('Unrecognized long argument passed to Torque CLI plugin: %s' % k) + template_pbsargs = '' + for k, v in pbsargs.items(): + template_pbsargs += '#PBS %s %s\n' % (k, v) + return dict(headers=template_pbsargs) + + def submit(self, script_file): + return 'qsub %s' % script_file + + def delete(self, job_id): + return 'qdel %s' % job_id + + def get_status(self, job_ids=None): + return 'qstat -x' + + def get_single_status(self, job_id): + return 'qstat -f %s' % job_id + + def parse_status(self, status, job_ids): + # in case there's noise in the output, find the big blob 'o xml + tree = None + rval = {} + for line in status.strip().splitlines(): + try: + tree = et.fromstring(line.strip()) + assert tree.tag == 'Data' + break + except Exception: + tree = None + if tree is None: + log.warning('No valid qstat XML return from `qstat -x`, got the following: %s' % status) + return None + else: + for job in tree.findall('Job'): + id = job.find('Job_Id').text + if id in job_ids: + state = job.find('job_state').text + # map PBS job states to Galaxy job states. + rval[id] = self._get_job_state(state) + return rval + + def parse_single_status(self, status, job_id): + for line in status.splitlines(): + line = line.split(' = ') + if line[0] == 'job_state': + return self._get_job_state(line[1].strip()) + # no state found, job has exited + return job_states.OK + + def _get_job_state(self, state): + try: + return { + 'E': job_states.RUNNING, + 'R': job_states.RUNNING, + 'Q': job_states.QUEUED, + 'C': job_states.OK + }.get(state) + except KeyError: + raise KeyError("Failed to map torque status code [%s] to job state." % state) diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/shell/__init__.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/shell/__init__.py @@ -0,0 +1,19 @@ +""" +Abstract base class for runners which execute commands via a shell. +""" +from abc import ABCMeta, abstractmethod + + +class BaseShellExec(object): + __metaclass__ = ABCMeta + + @abstractmethod + def __init__(self, *args, **kwargs): + """ + Constructor for shell executor instance. + """ + + def execute(self, cmd, persist=False, timeout=60): + """ + Execute the specified command via defined shell. + """ diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/shell/local.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/shell/local.py @@ -0,0 +1,61 @@ +from tempfile import TemporaryFile +from time import sleep +from subprocess import Popen, PIPE + +from ..shell import BaseShellExec +from ....util import Bunch, kill_pid + +from logging import getLogger +log = getLogger(__name__) + +TIMEOUT_ERROR_MESSAGE = u'Execution timed out' +TIMEOUT_RETURN_CODE = -1 +DEFAULT_TIMEOUT = 60 +DEFAULT_TIMEOUT_CHECK_INTERVAL = 3 + + +class LocalShell(BaseShellExec): + """ + + >>> shell = LocalShell() + >>> def exec_python(script, **kwds): return shell.execute('python -c "%s"' % script, **kwds) + >>> exec_result = exec_python("from __future__ import print_function; print('Hello World')") + >>> exec_result.stderr == u'' + True + >>> exec_result.stdout.strip() == u'Hello World' + True + >>> exec_result = exec_python("import time; time.sleep(90)", timeout=3, timeout_check_interval=1) + >>> exec_result.stdout == u'' + True + >>> exec_result.stderr == 'Execution timed out' + True + >>> exec_result.returncode == TIMEOUT_RETURN_CODE + True + """ + + def __init__(self, **kwds): + pass + + def execute(self, cmd, persist=False, timeout=DEFAULT_TIMEOUT, timeout_check_interval=DEFAULT_TIMEOUT_CHECK_INTERVAL, **kwds): + outf = TemporaryFile() + p = Popen(cmd, shell=True, stdin=None, stdout=outf, stderr=PIPE) + # poll until timeout + + for i in range(int(timeout / timeout_check_interval)): + r = p.poll() + if r is not None: + break + sleep(timeout_check_interval) + else: + kill_pid(p.pid) + return Bunch(stdout=u'', stderr=TIMEOUT_ERROR_MESSAGE, returncode=TIMEOUT_RETURN_CODE) + outf.seek(0) + return Bunch(stdout=_read_str(outf), stderr=_read_str(p.stderr), returncode=p.returncode) + + +def _read_str(stream): + contents = stream.read() + return contents.decode('UTF-8') if isinstance(contents, bytes) else contents + + +__all__ = ('LocalShell',) diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/shell/rsh.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/shell/rsh.py @@ -0,0 +1,40 @@ +from .local import LocalShell + +from logging import getLogger +log = getLogger(__name__) + +__all__ = ('RemoteShell', 'SecureShell', 'GlobusSecureShell') + + +class RemoteShell(LocalShell): + + def __init__(self, rsh='rsh', rcp='rcp', hostname='localhost', username=None, **kwargs): + super(RemoteShell, self).__init__(**kwargs) + self.rsh = rsh + self.rcp = rcp + self.hostname = hostname + self.username = username + self.sessions = {} + + def execute(self, cmd, persist=False, timeout=60): + # TODO: implement persistence + if self.username is None: + fullcmd = '%s %s %s' % (self.rsh, self.hostname, cmd) + else: + fullcmd = '%s -l %s %s %s' % (self.rsh, self.username, self.hostname, cmd) + return super(RemoteShell, self).execute(fullcmd, persist, timeout) + + +class SecureShell(RemoteShell): + SSH_NEW_KEY_STRING = 'Are you sure you want to continue connecting' + + def __init__(self, rsh='ssh', rcp='scp', **kwargs): + rsh += ' -oStrictHostKeyChecking=yes -oConnectTimeout=60' + rcp += ' -oStrictHostKeyChecking=yes -oConnectTimeout=60' + super(SecureShell, self).__init__(rsh=rsh, rcp=rcp, **kwargs) + + +class GlobusSecureShell(SecureShell): + + def __init__(self, rsh='gsissh', rcp='gsiscp', **kwargs): + super(GlobusSecureShell, self).__init__(rsh=rsh, rcp=rcp, **kwargs) diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/kill.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/kill.py @@ -0,0 +1,65 @@ +import os +from platform import system +from time import sleep +from subprocess import Popen + +try: + from psutil import Process, NoSuchProcess +except ImportError: + """ Don't make psutil a strict requirement, but use if available. """ + Process = None + + +def kill_pid(pid, use_psutil=True): + if use_psutil and Process: + _psutil_kill_pid(pid) + else: + _stock_kill_pid(pid) + + +def _psutil_kill_pid(pid): + """ + http://stackoverflow.com/questions/1230669/subprocess-deleting-child-proces… + """ + try: + parent = Process(pid) + for child in parent.get_children(recursive=True): + child.kill() + parent.kill() + except NoSuchProcess: + return + + +def _stock_kill_pid(pid): + is_windows = system() == 'Windows' + + if is_windows: + __kill_windows(pid) + else: + __kill_posix(pid) + + +def __kill_windows(pid): + try: + Popen("taskkill /F /T /PID %i" % pid, shell=True) + except Exception: + pass + + +def __kill_posix(pid): + def __check_pid(): + try: + os.kill(pid, 0) + return True + except OSError: + return False + + if __check_pid(): + for sig in [15, 9]: + try: + os.killpg(pid, sig) + except OSError: + return + sleep(1) + if not __check_pid(): + return Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: 2 new changesets
by commits-noreply＠bitbucket.org 07 May '14

07 May '14

2 new commits in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/a300bc54ca1f/ Changeset: a300bc54ca1f User: jmchilton Date: 2014-05-05 19:34:38 Summary: Backport CLI-manager infrastructure from LWR. I had previously reworked this CLI stuff to work with LWR - this changeset brings those changes back to Galaxy. This version has the advantage of using the job script module - the upshot is that GALAXY_SLOTS, job_conf env directives, and job metrics should all be usable now with the CLI runner. This version also has a local shell variant (to eliminate the use of SSH and run qsub directly on the same machine), a variant of the Torque executor to target slurm's torque compatibility executables (doesn't support -x), and some other small bug fixes. Affected #: 14 files diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/cli.py --- a/lib/galaxy/jobs/runners/cli.py +++ b/lib/galaxy/jobs/runners/cli.py @@ -3,18 +3,18 @@ """ import os -import time -import glob import logging from galaxy import model from galaxy.jobs import JobDestination from galaxy.jobs.runners import AsynchronousJobState, AsynchronousJobRunner +from .util.cli import CliInterface, split_params log = logging.getLogger( __name__ ) __all__ = [ 'ShellJobRunner' ] + class ShellJobRunner( AsynchronousJobRunner ): """ Job runner backed by a finite pool of worker threads. FIFO scheduling @@ -25,53 +25,27 @@ """Start the job runner """ super( ShellJobRunner, self ).__init__( app, nworkers ) - self.cli_shells = None - self.cli_job_interfaces = None - self.__load_cli_plugins() - + self.cli_interface = CliInterface() self._init_monitor_thread() self._init_worker_threads() - def __load_cli_plugins(self): - def __load(module_path, d): - for file in glob.glob(os.path.join(os.path.join(os.getcwd(), 'lib', *module_path.split('.')), '*.py')): - if os.path.basename(file).startswith('_'): - continue - module_name = '%s.%s' % (module_path, os.path.basename(file).rsplit('.py', 1)[0]) - module = __import__(module_name) - for comp in module_name.split( "." )[1:]: - module = getattr(module, comp) - for name in module.__all__: - log.debug('Loaded cli plugin %s' % name) - d[name] = getattr(module, name) - - self.cli_shells = {} - self.cli_job_interfaces = {} - __load('galaxy.jobs.runners.cli_shell', self.cli_shells) - __load('galaxy.jobs.runners.cli_job', self.cli_job_interfaces) - def get_cli_plugins( self, shell_params, job_params ): - # load shell plugin - shell = self.cli_shells[shell_params['plugin']](**shell_params) - job_interface = self.cli_job_interfaces[job_params['plugin']](**job_params) - return shell, job_interface + return self.cli_interface.get_plugins( shell_params, job_params ) def url_to_destination( self, url ): params = {} - shell_params, job_params = url.split('/')[2:4] + shell_params, job_params = url.split( '/' )[ 2:4 ] # split 'foo=bar&baz=quux' into { 'foo' : 'bar', 'baz' : 'quux' } - shell_params = dict ( [ ( 'shell_' + k, v ) for k, v in [ kv.split('=', 1) for kv in shell_params.split('&') ] ] ) - job_params = dict ( [ ( 'job_' + k, v ) for k, v in [ kv.split('=', 1) for kv in job_params.split('&') ] ] ) + shell_params = dict( [ ( 'shell_' + k, v ) for k, v in [ kv.split( '=', 1 ) for kv in shell_params.split( '&' ) ] ] ) + job_params = dict( [ ( 'job_' + k, v ) for k, v in [ kv.split( '=', 1 ) for kv in job_params.split( '&' ) ] ] ) params.update( shell_params ) params.update( job_params ) - log.debug("Converted URL '%s' to destination runner=cli, params=%s" % (url, params)) + log.debug( "Converted URL '%s' to destination runner=cli, params=%s" % ( url, params ) ) # Create a dynamic JobDestination - return JobDestination(runner='cli', params=params) + return JobDestination( runner='cli', params=params ) def parse_destination_params( self, params ): - shell_params = dict((k.replace('shell_', '', 1), v) for k, v in params.items() if k.startswith('shell_')) - job_params = dict((k.replace('job_', '', 1), v) for k, v in params.items() if k.startswith('job_')) - return shell_params, job_params + return split_params( params ) def queue_job( self, job_wrapper ): """Create job script and submit it to the DRM""" @@ -93,8 +67,12 @@ # define job attributes ajs = AsynchronousJobState( files_dir=job_wrapper.working_directory, job_wrapper=job_wrapper ) - # fill in the DRM's job run template - script = job_interface.get_job_template(ajs.output_file, ajs.error_file, ajs.job_name, job_wrapper, command_line, ajs.exit_code_file) + job_file_kwargs = job_interface.job_script_kwargs(ajs.output_file, ajs.error_file, ajs.job_name) + script = self.get_job_file( + job_wrapper, + exit_code_path=ajs.exit_code_file, + **job_file_kwargs + ) try: fh = file(ajs.job_file, "w") diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/cli_job/__init__.py --- a/lib/galaxy/jobs/runners/cli_job/__init__.py +++ /dev/null @@ -1,21 +0,0 @@ -""" -Base class for cli job plugins -""" - -class BaseJobExec(object): - def __init__(self, **params): - raise NotImplementedError() - def get_job_template(self, ofile, efile, job_name, job_wrapper, command_line, ecfile): - raise NotImplementedError() - def submit(self, script_file): - raise NotImplementedError() - def delete(self, job_id): - raise NotImplementedError() - def get_status(self, job_ids=None): - raise NotImplementedError() - def get_single_status(self, job_id): - raise NotImplementedError() - def parse_status(self, status, job_ids): - raise NotImplementedError() - def parse_single_status(self, status, job_id): - raise NotImplementedError() diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/cli_job/torque.py --- a/lib/galaxy/jobs/runners/cli_job/torque.py +++ /dev/null @@ -1,135 +0,0 @@ -""" -Command-line interface to TORQUE PBS -""" - -import os -import logging - -from galaxy.model import Job -job_states = Job.states - -from galaxy.jobs.runners.cli_job import BaseJobExec - -log = logging.getLogger( __name__ ) - -__all__ = ('Torque',) - -try: - import xml.etree.cElementTree as et -except: - import xml.etree.ElementTree as et - -job_template = """#!/bin/sh -%s -GALAXY_LIB="%s" -if [ "$GALAXY_LIB" != "None" ]; then - if [ -n "$PYTHONPATH" ]; then - export PYTHONPATH="$GALAXY_LIB:$PYTHONPATH" - else - export PYTHONPATH="$GALAXY_LIB" - fi -fi -%s -cd %s -%s -echo $? > %s -""" - -argmap = { 'destination' : '-q', - 'Execution_Time' : '-a', - 'Account_Name' : '-A', - 'Checkpoint' : '-c', - 'Error_Path' : '-e', - 'Group_List' : '-g', - 'Hold_Types' : '-h', - 'Join_Paths' : '-j', - 'Keep_Files' : '-k', - 'Resource_List' : '-l', - 'Mail_Points' : '-m', - 'Mail_Users' : '-M', - 'Job_Name' : '-N', - 'Output_Path' : '-o', - 'Priority' : '-p', - 'Rerunable' : '-r', - 'Shell_Path_List' : '-S', - 'job_array_request' : '-t', - 'User_List' : '-u', - 'Variable_List' : '-v' } - -class Torque(BaseJobExec): - def __init__(self, **params): - self.params = {} - for k, v in params.items(): - self.params[k] = v - - def get_job_template(self, ofile, efile, job_name, job_wrapper, command_line, ecfile): - pbsargs = { '-o' : ofile, - '-e' : efile, - '-N' : job_name } - for k, v in self.params.items(): - if k == 'plugin': - continue - try: - if not k.startswith('-'): - k = argmap[k] - pbsargs[k] = v - except: - log.warning('Unrecognized long argument passed to Torque CLI plugin: %s' % k) - template_pbsargs = '' - for k, v in pbsargs.items(): - template_pbsargs += '#PBS %s %s\n' % (k, v) - return job_template % (template_pbsargs, - job_wrapper.galaxy_lib_dir, - job_wrapper.get_env_setup_clause(), - os.path.abspath(job_wrapper.working_directory), - command_line, - ecfile) - - def submit(self, script_file): - return 'qsub %s' % script_file - - def delete(self, job_id): - return 'qdel %s' % job_id - - def get_status(self, job_ids=None): - return 'qstat -x' - - def get_single_status(self, job_id): - return 'qstat -f %s' % job_id - - def parse_status(self, status, job_ids): - # in case there's noise in the output, find the big blob 'o xml - tree = None - rval = {} - for line in status.strip().splitlines(): - try: - tree = et.fromstring(line.strip()) - assert tree.tag == 'Data' - break - except Exception, e: - tree = None - if tree is None: - log.warning('No valid qstat XML return from `qstat -x`, got the following: %s' % status) - return None - else: - for job in tree.findall('Job'): - id = job.find('Job_Id').text - if id in job_ids: - state = job.find('job_state').text - # map PBS job states to Galaxy job states. - rval[id] = self.__get_job_state(state) - return rval - - def parse_single_status(self, status, job_id): - for line in status.splitlines(): - line = line.split(' = ') - if line[0] == 'job_state': - return line[1] - # no state found, job has exited - return job_states.OK - - def __get_job_state(self, state): - return { 'E' : job_states.RUNNING, - 'R' : job_states.RUNNING, - 'Q' : job_states.QUEUED, - 'C' : job_states.OK }.get(state, state) diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/cli_shell/__init__.py --- a/lib/galaxy/jobs/runners/cli_shell/__init__.py +++ /dev/null @@ -1,11 +0,0 @@ -""" -Base class for runners which execute commands via a shell -""" - -class BaseShellExec(object): - def __init__(self, *args, **kwargs): - raise NotImplementedError() - def copy(self, rcp_cmd, files, dest): - raise NotImplementedError() - def execute(self, cmd, persist=False, timeout=60): - raise NotImplementedError() diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/cli_shell/rsh.py --- a/lib/galaxy/jobs/runners/cli_shell/rsh.py +++ /dev/null @@ -1,64 +0,0 @@ -""" -Interface for remote shell commands (rsh, rcp) and derivatives that use the same syntax (ssh, scp) -""" - -import logging -import os -import subprocess -import tempfile -import time - -from galaxy.util.bunch import Bunch -from galaxy.jobs.runners.cli_shell import BaseShellExec - -log = logging.getLogger( __name__ ) - -__all__ = ('RemoteShell', 'SecureShell', 'GlobusSecureShell') - -class RemoteShell(BaseShellExec): - def __init__(self, rsh='rsh', rcp='rcp', hostname=None, username=None, **kwargs): - self.rsh = rsh - self.rcp = rcp - self.hostname = hostname - self.username = username - self.sessions = {} - def copy(self, rcp_cmd, files, dest): - pass - def execute(self, cmd, persist=False, timeout=60): - # TODO: implement persistence - if self.username is None: - fullcmd = '%s %s %s' % (self.rsh, self.hostname, cmd) - else: - fullcmd = '%s -l %s %s %s' % (self.rsh, self.username, self.hostname, cmd) - # Read stdout to a tempfile in case it's large (>65K) - outf = tempfile.TemporaryFile() - p = subprocess.Popen(fullcmd, shell=True, stdin=None, stdout=outf, stderr=subprocess.PIPE) - # poll until timeout - for i in range(timeout/3): - r = p.poll() - if r is not None: - break - time.sleep(3) - else: - pid = int(p.pid) - for sig in (15, 9): - try: - os.kill(pid, sig) - time.sleep(3) - except: - log.warning('Killing pid %s (cmd: "%s") with signal %s failed' % (p.pid, fullcmd, sig)) - return Bunch(stdout='', stderr='Execution timed out', returncode=-1) - outf.seek(0) - return Bunch(stdout=outf.read(), stderr=p.stderr.read(), returncode=p.returncode) - - -class SecureShell(RemoteShell): - SSH_NEW_KEY_STRING = 'Are you sure you want to continue connecting' - def __init__(self, rsh='ssh', rcp='scp', **kwargs): - rsh += ' -oStrictHostKeyChecking=yes -oConnectTimeout=60' - rcp += ' -oStrictHostKeyChecking=yes -oConnectTimeout=60' - super(SecureShell, self).__init__(rsh=rsh, rcp=rcp, **kwargs) - -class GlobusSecureShell(SecureShell): - def __init__(self, rsh='gsissh', rcp='gsiscp', **kwargs): - super(SecureShell, self).__init__(rsh=rsh, rcp=rcp, **kwargs) diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/util/__init__.py --- a/lib/galaxy/jobs/runners/util/__init__.py +++ b/lib/galaxy/jobs/runners/util/__init__.py @@ -3,5 +3,8 @@ processes and interfacing with job managers. This module should contain functionality shared between Galaxy and the LWR. """ +from galaxy.util.bunch import Bunch -from galaxy.util.bunch import Bunch +from .kill import kill_pid + +__all__ = [kill_pid, Bunch] diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/util/cli/__init__.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/__init__.py @@ -0,0 +1,56 @@ +""" +""" +from glob import glob +from os.path import basename, join +from os import getcwd + +DEFAULT_SHELL_PLUGIN = 'LocalShell' + + +class CliInterface(object): + """ + High-level interface for loading shell and job plugins and matching + them to specified parameters. + """ + + def __init__(self, code_dir='lib'): + """ + """ + def __load(module_path, d): + module_pattern = join(join(getcwd(), code_dir, *module_path.split('.')), '*.py') + for file in glob(module_pattern): + if basename(file).startswith('_'): + continue + module_name = '%s.%s' % (module_path, basename(file).rsplit('.py', 1)[0]) + module = __import__(module_name) + for comp in module_name.split(".")[1:]: + module = getattr(module, comp) + for name in module.__all__: + try: + d[name] = getattr(module, name) + except TypeError: + raise TypeError("Invalid type for name %s" % name) + + self.cli_shells = {} + self.cli_job_interfaces = {} + + module_prefix = self.__module__ + __load('%s.shell' % module_prefix, self.cli_shells) + __load('%s.job' % module_prefix, self.cli_job_interfaces) + + def get_plugins(self, shell_params, job_params): + """ + Return shell and job interface defined by and configured via + specified params. + """ + shell_plugin = shell_params.get('plugin', DEFAULT_SHELL_PLUGIN) + job_plugin = job_params['plugin'] + shell = self.cli_shells[shell_plugin](**shell_params) + job_interface = self.cli_job_interfaces[job_plugin](**job_params) + return shell, job_interface + + +def split_params(params): + shell_params = dict((k.replace('shell_', '', 1), v) for k, v in params.items() if k.startswith('shell_')) + job_params = dict((k.replace('job_', '', 1), v) for k, v in params.items() if k.startswith('job_')) + return shell_params, job_params diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/util/cli/job/__init__.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/job/__init__.py @@ -0,0 +1,58 @@ +""" +Abstract base class for cli job plugins. +""" +from abc import ABCMeta, abstractmethod + + +class BaseJobExec(object): + __metaclass__ = ABCMeta + + @abstractmethod + def __init__(self, **params): + """ + Constructor for CLI job executor. + """ + + def job_script_kwargs(self, ofile, efile, job_name): + """ Return extra keyword argument for consumption by job script + module. + """ + return {} + + @abstractmethod + def submit(self, script_file): + """ + Given specified script_file path, yield command to submit it + to external job manager. + """ + + @abstractmethod + def delete(self, job_id): + """ + Given job id, return command to stop execution or dequeue specified + job. + """ + + @abstractmethod + def get_status(self, job_ids=None): + """ + Return command to get statuses of specified job ids. + """ + + @abstractmethod + def get_single_status(self, job_id): + """ + Return command to get the status of a single, specified job. + """ + + @abstractmethod + def parse_status(self, status, job_ids): + """ + Parse the statuses of output from get_status command. + """ + + @abstractmethod + def parse_single_status(self, status, job_id): + """ + Parse the status of output from get_single_status command. + """ diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/util/cli/job/slurm_torque.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/job/slurm_torque.py @@ -0,0 +1,29 @@ +import re +from .torque import Torque + +__all__ = ('SlurmTorque',) + + +class SlurmTorque(Torque): + """ A CLI job executor for Slurm's Torque compatibility mode. This differs + from real torque CLI in that -x command line is not available so job status + needs to be parsed from qstat table instead of XML. + """ + + def get_status(self, job_ids=None): + return 'qstat' + + def parse_status(self, status, job_ids): + rval = {} + for line in status.strip().splitlines(): + if line.startswith("Job ID"): + continue + line_parts = re.compile("\s+").split(line) + if len(line_parts) < 5: + continue + id = line_parts[0] + state = line_parts[4] + if id in job_ids: + # map PBS job states to Galaxy job states. + rval[id] = self._get_job_state(state) + return rval diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/util/cli/job/torque.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/job/torque.py @@ -0,0 +1,120 @@ +try: + import xml.etree.cElementTree as et +except: + import xml.etree.ElementTree as et + +try: + from galaxy.model import Job + job_states = Job.states +except ImportError: + # Not in Galaxy, map Galaxy job states to LWR ones. + from galaxy.util import enum + job_states = enum(RUNNING='running', OK='complete', QUEUED='queued') + +from ..job import BaseJobExec + +__all__ = ('Torque',) + +from logging import getLogger +log = getLogger(__name__) + +argmap = {'destination': '-q', + 'Execution_Time': '-a', + 'Account_Name': '-A', + 'Checkpoint': '-c', + 'Error_Path': '-e', + 'Group_List': '-g', + 'Hold_Types': '-h', + 'Join_Paths': '-j', + 'Keep_Files': '-k', + 'Resource_List': '-l', + 'Mail_Points': '-m', + 'Mail_Users': '-M', + 'Job_Name': '-N', + 'Output_Path': '-o', + 'Priority': '-p', + 'Rerunable': '-r', + 'Shell_Path_List': '-S', + 'job_array_request': '-t', + 'User_List': '-u', + 'Variable_List': '-v'} + + +class Torque(BaseJobExec): + + def __init__(self, **params): + self.params = {} + for k, v in params.items(): + self.params[k] = v + + def job_script_kwargs(self, ofile, efile, job_name): + pbsargs = {'-o': ofile, + '-e': efile, + '-N': job_name} + for k, v in self.params.items(): + if k == 'plugin': + continue + try: + if not k.startswith('-'): + k = argmap[k] + pbsargs[k] = v + except: + log.warning('Unrecognized long argument passed to Torque CLI plugin: %s' % k) + template_pbsargs = '' + for k, v in pbsargs.items(): + template_pbsargs += '#PBS %s %s\n' % (k, v) + return dict(headers=template_pbsargs) + + def submit(self, script_file): + return 'qsub %s' % script_file + + def delete(self, job_id): + return 'qdel %s' % job_id + + def get_status(self, job_ids=None): + return 'qstat -x' + + def get_single_status(self, job_id): + return 'qstat -f %s' % job_id + + def parse_status(self, status, job_ids): + # in case there's noise in the output, find the big blob 'o xml + tree = None + rval = {} + for line in status.strip().splitlines(): + try: + tree = et.fromstring(line.strip()) + assert tree.tag == 'Data' + break + except Exception: + tree = None + if tree is None: + log.warning('No valid qstat XML return from `qstat -x`, got the following: %s' % status) + return None + else: + for job in tree.findall('Job'): + id = job.find('Job_Id').text + if id in job_ids: + state = job.find('job_state').text + # map PBS job states to Galaxy job states. + rval[id] = self._get_job_state(state) + return rval + + def parse_single_status(self, status, job_id): + for line in status.splitlines(): + line = line.split(' = ') + if line[0] == 'job_state': + return self._get_job_state(line[1].strip()) + # no state found, job has exited + return job_states.OK + + def _get_job_state(self, state): + try: + return { + 'E': job_states.RUNNING, + 'R': job_states.RUNNING, + 'Q': job_states.QUEUED, + 'C': job_states.OK + }.get(state) + except KeyError: + raise KeyError("Failed to map torque status code [%s] to job state." % state) diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/util/cli/shell/__init__.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/shell/__init__.py @@ -0,0 +1,19 @@ +""" +Abstract base class for runners which execute commands via a shell. +""" +from abc import ABCMeta, abstractmethod + + +class BaseShellExec(object): + __metaclass__ = ABCMeta + + @abstractmethod + def __init__(self, *args, **kwargs): + """ + Constructor for shell executor instance. + """ + + def execute(self, cmd, persist=False, timeout=60): + """ + Execute the specified command via defined shell. + """ diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/util/cli/shell/local.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/shell/local.py @@ -0,0 +1,61 @@ +from tempfile import TemporaryFile +from time import sleep +from subprocess import Popen, PIPE + +from ..shell import BaseShellExec +from ....util import Bunch, kill_pid + +from logging import getLogger +log = getLogger(__name__) + +TIMEOUT_ERROR_MESSAGE = u'Execution timed out' +TIMEOUT_RETURN_CODE = -1 +DEFAULT_TIMEOUT = 60 +DEFAULT_TIMEOUT_CHECK_INTERVAL = 3 + + +class LocalShell(BaseShellExec): + """ + + >>> shell = LocalShell() + >>> def exec_python(script, **kwds): return shell.execute('python -c "%s"' % script, **kwds) + >>> exec_result = exec_python("from __future__ import print_function; print('Hello World')") + >>> exec_result.stderr == u'' + True + >>> exec_result.stdout.strip() == u'Hello World' + True + >>> exec_result = exec_python("import time; time.sleep(90)", timeout=3, timeout_check_interval=1) + >>> exec_result.stdout == u'' + True + >>> exec_result.stderr == 'Execution timed out' + True + >>> exec_result.returncode == TIMEOUT_RETURN_CODE + True + """ + + def __init__(self, **kwds): + pass + + def execute(self, cmd, persist=False, timeout=DEFAULT_TIMEOUT, timeout_check_interval=DEFAULT_TIMEOUT_CHECK_INTERVAL, **kwds): + outf = TemporaryFile() + p = Popen(cmd, shell=True, stdin=None, stdout=outf, stderr=PIPE) + # poll until timeout + + for i in range(int(timeout / timeout_check_interval)): + r = p.poll() + if r is not None: + break + sleep(timeout_check_interval) + else: + kill_pid(p.pid) + return Bunch(stdout=u'', stderr=TIMEOUT_ERROR_MESSAGE, returncode=TIMEOUT_RETURN_CODE) + outf.seek(0) + return Bunch(stdout=_read_str(outf), stderr=_read_str(p.stderr), returncode=p.returncode) + + +def _read_str(stream): + contents = stream.read() + return contents.decode('UTF-8') if isinstance(contents, bytes) else contents + + +__all__ = ('LocalShell',) diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/util/cli/shell/rsh.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/shell/rsh.py @@ -0,0 +1,40 @@ +from .local import LocalShell + +from logging import getLogger +log = getLogger(__name__) + +__all__ = ('RemoteShell', 'SecureShell', 'GlobusSecureShell') + + +class RemoteShell(LocalShell): + + def __init__(self, rsh='rsh', rcp='rcp', hostname='localhost', username=None, **kwargs): + super(RemoteShell, self).__init__(**kwargs) + self.rsh = rsh + self.rcp = rcp + self.hostname = hostname + self.username = username + self.sessions = {} + + def execute(self, cmd, persist=False, timeout=60): + # TODO: implement persistence + if self.username is None: + fullcmd = '%s %s %s' % (self.rsh, self.hostname, cmd) + else: + fullcmd = '%s -l %s %s %s' % (self.rsh, self.username, self.hostname, cmd) + return super(RemoteShell, self).execute(fullcmd, persist, timeout) + + +class SecureShell(RemoteShell): + SSH_NEW_KEY_STRING = 'Are you sure you want to continue connecting' + + def __init__(self, rsh='ssh', rcp='scp', **kwargs): + rsh += ' -oStrictHostKeyChecking=yes -oConnectTimeout=60' + rcp += ' -oStrictHostKeyChecking=yes -oConnectTimeout=60' + super(SecureShell, self).__init__(rsh=rsh, rcp=rcp, **kwargs) + + +class GlobusSecureShell(SecureShell): + + def __init__(self, rsh='gsissh', rcp='gsiscp', **kwargs): + super(GlobusSecureShell, self).__init__(rsh=rsh, rcp=rcp, **kwargs) diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r a300bc54ca1fd0479ae94541f56ec7899f2e2c90 lib/galaxy/jobs/runners/util/kill.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/kill.py @@ -0,0 +1,65 @@ +import os +from platform import system +from time import sleep +from subprocess import Popen + +try: + from psutil import Process, NoSuchProcess +except ImportError: + """ Don't make psutil a strict requirement, but use if available. """ + Process = None + + +def kill_pid(pid, use_psutil=True): + if use_psutil and Process: + _psutil_kill_pid(pid) + else: + _stock_kill_pid(pid) + + +def _psutil_kill_pid(pid): + """ + http://stackoverflow.com/questions/1230669/subprocess-deleting-child-proces… + """ + try: + parent = Process(pid) + for child in parent.get_children(recursive=True): + child.kill() + parent.kill() + except NoSuchProcess: + return + + +def _stock_kill_pid(pid): + is_windows = system() == 'Windows' + + if is_windows: + __kill_windows(pid) + else: + __kill_posix(pid) + + +def __kill_windows(pid): + try: + Popen("taskkill /F /T /PID %i" % pid, shell=True) + except Exception: + pass + + +def __kill_posix(pid): + def __check_pid(): + try: + os.kill(pid, 0) + return True + except OSError: + return False + + if __check_pid(): + for sig in [15, 9]: + try: + os.killpg(pid, sig) + except OSError: + return + sleep(1) + if not __check_pid(): + return https://bitbucket.org/galaxy/galaxy-central/commits/9eb2b42e21fe/ Changeset: 9eb2b42e21fe User: jmchilton Date: 2014-05-07 18:57:21 Summary: Merged in jmchilton/galaxy-central-fork-1 (pull request #385) CLI Job Runner Enhancements Affected #: 14 files diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/cli.py --- a/lib/galaxy/jobs/runners/cli.py +++ b/lib/galaxy/jobs/runners/cli.py @@ -3,18 +3,18 @@ """ import os -import time -import glob import logging from galaxy import model from galaxy.jobs import JobDestination from galaxy.jobs.runners import AsynchronousJobState, AsynchronousJobRunner +from .util.cli import CliInterface, split_params log = logging.getLogger( __name__ ) __all__ = [ 'ShellJobRunner' ] + class ShellJobRunner( AsynchronousJobRunner ): """ Job runner backed by a finite pool of worker threads. FIFO scheduling @@ -25,53 +25,27 @@ """Start the job runner """ super( ShellJobRunner, self ).__init__( app, nworkers ) - self.cli_shells = None - self.cli_job_interfaces = None - self.__load_cli_plugins() - + self.cli_interface = CliInterface() self._init_monitor_thread() self._init_worker_threads() - def __load_cli_plugins(self): - def __load(module_path, d): - for file in glob.glob(os.path.join(os.path.join(os.getcwd(), 'lib', *module_path.split('.')), '*.py')): - if os.path.basename(file).startswith('_'): - continue - module_name = '%s.%s' % (module_path, os.path.basename(file).rsplit('.py', 1)[0]) - module = __import__(module_name) - for comp in module_name.split( "." )[1:]: - module = getattr(module, comp) - for name in module.__all__: - log.debug('Loaded cli plugin %s' % name) - d[name] = getattr(module, name) - - self.cli_shells = {} - self.cli_job_interfaces = {} - __load('galaxy.jobs.runners.cli_shell', self.cli_shells) - __load('galaxy.jobs.runners.cli_job', self.cli_job_interfaces) - def get_cli_plugins( self, shell_params, job_params ): - # load shell plugin - shell = self.cli_shells[shell_params['plugin']](**shell_params) - job_interface = self.cli_job_interfaces[job_params['plugin']](**job_params) - return shell, job_interface + return self.cli_interface.get_plugins( shell_params, job_params ) def url_to_destination( self, url ): params = {} - shell_params, job_params = url.split('/')[2:4] + shell_params, job_params = url.split( '/' )[ 2:4 ] # split 'foo=bar&baz=quux' into { 'foo' : 'bar', 'baz' : 'quux' } - shell_params = dict ( [ ( 'shell_' + k, v ) for k, v in [ kv.split('=', 1) for kv in shell_params.split('&') ] ] ) - job_params = dict ( [ ( 'job_' + k, v ) for k, v in [ kv.split('=', 1) for kv in job_params.split('&') ] ] ) + shell_params = dict( [ ( 'shell_' + k, v ) for k, v in [ kv.split( '=', 1 ) for kv in shell_params.split( '&' ) ] ] ) + job_params = dict( [ ( 'job_' + k, v ) for k, v in [ kv.split( '=', 1 ) for kv in job_params.split( '&' ) ] ] ) params.update( shell_params ) params.update( job_params ) - log.debug("Converted URL '%s' to destination runner=cli, params=%s" % (url, params)) + log.debug( "Converted URL '%s' to destination runner=cli, params=%s" % ( url, params ) ) # Create a dynamic JobDestination - return JobDestination(runner='cli', params=params) + return JobDestination( runner='cli', params=params ) def parse_destination_params( self, params ): - shell_params = dict((k.replace('shell_', '', 1), v) for k, v in params.items() if k.startswith('shell_')) - job_params = dict((k.replace('job_', '', 1), v) for k, v in params.items() if k.startswith('job_')) - return shell_params, job_params + return split_params( params ) def queue_job( self, job_wrapper ): """Create job script and submit it to the DRM""" @@ -93,8 +67,12 @@ # define job attributes ajs = AsynchronousJobState( files_dir=job_wrapper.working_directory, job_wrapper=job_wrapper ) - # fill in the DRM's job run template - script = job_interface.get_job_template(ajs.output_file, ajs.error_file, ajs.job_name, job_wrapper, command_line, ajs.exit_code_file) + job_file_kwargs = job_interface.job_script_kwargs(ajs.output_file, ajs.error_file, ajs.job_name) + script = self.get_job_file( + job_wrapper, + exit_code_path=ajs.exit_code_file, + **job_file_kwargs + ) try: fh = file(ajs.job_file, "w") diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/cli_job/__init__.py --- a/lib/galaxy/jobs/runners/cli_job/__init__.py +++ /dev/null @@ -1,21 +0,0 @@ -""" -Base class for cli job plugins -""" - -class BaseJobExec(object): - def __init__(self, **params): - raise NotImplementedError() - def get_job_template(self, ofile, efile, job_name, job_wrapper, command_line, ecfile): - raise NotImplementedError() - def submit(self, script_file): - raise NotImplementedError() - def delete(self, job_id): - raise NotImplementedError() - def get_status(self, job_ids=None): - raise NotImplementedError() - def get_single_status(self, job_id): - raise NotImplementedError() - def parse_status(self, status, job_ids): - raise NotImplementedError() - def parse_single_status(self, status, job_id): - raise NotImplementedError() diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/cli_job/torque.py --- a/lib/galaxy/jobs/runners/cli_job/torque.py +++ /dev/null @@ -1,135 +0,0 @@ -""" -Command-line interface to TORQUE PBS -""" - -import os -import logging - -from galaxy.model import Job -job_states = Job.states - -from galaxy.jobs.runners.cli_job import BaseJobExec - -log = logging.getLogger( __name__ ) - -__all__ = ('Torque',) - -try: - import xml.etree.cElementTree as et -except: - import xml.etree.ElementTree as et - -job_template = """#!/bin/sh -%s -GALAXY_LIB="%s" -if [ "$GALAXY_LIB" != "None" ]; then - if [ -n "$PYTHONPATH" ]; then - export PYTHONPATH="$GALAXY_LIB:$PYTHONPATH" - else - export PYTHONPATH="$GALAXY_LIB" - fi -fi -%s -cd %s -%s -echo $? > %s -""" - -argmap = { 'destination' : '-q', - 'Execution_Time' : '-a', - 'Account_Name' : '-A', - 'Checkpoint' : '-c', - 'Error_Path' : '-e', - 'Group_List' : '-g', - 'Hold_Types' : '-h', - 'Join_Paths' : '-j', - 'Keep_Files' : '-k', - 'Resource_List' : '-l', - 'Mail_Points' : '-m', - 'Mail_Users' : '-M', - 'Job_Name' : '-N', - 'Output_Path' : '-o', - 'Priority' : '-p', - 'Rerunable' : '-r', - 'Shell_Path_List' : '-S', - 'job_array_request' : '-t', - 'User_List' : '-u', - 'Variable_List' : '-v' } - -class Torque(BaseJobExec): - def __init__(self, **params): - self.params = {} - for k, v in params.items(): - self.params[k] = v - - def get_job_template(self, ofile, efile, job_name, job_wrapper, command_line, ecfile): - pbsargs = { '-o' : ofile, - '-e' : efile, - '-N' : job_name } - for k, v in self.params.items(): - if k == 'plugin': - continue - try: - if not k.startswith('-'): - k = argmap[k] - pbsargs[k] = v - except: - log.warning('Unrecognized long argument passed to Torque CLI plugin: %s' % k) - template_pbsargs = '' - for k, v in pbsargs.items(): - template_pbsargs += '#PBS %s %s\n' % (k, v) - return job_template % (template_pbsargs, - job_wrapper.galaxy_lib_dir, - job_wrapper.get_env_setup_clause(), - os.path.abspath(job_wrapper.working_directory), - command_line, - ecfile) - - def submit(self, script_file): - return 'qsub %s' % script_file - - def delete(self, job_id): - return 'qdel %s' % job_id - - def get_status(self, job_ids=None): - return 'qstat -x' - - def get_single_status(self, job_id): - return 'qstat -f %s' % job_id - - def parse_status(self, status, job_ids): - # in case there's noise in the output, find the big blob 'o xml - tree = None - rval = {} - for line in status.strip().splitlines(): - try: - tree = et.fromstring(line.strip()) - assert tree.tag == 'Data' - break - except Exception, e: - tree = None - if tree is None: - log.warning('No valid qstat XML return from `qstat -x`, got the following: %s' % status) - return None - else: - for job in tree.findall('Job'): - id = job.find('Job_Id').text - if id in job_ids: - state = job.find('job_state').text - # map PBS job states to Galaxy job states. - rval[id] = self.__get_job_state(state) - return rval - - def parse_single_status(self, status, job_id): - for line in status.splitlines(): - line = line.split(' = ') - if line[0] == 'job_state': - return line[1] - # no state found, job has exited - return job_states.OK - - def __get_job_state(self, state): - return { 'E' : job_states.RUNNING, - 'R' : job_states.RUNNING, - 'Q' : job_states.QUEUED, - 'C' : job_states.OK }.get(state, state) diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/cli_shell/__init__.py --- a/lib/galaxy/jobs/runners/cli_shell/__init__.py +++ /dev/null @@ -1,11 +0,0 @@ -""" -Base class for runners which execute commands via a shell -""" - -class BaseShellExec(object): - def __init__(self, *args, **kwargs): - raise NotImplementedError() - def copy(self, rcp_cmd, files, dest): - raise NotImplementedError() - def execute(self, cmd, persist=False, timeout=60): - raise NotImplementedError() diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/cli_shell/rsh.py --- a/lib/galaxy/jobs/runners/cli_shell/rsh.py +++ /dev/null @@ -1,64 +0,0 @@ -""" -Interface for remote shell commands (rsh, rcp) and derivatives that use the same syntax (ssh, scp) -""" - -import logging -import os -import subprocess -import tempfile -import time - -from galaxy.util.bunch import Bunch -from galaxy.jobs.runners.cli_shell import BaseShellExec - -log = logging.getLogger( __name__ ) - -__all__ = ('RemoteShell', 'SecureShell', 'GlobusSecureShell') - -class RemoteShell(BaseShellExec): - def __init__(self, rsh='rsh', rcp='rcp', hostname=None, username=None, **kwargs): - self.rsh = rsh - self.rcp = rcp - self.hostname = hostname - self.username = username - self.sessions = {} - def copy(self, rcp_cmd, files, dest): - pass - def execute(self, cmd, persist=False, timeout=60): - # TODO: implement persistence - if self.username is None: - fullcmd = '%s %s %s' % (self.rsh, self.hostname, cmd) - else: - fullcmd = '%s -l %s %s %s' % (self.rsh, self.username, self.hostname, cmd) - # Read stdout to a tempfile in case it's large (>65K) - outf = tempfile.TemporaryFile() - p = subprocess.Popen(fullcmd, shell=True, stdin=None, stdout=outf, stderr=subprocess.PIPE) - # poll until timeout - for i in range(timeout/3): - r = p.poll() - if r is not None: - break - time.sleep(3) - else: - pid = int(p.pid) - for sig in (15, 9): - try: - os.kill(pid, sig) - time.sleep(3) - except: - log.warning('Killing pid %s (cmd: "%s") with signal %s failed' % (p.pid, fullcmd, sig)) - return Bunch(stdout='', stderr='Execution timed out', returncode=-1) - outf.seek(0) - return Bunch(stdout=outf.read(), stderr=p.stderr.read(), returncode=p.returncode) - - -class SecureShell(RemoteShell): - SSH_NEW_KEY_STRING = 'Are you sure you want to continue connecting' - def __init__(self, rsh='ssh', rcp='scp', **kwargs): - rsh += ' -oStrictHostKeyChecking=yes -oConnectTimeout=60' - rcp += ' -oStrictHostKeyChecking=yes -oConnectTimeout=60' - super(SecureShell, self).__init__(rsh=rsh, rcp=rcp, **kwargs) - -class GlobusSecureShell(SecureShell): - def __init__(self, rsh='gsissh', rcp='gsiscp', **kwargs): - super(SecureShell, self).__init__(rsh=rsh, rcp=rcp, **kwargs) diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/__init__.py --- a/lib/galaxy/jobs/runners/util/__init__.py +++ b/lib/galaxy/jobs/runners/util/__init__.py @@ -3,5 +3,8 @@ processes and interfacing with job managers. This module should contain functionality shared between Galaxy and the LWR. """ +from galaxy.util.bunch import Bunch -from galaxy.util.bunch import Bunch +from .kill import kill_pid + +__all__ = [kill_pid, Bunch] diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/__init__.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/__init__.py @@ -0,0 +1,56 @@ +""" +""" +from glob import glob +from os.path import basename, join +from os import getcwd + +DEFAULT_SHELL_PLUGIN = 'LocalShell' + + +class CliInterface(object): + """ + High-level interface for loading shell and job plugins and matching + them to specified parameters. + """ + + def __init__(self, code_dir='lib'): + """ + """ + def __load(module_path, d): + module_pattern = join(join(getcwd(), code_dir, *module_path.split('.')), '*.py') + for file in glob(module_pattern): + if basename(file).startswith('_'): + continue + module_name = '%s.%s' % (module_path, basename(file).rsplit('.py', 1)[0]) + module = __import__(module_name) + for comp in module_name.split(".")[1:]: + module = getattr(module, comp) + for name in module.__all__: + try: + d[name] = getattr(module, name) + except TypeError: + raise TypeError("Invalid type for name %s" % name) + + self.cli_shells = {} + self.cli_job_interfaces = {} + + module_prefix = self.__module__ + __load('%s.shell' % module_prefix, self.cli_shells) + __load('%s.job' % module_prefix, self.cli_job_interfaces) + + def get_plugins(self, shell_params, job_params): + """ + Return shell and job interface defined by and configured via + specified params. + """ + shell_plugin = shell_params.get('plugin', DEFAULT_SHELL_PLUGIN) + job_plugin = job_params['plugin'] + shell = self.cli_shells[shell_plugin](**shell_params) + job_interface = self.cli_job_interfaces[job_plugin](**job_params) + return shell, job_interface + + +def split_params(params): + shell_params = dict((k.replace('shell_', '', 1), v) for k, v in params.items() if k.startswith('shell_')) + job_params = dict((k.replace('job_', '', 1), v) for k, v in params.items() if k.startswith('job_')) + return shell_params, job_params diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/job/__init__.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/job/__init__.py @@ -0,0 +1,58 @@ +""" +Abstract base class for cli job plugins. +""" +from abc import ABCMeta, abstractmethod + + +class BaseJobExec(object): + __metaclass__ = ABCMeta + + @abstractmethod + def __init__(self, **params): + """ + Constructor for CLI job executor. + """ + + def job_script_kwargs(self, ofile, efile, job_name): + """ Return extra keyword argument for consumption by job script + module. + """ + return {} + + @abstractmethod + def submit(self, script_file): + """ + Given specified script_file path, yield command to submit it + to external job manager. + """ + + @abstractmethod + def delete(self, job_id): + """ + Given job id, return command to stop execution or dequeue specified + job. + """ + + @abstractmethod + def get_status(self, job_ids=None): + """ + Return command to get statuses of specified job ids. + """ + + @abstractmethod + def get_single_status(self, job_id): + """ + Return command to get the status of a single, specified job. + """ + + @abstractmethod + def parse_status(self, status, job_ids): + """ + Parse the statuses of output from get_status command. + """ + + @abstractmethod + def parse_single_status(self, status, job_id): + """ + Parse the status of output from get_single_status command. + """ diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/job/slurm_torque.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/job/slurm_torque.py @@ -0,0 +1,29 @@ +import re +from .torque import Torque + +__all__ = ('SlurmTorque',) + + +class SlurmTorque(Torque): + """ A CLI job executor for Slurm's Torque compatibility mode. This differs + from real torque CLI in that -x command line is not available so job status + needs to be parsed from qstat table instead of XML. + """ + + def get_status(self, job_ids=None): + return 'qstat' + + def parse_status(self, status, job_ids): + rval = {} + for line in status.strip().splitlines(): + if line.startswith("Job ID"): + continue + line_parts = re.compile("\s+").split(line) + if len(line_parts) < 5: + continue + id = line_parts[0] + state = line_parts[4] + if id in job_ids: + # map PBS job states to Galaxy job states. + rval[id] = self._get_job_state(state) + return rval diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/job/torque.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/job/torque.py @@ -0,0 +1,120 @@ +try: + import xml.etree.cElementTree as et +except: + import xml.etree.ElementTree as et + +try: + from galaxy.model import Job + job_states = Job.states +except ImportError: + # Not in Galaxy, map Galaxy job states to LWR ones. + from galaxy.util import enum + job_states = enum(RUNNING='running', OK='complete', QUEUED='queued') + +from ..job import BaseJobExec + +__all__ = ('Torque',) + +from logging import getLogger +log = getLogger(__name__) + +argmap = {'destination': '-q', + 'Execution_Time': '-a', + 'Account_Name': '-A', + 'Checkpoint': '-c', + 'Error_Path': '-e', + 'Group_List': '-g', + 'Hold_Types': '-h', + 'Join_Paths': '-j', + 'Keep_Files': '-k', + 'Resource_List': '-l', + 'Mail_Points': '-m', + 'Mail_Users': '-M', + 'Job_Name': '-N', + 'Output_Path': '-o', + 'Priority': '-p', + 'Rerunable': '-r', + 'Shell_Path_List': '-S', + 'job_array_request': '-t', + 'User_List': '-u', + 'Variable_List': '-v'} + + +class Torque(BaseJobExec): + + def __init__(self, **params): + self.params = {} + for k, v in params.items(): + self.params[k] = v + + def job_script_kwargs(self, ofile, efile, job_name): + pbsargs = {'-o': ofile, + '-e': efile, + '-N': job_name} + for k, v in self.params.items(): + if k == 'plugin': + continue + try: + if not k.startswith('-'): + k = argmap[k] + pbsargs[k] = v + except: + log.warning('Unrecognized long argument passed to Torque CLI plugin: %s' % k) + template_pbsargs = '' + for k, v in pbsargs.items(): + template_pbsargs += '#PBS %s %s\n' % (k, v) + return dict(headers=template_pbsargs) + + def submit(self, script_file): + return 'qsub %s' % script_file + + def delete(self, job_id): + return 'qdel %s' % job_id + + def get_status(self, job_ids=None): + return 'qstat -x' + + def get_single_status(self, job_id): + return 'qstat -f %s' % job_id + + def parse_status(self, status, job_ids): + # in case there's noise in the output, find the big blob 'o xml + tree = None + rval = {} + for line in status.strip().splitlines(): + try: + tree = et.fromstring(line.strip()) + assert tree.tag == 'Data' + break + except Exception: + tree = None + if tree is None: + log.warning('No valid qstat XML return from `qstat -x`, got the following: %s' % status) + return None + else: + for job in tree.findall('Job'): + id = job.find('Job_Id').text + if id in job_ids: + state = job.find('job_state').text + # map PBS job states to Galaxy job states. + rval[id] = self._get_job_state(state) + return rval + + def parse_single_status(self, status, job_id): + for line in status.splitlines(): + line = line.split(' = ') + if line[0] == 'job_state': + return self._get_job_state(line[1].strip()) + # no state found, job has exited + return job_states.OK + + def _get_job_state(self, state): + try: + return { + 'E': job_states.RUNNING, + 'R': job_states.RUNNING, + 'Q': job_states.QUEUED, + 'C': job_states.OK + }.get(state) + except KeyError: + raise KeyError("Failed to map torque status code [%s] to job state." % state) diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/shell/__init__.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/shell/__init__.py @@ -0,0 +1,19 @@ +""" +Abstract base class for runners which execute commands via a shell. +""" +from abc import ABCMeta, abstractmethod + + +class BaseShellExec(object): + __metaclass__ = ABCMeta + + @abstractmethod + def __init__(self, *args, **kwargs): + """ + Constructor for shell executor instance. + """ + + def execute(self, cmd, persist=False, timeout=60): + """ + Execute the specified command via defined shell. + """ diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/shell/local.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/shell/local.py @@ -0,0 +1,61 @@ +from tempfile import TemporaryFile +from time import sleep +from subprocess import Popen, PIPE + +from ..shell import BaseShellExec +from ....util import Bunch, kill_pid + +from logging import getLogger +log = getLogger(__name__) + +TIMEOUT_ERROR_MESSAGE = u'Execution timed out' +TIMEOUT_RETURN_CODE = -1 +DEFAULT_TIMEOUT = 60 +DEFAULT_TIMEOUT_CHECK_INTERVAL = 3 + + +class LocalShell(BaseShellExec): + """ + + >>> shell = LocalShell() + >>> def exec_python(script, **kwds): return shell.execute('python -c "%s"' % script, **kwds) + >>> exec_result = exec_python("from __future__ import print_function; print('Hello World')") + >>> exec_result.stderr == u'' + True + >>> exec_result.stdout.strip() == u'Hello World' + True + >>> exec_result = exec_python("import time; time.sleep(90)", timeout=3, timeout_check_interval=1) + >>> exec_result.stdout == u'' + True + >>> exec_result.stderr == 'Execution timed out' + True + >>> exec_result.returncode == TIMEOUT_RETURN_CODE + True + """ + + def __init__(self, **kwds): + pass + + def execute(self, cmd, persist=False, timeout=DEFAULT_TIMEOUT, timeout_check_interval=DEFAULT_TIMEOUT_CHECK_INTERVAL, **kwds): + outf = TemporaryFile() + p = Popen(cmd, shell=True, stdin=None, stdout=outf, stderr=PIPE) + # poll until timeout + + for i in range(int(timeout / timeout_check_interval)): + r = p.poll() + if r is not None: + break + sleep(timeout_check_interval) + else: + kill_pid(p.pid) + return Bunch(stdout=u'', stderr=TIMEOUT_ERROR_MESSAGE, returncode=TIMEOUT_RETURN_CODE) + outf.seek(0) + return Bunch(stdout=_read_str(outf), stderr=_read_str(p.stderr), returncode=p.returncode) + + +def _read_str(stream): + contents = stream.read() + return contents.decode('UTF-8') if isinstance(contents, bytes) else contents + + +__all__ = ('LocalShell',) diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/cli/shell/rsh.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/cli/shell/rsh.py @@ -0,0 +1,40 @@ +from .local import LocalShell + +from logging import getLogger +log = getLogger(__name__) + +__all__ = ('RemoteShell', 'SecureShell', 'GlobusSecureShell') + + +class RemoteShell(LocalShell): + + def __init__(self, rsh='rsh', rcp='rcp', hostname='localhost', username=None, **kwargs): + super(RemoteShell, self).__init__(**kwargs) + self.rsh = rsh + self.rcp = rcp + self.hostname = hostname + self.username = username + self.sessions = {} + + def execute(self, cmd, persist=False, timeout=60): + # TODO: implement persistence + if self.username is None: + fullcmd = '%s %s %s' % (self.rsh, self.hostname, cmd) + else: + fullcmd = '%s -l %s %s %s' % (self.rsh, self.username, self.hostname, cmd) + return super(RemoteShell, self).execute(fullcmd, persist, timeout) + + +class SecureShell(RemoteShell): + SSH_NEW_KEY_STRING = 'Are you sure you want to continue connecting' + + def __init__(self, rsh='ssh', rcp='scp', **kwargs): + rsh += ' -oStrictHostKeyChecking=yes -oConnectTimeout=60' + rcp += ' -oStrictHostKeyChecking=yes -oConnectTimeout=60' + super(SecureShell, self).__init__(rsh=rsh, rcp=rcp, **kwargs) + + +class GlobusSecureShell(SecureShell): + + def __init__(self, rsh='gsissh', rcp='gsiscp', **kwargs): + super(GlobusSecureShell, self).__init__(rsh=rsh, rcp=rcp, **kwargs) diff -r a014c2d841a8490f017086c687997a191f3a602f -r 9eb2b42e21fee519b1ea5e9c791bb6b8989f6b5c lib/galaxy/jobs/runners/util/kill.py --- /dev/null +++ b/lib/galaxy/jobs/runners/util/kill.py @@ -0,0 +1,65 @@ +import os +from platform import system +from time import sleep +from subprocess import Popen + +try: + from psutil import Process, NoSuchProcess +except ImportError: + """ Don't make psutil a strict requirement, but use if available. """ + Process = None + + +def kill_pid(pid, use_psutil=True): + if use_psutil and Process: + _psutil_kill_pid(pid) + else: + _stock_kill_pid(pid) + + +def _psutil_kill_pid(pid): + """ + http://stackoverflow.com/questions/1230669/subprocess-deleting-child-proces… + """ + try: + parent = Process(pid) + for child in parent.get_children(recursive=True): + child.kill() + parent.kill() + except NoSuchProcess: + return + + +def _stock_kill_pid(pid): + is_windows = system() == 'Windows' + + if is_windows: + __kill_windows(pid) + else: + __kill_posix(pid) + + +def __kill_windows(pid): + try: + Popen("taskkill /F /T /PID %i" % pid, shell=True) + except Exception: + pass + + +def __kill_posix(pid): + def __check_pid(): + try: + os.kill(pid, 0) + return True + except OSError: + return False + + if __check_pid(): + for sig in [15, 9]: + try: + os.killpg(pid, sig) + except OSError: + return + sleep(1) + if not __check_pid(): + return Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: 6 new changesets
by commits-noreply＠bitbucket.org 07 May '14

07 May '14

6 new commits in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/fef831b5e2bd/ Changeset: fef831b5e2bd User: nsoranzo Date: 2014-05-05 19:07:07 Summary: Update code checks and documentation after commit 17a2dc7. Affected #: 1 file diff -r fe6c48de949dd4e640d3632ab23cfe6ab5fa0494 -r fef831b5e2bd3feaaef58212eb695943d8969212 lib/galaxy/webapps/galaxy/api/workflows.py --- a/lib/galaxy/webapps/galaxy/api/workflows.py +++ b/lib/galaxy/webapps/galaxy/api/workflows.py @@ -1,9 +1,9 @@ -from __future__ import absolute_import - """ API operations for Workflows """ +from __future__ import absolute_import + import logging from sqlalchemy import desc, or_ from galaxy import exceptions @@ -12,7 +12,6 @@ from galaxy.web import _future_expose_api as expose_api from galaxy.web.base.controller import BaseAPIController, url_for, UsesStoredWorkflowMixin from galaxy.web.base.controller import UsesHistoryMixin -from galaxy.workflow.modules import module_factory from galaxy.workflow.run import invoke from galaxy.workflow.run import WorkflowRunConfig from galaxy.workflow.extract import extract_workflow @@ -120,76 +119,79 @@ workflow will be created for this user. Otherwise, workflow_id must be specified and this API method will cause a workflow to execute. - :param installed_repository_file The path of a workflow to import. Either workflow_id or installed_repository_file must be specified + :param installed_repository_file The path of a workflow to import. Either workflow_id, installed_repository_file or from_history_id must be specified :type installed_repository_file str - :param workflow_id: an existing workflow id. Either workflow_id or installed_repository_file must be specified + :param workflow_id: An existing workflow id. Either workflow_id, installed_repository_file or from_history_id must be specified :type workflow_id: str - :param parameters: See _update_step_parameters() + :param parameters: If workflow_id is set - see _update_step_parameters() :type parameters: dict - :param ds_map: A dictionary mapping each input step id to a dictionary with 2 keys: 'src' (which can be 'ldda', 'ld' or 'hda') and 'id' (which should be the id of a LibraryDatasetDatasetAssociation, LibraryDataset or HistoryDatasetAssociation respectively) + :param ds_map: If workflow_id is set - a dictionary mapping each input step id to a dictionary with 2 keys: 'src' (which can be 'ldda', 'ld' or 'hda') and 'id' (which should be the id of a LibraryDatasetDatasetAssociation, LibraryDataset or HistoryDatasetAssociation respectively) :type ds_map: dict - :param no_add_to_history: if present in the payload with any value, the input datasets will not be added to the selected history + :param no_add_to_history: If workflow_id is set - if present in the payload with any value, the input datasets will not be added to the selected history :type no_add_to_history: str - :param history: Either the name of a new history or "hist_id=HIST_ID" where HIST_ID is the id of an existing history + :param history: If workflow_id is set - optional history where to run the workflow, either the name of a new history or "hist_id=HIST_ID" where HIST_ID is the id of an existing history. If not specified, the workflow will be run a new unnamed history :type history: str - :param replacement_params: A dictionary used when renaming datasets + :param replacement_params: If workflow_id is set - an optional dictionary used when renaming datasets :type replacement_params: dict - :param from_history_id: Id of history to extract a workflow from. Should not be used with worfklow_id or installed_repository_file. + :param from_history_id: Id of history to extract a workflow from. Either workflow_id, installed_repository_file or from_history_id must be specified :type from_history_id: str - :param job_ids: If from_history_id is set - this should be a list of jobs to include when extracting workflow from history. + :param job_ids: If from_history_id is set - optional list of jobs to include when extracting workflow from history. :type job_ids: str - :param dataset_ids: If from_history_id is set - this should be a list of HDA ids corresponding to workflow inputs when extracting workflow from history. + :param dataset_ids: If from_history_id is set - optional list of HDA ids corresponding to workflow inputs when extracting workflow from history. :type dataset_ids: str + + :param workflow_name: If from_history_id is set - name of the workflow to create + :type workflow_name: str """ - # Pull parameters out of payload. - workflow_id = payload.get('workflow_id', None) - param_map = payload.get('parameters', {}) - ds_map = payload.get('ds_map', {}) + if len( set( ['workflow_id', 'installed_repository_file', 'from_history_id'] ).intersection( payload ) ) > 1: + trans.response.status = 403 + return "Only one among 'workflow_id', 'installed_repository_file', 'from_history_id' must be specified" + + if 'installed_repository_file' in payload: + workflow_controller = trans.webapp.controllers[ 'workflow' ] + result = workflow_controller.import_workflow( trans=trans, + cntrller='api', + **payload) + return result + + if 'from_history_id' in payload: + from_history_id = payload.get( 'from_history_id' ) + history = self.get_history( trans, from_history_id, check_ownership=False, check_accessible=True ) + job_ids = map( trans.security.decode_id, payload.get( 'job_ids', [] ) ) + dataset_ids = map( trans.security.decode_id, payload.get( 'dataset_ids', [] ) ) + workflow_name = payload[ 'workflow_name' ] + stored_workflow = extract_workflow( + trans=trans, + user=trans.get_user(), + history=history, + job_ids=job_ids, + dataset_ids=dataset_ids, + workflow_name=workflow_name, + ) + item = stored_workflow.to_dict( value_mapper={ 'id': trans.security.encode_id } ) + item[ 'url' ] = url_for( 'workflow', id=item[ 'id' ] ) + return item + + workflow_id = payload.get( 'workflow_id', None ) + if not workflow_id: + trans.response.status = 403 + return "Either workflow_id, installed_repository_file or from_history_id must be specified" + + # Pull other parameters out of payload. + param_map = payload.get( 'parameters', {} ) + ds_map = payload.get( 'ds_map', {} ) add_to_history = 'no_add_to_history' not in payload - history_param = payload.get('history', '') - - # Get/create workflow. - if not workflow_id: - # create new - if 'installed_repository_file' in payload: - workflow_controller = trans.webapp.controllers[ 'workflow' ] - result = workflow_controller.import_workflow( trans=trans, - cntrller='api', - **payload) - return result - if 'from_history_id' in payload: - from_history_id = payload.get( 'from_history_id' ) - history = self.get_history( trans, from_history_id, check_ownership=False, check_accessible=True ) - job_ids = map( trans.security.decode_id, payload.get( "job_ids", [] ) ) - dataset_ids = map( trans.security.decode_id, payload.get( "dataset_ids", [] ) ) - workflow_name = payload[ "workflow_name" ] - stored_workflow = extract_workflow( - trans=trans, - user=trans.get_user(), - history=history, - job_ids=job_ids, - dataset_ids=dataset_ids, - workflow_name=workflow_name, - ) - item = stored_workflow.to_dict( value_mapper={ "id": trans.security.encode_id } ) - item[ 'url' ] = url_for( 'workflow', id=item[ "id" ] ) - return item - - trans.response.status = 403 - return "Either workflow_id or installed_repository_file must be specified" - if 'installed_repository_file' in payload: - trans.response.status = 403 - return "installed_repository_file may not be specified with workflow_id" + history_param = payload.get( 'history', '' ) # Get workflow + accessibility check. stored_workflow = trans.sa_session.query(self.app.model.StoredWorkflow).get( https://bitbucket.org/galaxy/galaxy-central/commits/ef7bd8c935bc/ Changeset: ef7bd8c935bc User: nsoranzo Date: 2014-05-06 19:15:26 Summary: Merged galaxy/galaxy-central into default Affected #: 117 files diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 config/plugins/visualizations/charts/static/charts/heatmap/config.js --- a/config/plugins/visualizations/charts/static/charts/heatmap/config.js +++ b/config/plugins/visualizations/charts/static/charts/heatmap/config.js @@ -1,10 +1,12 @@ define([], function() { return { - title : 'Heatmap', - library : '', - tag : 'div', + title : 'Heatmap', + library : '', + tag : 'div', use_panels : true, + + // columns columns : { col_label : { title : 'Columns', @@ -19,6 +21,7 @@ }, }, + // settings settings: { color_set : { title : 'Color scheme', @@ -107,6 +110,38 @@ value : 'wysiwyg' } ] + }, + + sorting : { + title : 'Sorting', + info : 'How should the columns be clustered?', + type : 'select', + init : 'hclust', + data : [ + { + label : 'Read from dataset', + value : 'hclust' + }, + { + label : 'Sort column and row labels', + value : 'byboth' + }, + { + label : 'Sort column labels', + value : 'bycolumns' + }, + { + label : 'Sort by rows', + value : 'byrow' + } + ] + } + }, + + // menu definition + menu : function() { + return { + color_set : this.settings.color_set } } }; diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 config/plugins/visualizations/charts/static/charts/heatmap/heatmap-plugin.js --- a/config/plugins/visualizations/charts/static/charts/heatmap/heatmap-plugin.js +++ b/config/plugins/visualizations/charts/static/charts/heatmap/heatmap-plugin.js @@ -180,26 +180,22 @@ // // add ui elements // - // create ui elements + // create tooltip this.$tooltip = $(this._templateTooltip()); - this.$select = $(this._templateSelect()); - - // append this.$el.append(this.$tooltip); - this.$el.append(this.$select); - - // add event to select field - this.$select.on('change', function(){ - self._sortByOrder(this.value); - }); // // finally draw the heatmap // - this._draw(); + this._build(); + + // catch window resize event + $(window).resize(function () { + self._build(); + }); }, - _draw: function() { + _build: function() { // link this var self = this; @@ -212,7 +208,7 @@ this.height = this.heightContainer; // calculate cell size - this.cellSize = Math.min(((this.height - 50) / (this.rowNumber + this.margin.top + this.margin.bottom)), + this.cellSize = Math.min(((this.height) / (this.rowNumber + this.margin.top + this.margin.bottom)), (this.width / (this.colNumber + this.margin.left + this.margin.right))); // set font size @@ -227,22 +223,27 @@ var width = this.width; var height = this.height; + // reset svg + if (this.svg !== undefined) { + this.$el.find('svg').remove(); + } + // add main group and translate this.svg = d3.select(this.$el[0]).append('svg') .append('g') .attr('transform', 'translate(' + (this.widthContainer - width) / 2 + ',' + - (this.heightContainer - height) / 2 + ')') - + (this.heightContainer - height) / 2 + ')'); + // reset sorting this.rowSortOrder = false; this.colSortOrder = false; // build - this._buildRowLabels(); - this._buildColLabels(); - this._buildHeatMap(); - this._buildLegend(); - this._buildTitle(); + this.d3RowLabels = this._buildRowLabels(); + this.d3ColLabels = this._buildColLabels(); + this.d3HeatMap = this._buildHeatMap(); + this.d3Legend = this._buildLegend(); + this.d3Title = this._buildTitle(); }, // build title @@ -258,7 +259,7 @@ var title = this.options.title; // add title - this.svg.append('g') + return this.svg.append('g') .append('text') .attr('x', width / 2) .attr('y', height + 3 * cellSize + fontSize + 3) @@ -323,6 +324,9 @@ }) .attr('y', height + cellSize - 3) .style('font-size', fontSize + 'px'); + + // return + return legend; }, // build column labels @@ -366,6 +370,9 @@ self._sortByLabel('c', 'row', self.rowNumber, i, self.colSortOrder); d3.select('#order').property('selectedIndex', 4).node().focus(); }); + + // return + return colLabels; }, // build row labels @@ -409,6 +416,9 @@ self._sortByLabel('r', 'col', self.colNumber, i, self.rowSortOrder); d3.select('#order').property('selectedIndex', 4).node().focus(); }); + + // return + return rowLabels; }, // build heat map @@ -424,7 +434,7 @@ var colLabel = this.colLabel; // heat map - var heatMap = this.svg.append('g').attr('class','g3') + var heatmap = this.svg.append('g').attr('class','g3') .selectAll('.cellg') .data(self.data, function(d) { return d.row + ':' + d.col; @@ -466,6 +476,9 @@ d3.selectAll('.colLabel').classed('text-highlight',false); d3.select('#heatmap-tooltip').classed('hidden', true); }); + + // return + return heatmap; }, // change ordering of cells diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 config/plugins/visualizations/charts/static/views/viewport.js --- a/config/plugins/visualizations/charts/static/views/viewport.js +++ b/config/plugins/visualizations/charts/static/views/viewport.js @@ -6,7 +6,8 @@ return Backbone.View.extend({ // list of canvas elements - canvas: [], + container_list: [], + canvas_list: [], // initialize initialize: function(app, options) { @@ -26,7 +27,7 @@ this._fullscreen(this.$el, 80); // create canvas element - this._createCanvas('div'); + this._createContainer('div'); // events var self = this; @@ -38,6 +39,7 @@ this.chart.on('set:state', function() { // get info element var $info = self.$el.find('#info'); + var $container = self.$el.find('container'); // get icon var $icon = $info.find('#icon'); @@ -49,11 +51,15 @@ $info.show(); $info.find('#text').html(self.chart.get('state_info')); + // hide containers + $container.hide(); + // check status var state = self.chart.get('state'); switch (state) { case 'ok': $info.hide(); + $container.show() break; case 'failed': $icon.addClass('fa fa-warning'); @@ -86,29 +92,35 @@ }, // creates n canvas elements - _createCanvas: function(tag, n) { + _createContainer: function(tag, n) { // check size of requested canvas elements n = n || 1; // clear previous canvas elements - for (var i in this.canvas) { - this.canvas[i].remove(); - this.canvas.slice(i, 0); + for (var i in this.container_list) { + this.container_list[i].remove(); } + // reset lists + this.container_list = []; + this.canvas_list = []; + // create requested canvas elements for (var i = 0; i < n; i++) { // create element - var canvas_el = $(this._templateCanvas(tag, parseInt(100 / n))); + var canvas_el = $(this._templateContainer(tag, parseInt(100 / n))); // add to view this.$el.append(canvas_el); - // find canvas element + // add to list + this.container_list[i] = canvas_el; + + // add a separate list for canvas elements if (tag == 'svg') { - this.canvas[i] = d3.select(canvas_el[0]); + this.canvas_list[i] = d3.select(canvas_el.find('#canvas')[0]); } else { - this.canvas[i] = canvas_el; + this.canvas_list[i] = canvas_el.find('#canvas'); } } }, @@ -137,7 +149,7 @@ } // create canvas element and add to canvas list - this._createCanvas(this.chart_settings.tag, n_panels); + this._createContainer(this.chart_settings.tag, n_panels); // set chart state chart.state('wait', 'Please wait...'); @@ -145,6 +157,7 @@ // clean up data if there is any from previous jobs if (!this.chart_settings.execute || (this.chart_settings.execute && chart.get('modified'))) { + // reset jobs this.app.jobs.cleanup(chart); @@ -156,7 +169,7 @@ var self = this; require(['plugin/charts/' + chart_type + '/' + chart_type], function(ChartView) { // create chart - var view = new ChartView(self.app, {canvas : self.canvas}); + var view = new ChartView(self.app, {canvas : self.canvas_list}); // request data if (self.chart_settings.execute) { @@ -269,8 +282,11 @@ }, // template svg/div element - _templateCanvas: function(tag, width) { - return '<' + tag + ' class="canvas" style="float: left; display: block; width:' + width + '%; height: 100%;"/>'; + _templateContainer: function(tag, width) { + return '<div class="container" style="float: left; display: block; width:' + width + '%; height: 100%;">' + + '<div id="menu"/>' + + '<' + tag + ' id="canvas" class="canvas" style="display: block; width:100%; height: inherit;">' + + '</div>'; } }); diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/app.py --- a/lib/galaxy/app.py +++ b/lib/galaxy/app.py @@ -5,6 +5,7 @@ from galaxy import config, jobs import galaxy.model import galaxy.security +from galaxy import dataset_collections import galaxy.quota from galaxy.tags.tag_handler import GalaxyTagHandler from galaxy.visualization.genomes import Genomes @@ -54,6 +55,8 @@ self._configure_security() # Tag handler self.tag_handler = GalaxyTagHandler() + # Dataset Collection Plugins + self.dataset_collections_service = dataset_collections.DatasetCollectionsService(self) # Genomes self.genomes = Genomes( self ) # Data providers registry. diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/dataset_collections/__init__.py --- /dev/null +++ b/lib/galaxy/dataset_collections/__init__.py @@ -0,0 +1,274 @@ +from .registry import DatasetCollectionTypesRegistry +from .matching import MatchingCollections +from .type_description import CollectionTypeDescriptionFactory + +from galaxy import model +from galaxy.exceptions import MessageException +from galaxy.exceptions import ItemAccessibilityException +from galaxy.exceptions import RequestParameterInvalidException +from galaxy.web.base.controller import ( + UsesHistoryDatasetAssociationMixin, + UsesLibraryMixinItems, + UsesTagsMixin, +) +from galaxy.managers import hdas # TODO: Refactor all mixin use into managers. + +from galaxy.util import validation +from galaxy.util import odict + +import logging +log = logging.getLogger( __name__ ) + + +ERROR_INVALID_ELEMENTS_SPECIFICATION = "Create called with invalid parameters, must specify element identifiers." +ERROR_NO_COLLECTION_TYPE = "Create called without specifing a collection type." + + +class DatasetCollectionsService( + UsesHistoryDatasetAssociationMixin, + UsesLibraryMixinItems, + UsesTagsMixin, +): + """ + Abstraction for interfacing with dataset collections instance - ideally abstarcts + out model and plugin details. + """ + + def __init__( self, app ): + self.type_registry = DatasetCollectionTypesRegistry( app ) + self.collection_type_descriptions = CollectionTypeDescriptionFactory( self.type_registry ) + self.model = app.model + self.security = app.security + self.hda_manager = hdas.HDAManager() + + def create( + self, + trans, + parent, # PRECONDITION: security checks on ability to add to parent occurred during load. + name, + collection_type, + element_identifiers=None, + elements=None, + implicit_collection_info=None, + ): + """ + """ + dataset_collection = self.__create_dataset_collection( + trans=trans, + collection_type=collection_type, + element_identifiers=element_identifiers, + elements=elements, + ) + if isinstance( parent, model.History ): + dataset_collection_instance = self.model.HistoryDatasetCollectionAssociation( + collection=dataset_collection, + name=name, + ) + if implicit_collection_info: + for input_name, input_collection in implicit_collection_info[ "implicit_inputs" ]: + dataset_collection_instance.add_implicit_input_collection( input_name, input_collection ) + for output_dataset in implicit_collection_info.get( "outputs_datasets" ): + output_dataset.hidden_beneath_collection_instance = dataset_collection_instance + trans.sa_session.add( output_dataset ) + + dataset_collection_instance.implicit_output_name = implicit_collection_info[ "implicit_output_name" ] + # Handle setting hid + parent.add_dataset_collection( dataset_collection_instance ) + elif isinstance( parent, model.LibraryFolder ): + dataset_collection_instance = self.model.LibraryDatasetCollectionAssociation( + collection=dataset_collection, + folder=parent, + name=name, + ) + else: + message = "Internal logic error - create called with unknown parent type %s" % type( parent ) + log.exception( message ) + raise MessageException( message ) + + return self.__persist( dataset_collection_instance ) + + def __create_dataset_collection( + self, + trans, + collection_type, + element_identifiers=None, + elements=None, + ): + if element_identifiers is None and elements is None: + raise RequestParameterInvalidException( ERROR_INVALID_ELEMENTS_SPECIFICATION ) + if not collection_type: + raise RequestParameterInvalidException( ERROR_NO_COLLECTION_TYPE ) + collection_type_description = self.collection_type_descriptions.for_collection_type( collection_type ) + if elements is None: + if collection_type_description.has_subcollections( ): + # Nested collection - recursively create collections and update identifiers. + self.__recursively_create_collections( trans, element_identifiers ) + elements = self.__load_elements( trans, element_identifiers ) + # else if elements is set, it better be an ordered dict! + + type_plugin = collection_type_description.rank_type_plugin() + dataset_collection = type_plugin.build_collection( elements ) + dataset_collection.collection_type = collection_type + return dataset_collection + + def delete( self, trans, instance_type, id ): + dataset_collection_instance = self.get_dataset_collection_instance( trans, instance_type, id, check_ownership=True ) + dataset_collection_instance.deleted = True + trans.sa_session.add( dataset_collection_instance ) + trans.sa_session.flush( ) + + def update( self, trans, instance_type, id, payload ): + dataset_collection_instance = self.get_dataset_collection_instance( trans, instance_type, id, check_ownership=True ) + if trans.user is None: + anon_allowed_payload = {} + if 'deleted' in payload: + anon_allowed_payload[ 'deleted' ] = payload[ 'deleted' ] + if 'visible' in payload: + anon_allowed_payload[ 'visible' ] = payload[ 'visible' ] + payload = self._validate_and_parse_update_payload( anon_allowed_payload ) + else: + payload = self._validate_and_parse_update_payload( payload ) + changed = self._set_from_dict( trans, dataset_collection_instance, payload ) + return changed + + def _set_from_dict( self, trans, dataset_collection_instance, new_data ): + # Blatantly stolen from UsesHistoryDatasetAssociationMixin.set_hda_from_dict. + + # send what we can down into the model + changed = dataset_collection_instance.set_from_dict( new_data ) + # the rest (often involving the trans) - do here + if 'annotation' in new_data.keys() and trans.get_user(): + dataset_collection_instance.add_item_annotation( trans.sa_session, trans.get_user(), dataset_collection_instance.collection, new_data[ 'annotation' ] ) + changed[ 'annotation' ] = new_data[ 'annotation' ] + if 'tags' in new_data.keys() and trans.get_user(): + self.set_tags_from_list( trans, dataset_collection_instance.collection, new_data[ 'tags' ], user=trans.user ) + + if changed.keys(): + trans.sa_session.flush() + + return changed + + def _validate_and_parse_update_payload( self, payload ): + validated_payload = {} + for key, val in payload.items(): + if val is None: + continue + if key in ( 'name' ): + val = validation.validate_and_sanitize_basestring( key, val ) + validated_payload[ key ] = val + if key in ( 'deleted', 'visible' ): + validated_payload[ key ] = validation.validate_boolean( key, val ) + elif key == 'tags': + validated_payload[ key ] = validation.validate_and_sanitize_basestring_list( key, val ) + return validated_payload + + def history_dataset_collections(self, history, query): + collections = history.dataset_collections + collections = filter( query.direct_match, collections ) + return collections + + def __persist( self, dataset_collection_instance ): + context = self.model.context + context.add( dataset_collection_instance ) + context.flush() + return dataset_collection_instance + + def __recursively_create_collections( self, trans, element_identifiers ): + # TODO: Optimize - don't recheck parent, reload created model, just use as is. + for index, element_identifier in enumerate( element_identifiers ): + try: + if not element_identifier[ "src" ] == "new_collection": + # not a new collection, keep moving... + continue + except KeyError: + # Not a dictionary, just an id of an HDA - move along. + continue + + # element identifier is a dict with src new_collection... + collection_type = element_identifier.get( "collection_type", None ) + if not collection_type: + raise RequestParameterInvalidException( "No collection_type define for nested collection." ) + collection = self.__create_dataset_collection( + trans=trans, + collection_type=collection_type, + element_identifiers=element_identifier[ "element_identifiers" ], + ) + self.__persist( collection ) + element_identifier[ "src" ] = "dc" + element_identifier[ "id" ] = trans.security.encode_id( collection.id ) + + return element_identifiers + + def __load_elements( self, trans, element_identifiers ): + elements = odict.odict() + for element_identifier in element_identifiers: + elements[ element_identifier[ "name" ] ] = self.__load_element( trans, element_identifier ) + return elements + + def __load_element( self, trans, element_identifier ): + #if not isinstance( element_identifier, dict ): + # # Is allowing this to just be the id of an hda too clever? Somewhat + # # consistent with other API methods though. + # element_identifier = dict( src='hda', id=str( element_identifier ) ) + + # dateset_identifier is dict {src=hda|ldda, id=<encoded_id>} + try: + src_type = element_identifier.get( 'src', 'hda' ) + except AttributeError: + raise MessageException( "Dataset collection element definition (%s) not dictionary-like." % element_identifier ) + encoded_id = element_identifier.get( 'id', None ) + if not src_type or not encoded_id: + raise RequestParameterInvalidException( "Problem decoding element identifier %s" % element_identifier ) + + if src_type == 'hda': + decoded_id = int( trans.app.security.decode_id( encoded_id ) ) + element = self.hda_manager.get( trans, decoded_id, check_ownership=False ) + elif src_type == 'ldda': + element = self.get_library_dataset_dataset_association( trans, encoded_id ) + elif src_type == 'hdca': + # TODO: Option to copy? Force copy? Copy or allow if not owned? + element = self.__get_history_collection_instance( trans, encoded_id ).collection + # TODO: ldca. + elif src_type == "dc": + # TODO: Force only used internally during nested creation so no + # need to recheck security. + element = self.get_dataset_collection( trans, encoded_id ) + else: + raise RequestParameterInvalidException( "Unknown src_type parameter supplied '%s'." % src_type ) + return element + + def match_collections( self, collections_to_match ): + """ + May seem odd to place it here, but planning to grow sophistication and + get plugin types involved so it will likely make sense in the future. + """ + return MatchingCollections.for_collections( collections_to_match, self.collection_type_descriptions ) + + def get_dataset_collection_instance( self, trans, instance_type, id, **kwds ): + """ + """ + if instance_type == "history": + return self.__get_history_collection_instance( trans, id, **kwds ) + elif instance_type == "library": + return self.__get_library_collection_instance( trans, id, **kwds ) + + def get_dataset_collection( self, trans, encoded_id ): + collection_id = int( trans.app.security.decode_id( encoded_id ) ) + collection = trans.sa_session.query( trans.app.model.DatasetCollection ).get( collection_id ) + return collection + + def __get_history_collection_instance( self, trans, id, check_ownership=False, check_accessible=True ): + instance_id = int( trans.app.security.decode_id( id ) ) + collection_instance = trans.sa_session.query( trans.app.model.HistoryDatasetCollectionAssociation ).get( instance_id ) + self.security_check( trans, collection_instance.history, check_ownership=check_ownership, check_accessible=check_accessible ) + return collection_instance + + def __get_library_collection_instance( self, trans, id, check_ownership=False, check_accessible=True ): + if check_ownership: + raise NotImplemented( "Functionality (getting library dataset collection with ownership check) unimplemented." ) + instance_id = int( trans.security.decode_id( id ) ) + collection_instance = trans.sa_session.query( trans.app.model.LibraryDatasetCollectionAssociation ).get( instance_id ) + if check_accessible: + if not trans.app.security_agent.can_access_library_item( trans.get_current_user_roles(), collection_instance, trans.user ): + raise ItemAccessibilityException( "LibraryDatasetCollectionAssociation is not accessible to the current user", type='error' ) + return collection_instance diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/dataset_collections/matching.py --- /dev/null +++ b/lib/galaxy/dataset_collections/matching.py @@ -0,0 +1,68 @@ +from galaxy.util import bunch +from galaxy import exceptions +from .structure import get_structure + +CANNOT_MATCH_ERROR_MESSAGE = "Cannot match collection types." + + +class CollectionsToMatch( object ): + """ Structure representing a set of collections that need to be matched up + when running tools (possibly workflows in the future as well). + """ + + def __init__( self ): + self.collections = {} + + def add( self, input_name, hdca, subcollection_type=None ): + self.collections[ input_name ] = bunch.Bunch( + hdca=hdca, + subcollection_type=subcollection_type, + ) + + def has_collections( self ): + return len( self.collections ) > 0 + + def iteritems( self ): + return self.collections.iteritems() + + +class MatchingCollections( object ): + """ Structure holding the result of matching a list of collections + together. This class being different than the class above and being + created in the dataset_collections_service layer may seem like + overkill but I suspect in the future plugins will be subtypable for + instance so matching collections will need to make heavy use of the + dataset collection type registry managed by the dataset collections + sevice - hence the complexity now. + """ + + def __init__( self ): + self.structure = None + self.collections = {} + + def __attempt_add_to_match( self, input_name, hdca, collection_type_description, subcollection_type ): + structure = get_structure( hdca, collection_type_description, leaf_subcollection_type=subcollection_type ) + if not self.structure: + self.structure = structure + self.collections[ input_name ] = hdca + else: + if not self.structure.can_match( structure ): + raise exceptions.MessageException( CANNOT_MATCH_ERROR_MESSAGE ) + self.collections[ input_name ] = hdca + + def slice_collections( self ): + return self.structure.walk_collections( self.collections ) + + @staticmethod + def for_collections( collections_to_match, collection_type_descriptions ): + if not collections_to_match.has_collections(): + return None + + matching_collections = MatchingCollections() + for input_key, to_match in collections_to_match.iteritems(): + hdca = to_match.hdca + subcollection_type = to_match = to_match.subcollection_type + collection_type_description = collection_type_descriptions.for_collection_type( hdca.collection.collection_type ) + matching_collections.__attempt_add_to_match( input_key, hdca, collection_type_description, subcollection_type ) + + return matching_collections diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/dataset_collections/registry.py --- /dev/null +++ b/lib/galaxy/dataset_collections/registry.py @@ -0,0 +1,14 @@ +from .types import list +from .types import paired + + +PLUGIN_CLASSES = [list.ListDatasetCollectionType, paired.PairedDatasetCollectionType] + + +class DatasetCollectionTypesRegistry(object): + + def __init__( self, app ): + self.__plugins = dict( [ ( p.collection_type, p() ) for p in PLUGIN_CLASSES ] ) + + def get( self, plugin_type ): + return self.__plugins[ plugin_type ] diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/dataset_collections/structure.py --- /dev/null +++ b/lib/galaxy/dataset_collections/structure.py @@ -0,0 +1,105 @@ +""" Module for reasoning about structure of and matching hierarchical collections of data. +""" +import logging +log = logging.getLogger( __name__ ) + + +class Leaf( object ): + + def __len__( self ): + return 1 + + @property + def is_leaf( self ): + return True + +leaf = Leaf() + + +class Tree( object ): + + def __init__( self, dataset_collection, collection_type_description, leaf_subcollection_type ): + self.collection_type_description = collection_type_description + self.leaf_subcollection_type = leaf_subcollection_type # collection_type to trim tree at... + children = [] + for element in dataset_collection.elements: + child_collection = element.child_collection + if child_collection: + subcollection_type_description = collection_type_description.subcollection_type_description() # Type description of children + if subcollection_type_description.can_match_type( leaf_subcollection_type ): + children.append( ( element.element_identifier, leaf ) ) + else: + children.append( ( element.element_identifier, Tree( child_collection, collection_type_description=subcollection_type_description, leaf_subcollection_type=leaf_subcollection_type ) ) ) + elif element.hda: + children.append( ( element.element_identifier, leaf ) ) + + self.children = children + + def walk_collections( self, hdca_dict ): + return self._walk_collections( dict_map( lambda hdca: hdca.collection, hdca_dict ) ) + + def _walk_collections( self, collection_dict ): + for ( identifier, substructure ) in self.children: + def element( collection ): + return collection[ identifier ] + + if substructure.is_leaf: + yield dict_map( element, collection_dict ) + else: + sub_collections = dict_map( lambda collection: element( collection ).child_collection ) + for element in substructure._walk_collections( sub_collections ): + yield element + + @property + def is_leaf( self ): + return False + + def can_match( self, other_structure ): + if not self.collection_type_description.can_match_type( other_structure.collection_type_description ): + # TODO: generalize + return False + + if len( self.children ) != len( other_structure.children ): + return False + + for my_child, other_child in zip( self.children, other_structure.children ): + if my_child[ 0 ] != other_child[ 0 ]: # Different identifiers, TODO: generalize + return False + + # At least one is nested collection... + if my_child[ 1 ].is_leaf != other_child[ 1 ].is_leaf: + return False + + if not my_child[ 1 ].is_leaf and not my_child[ 1 ].can_match( other_child[ 1 ]): + return False + + return True + + def __len__( self ): + return sum( [ len( c[ 1 ] ) for c in self.children ] ) + + def element_identifiers_for_datasets( self, trans, datasets ): + element_identifiers = [] + for identifier, child in self.children: + if isinstance( child, Tree ): + child_identifiers = child.element_identifiers_for_datasets( trans, datasets[ 0:len( child ) ] ) + child_identifiers[ "name" ] = identifier + element_identifiers.append( child_identifiers ) + else: + element_identifiers.append( dict( name=identifier, src="hda", id=trans.security.encode_id( datasets[ 0 ].id ) ) ) + + datasets = datasets[ len( child ): ] + + return dict( + src="new_collection", + collection_type=self.collection_type_description.collection_type, + element_identifiers=element_identifiers, + ) + + +def dict_map( func, input_dict ): + return dict( [ ( k, func(v) ) for k, v in input_dict.iteritems() ] ) + + +def get_structure( dataset_collection_instance, collection_type_description, leaf_subcollection_type=None ): + return Tree( dataset_collection_instance.collection, collection_type_description, leaf_subcollection_type=leaf_subcollection_type ) diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/dataset_collections/subcollections.py --- /dev/null +++ b/lib/galaxy/dataset_collections/subcollections.py @@ -0,0 +1,25 @@ +from galaxy import exceptions + + +def split_dataset_collection_instance( dataset_collection_instance, collection_type ): + """ Split up collection into collection. + """ + return _split_dataset_collection( dataset_collection_instance.collection, collection_type ) + + +def _split_dataset_collection( dataset_collection, collection_type ): + this_collection_type = dataset_collection.collection_type + if not this_collection_type.endswith( collection_type ) or this_collection_type == collection_type: + raise exceptions.MessageException( "Cannot split collection in desired fashion." ) + + split_elements = [] + for element in dataset_collection.elements: + child_collection = element.child_collection + if child_collection is None: + raise exceptions.MessageException( "Cannot split collection in desired fashion." ) + if child_collection.collection_type == collection_type: + split_elements.append( element ) + else: + split_elements.extend( _split_dataset_collection( element.child_collection ) ) + + return split_elements diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/dataset_collections/type_description.py --- /dev/null +++ b/lib/galaxy/dataset_collections/type_description.py @@ -0,0 +1,89 @@ + + +class CollectionTypeDescriptionFactory( object ): + + def __init__( self, type_registry ): + # taking in type_registry though not using it, because we will someday + # I think. + self.type_registry = type_registry + + def for_collection_type( self, collection_type ): + return CollectionTypeDescription( collection_type, self ) + + +class CollectionTypeDescription( object ): + """ Abstraction over dataset collection type that ties together string + reprentation in database/model with type registry. + + + >>> nested_type_description = CollectionTypeDescription( "list:paired", None ) + >>> paired_type_description = CollectionTypeDescription( "paired", None ) + >>> nested_type_description.has_subcollections_of_type( "list" ) + False + >>> nested_type_description.has_subcollections_of_type( "list:paired" ) + False + >>> nested_type_description.has_subcollections_of_type( "paired" ) + True + >>> nested_type_description.has_subcollections_of_type( paired_type_description ) + True + >>> nested_type_description.has_subcollections( ) + True + >>> paired_type_description.has_subcollections( ) + False + >>> paired_type_description.rank_collection_type() + 'paired' + >>> nested_type_description.rank_collection_type() + 'list' + """ + + def __init__( self, collection_type, collection_type_description_factory ): + self.collection_type = collection_type + self.collection_type_description_factory = collection_type_description_factory + self.__has_subcollections = self.collection_type.find( ":" ) > 0 + + def has_subcollections_of_type( self, other_collection_type ): + """ Take in another type (either flat string or another + CollectionTypeDescription) and determine if this collection contains + subcollections matching that type. + + The way this is used in map/reduce it seems to make the most sense + for this to return True if these subtypes are proper (i.e. a type + is not considered to have subcollections of its own type). + """ + if hasattr( other_collection_type, 'collection_type' ): + other_collection_type = other_collection_type.collection_type + collection_type = self.collection_type + return collection_type.endswith( other_collection_type ) and collection_type != other_collection_type + + def is_subcollection_of_type( self, other_collection_type ): + if not hasattr( other_collection_type, 'collection_type' ): + other_collection_type = self.collection_type_description_factory.for_collection_type( other_collection_type ) + return other_collection_type.has_subcollections_of_type( self ) + + def can_match_type( self, other_collection_type ): + if hasattr( other_collection_type, 'collection_type' ): + other_collection_type = other_collection_type.collection_type + collection_type = self.collection_type + return other_collection_type == collection_type + + def subcollection_type_description( self ): + if not self.__has_subcollections: + raise ValueError( "Cannot generate subcollection type description for flat type %s" % self.collection_type ) + subcollection_type = self.collection_type.split( ":", 1 )[ 1 ] + return self.collection_type_description_factory.for_collection_type( subcollection_type ) + + def has_subcollections( self ): + return self.__has_subcollections + + def rank_collection_type( self ): + """ Return the top-level collection type corresponding to this + collection type. For instance the "rank" type of a list of paired + data ("list:paired") is "list". + """ + return self.collection_type.split( ":" )[ 0 ] + + def rank_type_plugin( self ): + return self.collection_type_description_factory.type_registry.get( self.rank_collection_type() ) + + def __str__( self ): + return "CollectionTypeDescription[%s]" % self.collection_type diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/dataset_collections/types/__init__.py --- /dev/null +++ b/lib/galaxy/dataset_collections/types/__init__.py @@ -0,0 +1,34 @@ +from galaxy import exceptions +from abc import ABCMeta +from abc import abstractmethod + +from galaxy import model + +import logging +log = logging.getLogger( __name__ ) + + +class DatasetCollectionType(object): + __metaclass__ = ABCMeta + + @abstractmethod + def build_collection( self, dataset_instances ): + """ + Build DatasetCollection with populated DatasetcollectionElement objects + corresponding to the supplied dataset instances or throw exception if + this is not a valid collection of the specified type. + """ + + +class BaseDatasetCollectionType( DatasetCollectionType ): + + def _validation_failed( self, message ): + raise exceptions.ObjectAttributeInvalidException( message ) + + def _new_collection_for_elements( self, elements ): + dataset_collection = model.DatasetCollection( ) + for index, element in enumerate( elements ): + element.element_index = index + element.collection = dataset_collection + dataset_collection.elements = elements + return dataset_collection diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/dataset_collections/types/list.py --- /dev/null +++ b/lib/galaxy/dataset_collections/types/list.py @@ -0,0 +1,23 @@ +from ..types import BaseDatasetCollectionType + +from galaxy.model import DatasetCollectionElement + + +class ListDatasetCollectionType( BaseDatasetCollectionType ): + """ A flat list of named elements. + """ + collection_type = "list" + + def __init__( self ): + pass + + def build_collection( self, elements ): + associations = [] + for identifier, element in elements.iteritems(): + association = DatasetCollectionElement( + element=element, + element_identifier=identifier, + ) + associations.append( association ) + + return self._new_collection_for_elements( associations ) diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/dataset_collections/types/paired.py --- /dev/null +++ b/lib/galaxy/dataset_collections/types/paired.py @@ -0,0 +1,31 @@ +from ..types import BaseDatasetCollectionType + +from galaxy.model import DatasetCollectionElement + +LEFT_IDENTIFIER = "left" +RIGHT_IDENTIFIER = "right" + + +class PairedDatasetCollectionType( BaseDatasetCollectionType ): + """ + Paired (left/right) datasets. + """ + collection_type = "paired" + + def __init__( self ): + pass + + def build_collection( self, elements ): + left_dataset = elements.get("left", None) + right_dataset = elements.get("right", None) + if not left_dataset or not right_dataset: + self._validation_failed("Paired instance must define 'left' and 'right' datasets .") + left_association = DatasetCollectionElement( + element=left_dataset, + element_identifier=LEFT_IDENTIFIER, + ) + right_association = DatasetCollectionElement( + element=right_dataset, + element_identifier=RIGHT_IDENTIFIER, + ) + return self._new_collection_for_elements([left_association, right_association]) diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/dataset_collections/util.py --- /dev/null +++ b/lib/galaxy/dataset_collections/util.py @@ -0,0 +1,52 @@ +from galaxy import exceptions +from galaxy import web +from galaxy import model + + +def api_payload_to_create_params( payload ): + """ + Cleanup API payload to pass into dataset_collections. + """ + required_parameters = [ "collection_type", "element_identifiers" ] + missing_parameters = [ p for p in required_parameters if p not in payload ] + if missing_parameters: + message = "Missing required parameters %s" % missing_parameters + raise exceptions.ObjectAttributeMissingException( message ) + + params = dict( + collection_type=payload.get( "collection_type" ), + element_identifiers=payload.get( "element_identifiers" ), + name=payload.get( "name", None ), + ) + + return params + + +def dictify_dataset_collection_instance( dataset_colleciton_instance, parent, security, view="element" ): + dict_value = dataset_colleciton_instance.to_dict( view=view ) + encoded_id = security.encode_id( dataset_colleciton_instance.id ) + if isinstance( parent, model.History ): + encoded_history_id = security.encode_id( parent.id ) + dict_value[ 'url' ] = web.url_for( 'history_content', history_id=encoded_history_id, id=encoded_id, type="dataset_collection" ) + elif isinstance( parent, model.LibraryFolder ): + encoded_library_id = security.encode_id( parent.library.id ) + encoded_folder_id = security.encode_id( parent.id ) + # TODO: Work in progress - this end-point is not right yet... + dict_value[ 'url' ] = web.url_for( 'library_content', library_id=encoded_library_id, id=encoded_id, folder_id=encoded_folder_id ) + if view == "element": + dict_value[ 'elements' ] = map( dictify_element, dataset_colleciton_instance.collection.elements ) + security.encode_all_ids( dict_value, recursive=True ) # TODO: Use Kyle's recusrive formulation of this. + return dict_value + + +def dictify_element( element ): + dictified = element.to_dict( view="element" ) + object_detials = element.element_object.to_dict() + if element.child_collection: + # Recursively yield elements for each nested collection... + object_detials[ "elements" ] = map( dictify_element, element.child_collection.elements ) + + dictified[ "object" ] = object_detials + return dictified + +__all__ = [ api_payload_to_create_params, dictify_dataset_collection_instance ] diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/exceptions/__init__.py --- a/lib/galaxy/exceptions/__init__.py +++ b/lib/galaxy/exceptions/__init__.py @@ -68,6 +68,10 @@ status_code = 400 err_code = error_codes.USER_REQUEST_MISSING_PARAMETER +class ToolMetaParameterException( MessageException ): + status_code = 400 + err_code = error_codes.USER_TOOL_META_PARAMETER_PROBLEM + class RequestParameterInvalidException( MessageException ): status_code = 400 err_code = error_codes.USER_REQUEST_INVALID_PARAMETER diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/exceptions/error_codes.json --- a/lib/galaxy/exceptions/error_codes.json +++ b/lib/galaxy/exceptions/error_codes.json @@ -55,6 +55,11 @@ "message": "The request contains unknown type of contents." }, { + "name": "USER_TOOL_META_PARAMETER_PROBLEM", + "code": 400011, + "message": "Supplied incorrect or incompatible tool meta parameters." + }, + { "name": "USER_NO_API_KEY", "code": 403001, "message": "API authentication required for this request" diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/jobs/__init__.py --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -964,6 +964,8 @@ trynum += 1 log.warning( 'Error accessing %s, will retry: %s', dataset.dataset.file_name, e ) time.sleep( 2 ) + if getattr( dataset, "hidden_beneath_collection_instance", None ): + dataset.visible = False dataset.blurb = 'done' dataset.peek = 'no peek' dataset.info = (dataset.info or '') diff -r fef831b5e2bd3feaaef58212eb695943d8969212 -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 lib/galaxy/model/__init__.py --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -315,6 +315,8 @@ self.parameters = [] self.input_datasets = [] self.output_datasets = [] + self.input_dataset_collections = [] + self.output_dataset_collections = [] self.input_library_datasets = [] self.output_library_datasets = [] self.state = Job.states.NEW @@ -454,6 +456,10 @@ self.input_datasets.append( JobToInputDatasetAssociation( name, dataset ) ) def add_output_dataset( self, name, dataset ): self.output_datasets.append( JobToOutputDatasetAssociation( name, dataset ) ) + def add_input_dataset_collection( self, name, dataset ): + self.input_dataset_collections.append( JobToInputDatasetCollectionAssociation( name, dataset ) ) + def add_output_dataset_collection( self, name, dataset ): + self.output_dataset_collections.append( JobToOutputDatasetCollectionAssociation( name, dataset ) ) def add_input_library_dataset( self, name, dataset ): self.input_library_datasets.append( JobToInputLibraryDatasetAssociation( name, dataset ) ) def add_output_library_dataset( self, name, dataset ): @@ -696,6 +702,19 @@ self.name = name self.dataset = dataset + +class JobToInputDatasetCollectionAssociation( object ): + def __init__( self, name, dataset ): + self.name = name + self.dataset = dataset + + +class JobToOutputDatasetCollectionAssociation( object ): + def __init__( self, name, dataset_collection ): + self.name = name + self.dataset_collection = dataset_collection + + class JobToInputLibraryDatasetAssociation( object ): def __init__( self, name, dataset ): self.name = name @@ -706,6 +725,13 @@ self.name = name self.dataset = dataset + +class ImplicitlyCreatedDatasetCollectionInput( object ): + def __init__( self, name, input_dataset_collection ): + self.name = name + self.input_dataset_collection = input_dataset_collection + + class PostJobAction( object ): def __init__( self, action_type, workflow_step, output_name = None, action_arguments = None): self.action_type = action_type @@ -907,6 +933,14 @@ self.datasets.append( dataset ) return dataset + def add_dataset_collection( self, history_dataset_collection, set_hid=True ): + if set_hid: + history_dataset_collection.hid = self._next_hid() + history_dataset_collection.history = self + # TODO: quota? + self.dataset_collections.append( history_dataset_collection ) + return history_dataset_collection + def copy( self, name=None, target_user=None, activatable=False, all_datasets=False ): """ Return a copy of this history using the given `name` and `target_user`. @@ -947,6 +981,19 @@ db_session.flush() # Copy annotation. self.copy_item_annotation( db_session, self.user, hda, target_user, new_hda ) + # Copy history dataset collections + if all_datasets: + hdcas = self.dataset_collections + else: + hdcas = self.active_dataset_collections + for hdca in hdcas: + new_hdca = hdca.copy( ) + new_history.add_dataset_collection( new_hdca, set_hid=False ) + db_session.add( new_hdca ) + db_session.flush() + # Copy annotation. + self.copy_item_annotation( db_session, self.user, hdca, target_user, new_hdca ) + new_history.hid_counter = self.hid_counter db_session.add( new_history ) db_session.flush() @@ -1045,6 +1092,12 @@ self._active_datasets_children_and_roles = query.all() return self._active_datasets_children_and_roles + @property + def active_contents( self ): + """ Return all active contents ordered by hid. + """ + return self.contents_iter( types=[ "dataset", "dataset_collection" ], deleted=False, visible=True ) + def contents_iter( self, **kwds ): """ Fetch filtered list of contents of history. @@ -1056,6 +1109,8 @@ iters = [] if 'dataset' in types: iters.append( self.__dataset_contents_iter( **kwds ) ) + if 'dataset_collection' in types: + iters.append( self.__collection_contents_iter( **kwds ) ) return galaxy.util.merge_sorted_iterables( operator.attrgetter( "hid" ), *iters ) def __dataset_contents_iter(self, **kwds): @@ -1085,6 +1140,9 @@ else: return query + def __collection_contents_iter( self, **kwds ): + return self.__filter_contents( HistoryDatasetCollectionAssociation, **kwds ) + def copy_tags_from(self,target_user,source_history): for src_shta in source_history.tags: new_shta = src_shta.copy() @@ -1932,6 +1990,7 @@ purged = hda.purged, visible = hda.visible, state = hda.state, + history_content_type=hda.history_content_type, file_size = int( hda.get_size() ), update_time = hda.update_time.isoformat(), data_type = hda.ext, @@ -2001,6 +2060,10 @@ return changed + @property + def history_content_type( self ): + return "dataset" + class HistoryDatasetAssociationDisplayAtAuthorization( object ): def __init__( self, hda=None, user=None, site=None ): @@ -2516,6 +2579,313 @@ try: os.unlink( self.file_name ) except Exception, e: print "Failed to purge associated file (%s) from disk: %s" % ( self.file_name, e ) + +DEFAULT_COLLECTION_NAME = "Unnamed Collection" + + +class DatasetCollection( object, Dictifiable, UsesAnnotations ): + """ + """ + dict_collection_visible_keys = ( 'id', 'name', 'collection_type' ) + dict_element_visible_keys = ( 'id', 'name', 'collection_type' ) + + def __init__( + self, + id=None, + collection_type=None, + ): + self.id = id + self.collection_type = collection_type + + @property + def dataset_instances( self ): + instances = [] + for element in self.elements: + if element.is_collection: + instances.extend( element.child_collection.dataset_instances ) + else: + instance = element.dataset_instance + instances.append( instance ) + return instances + + @property + def state( self ): + # TODO: DatasetCollection state handling... + return 'ok' + + def validate( self ): + if self.collection_type is None: + raise Exception("Each dataset collection must define a collection type.") + + def __getitem__( self, key ): + get_by_attribute = "element_index" if isinstance( key, int ) else "element_identifier" + for element in self.elements: + if getattr( element, get_by_attribute ) == key: + return element + error_message = "Dataset collection has no %s with key %s." % ( get_by_attribute, key ) + raise KeyError( error_message ) + + def copy( self ): + new_collection = DatasetCollection( + collection_type=self.collection_type, + ) + for element in self.elements: + element.copy_to_collection( new_collection ) + object_session( self ).add( new_collection ) + object_session( self ).flush() + return new_collection + + def set_from_dict( self, new_data ): + editable_keys = ( 'name' ) + changed = {} + + # unknown keys are ignored here + for key in [ k for k in new_data.keys() if k in editable_keys ]: + new_val = new_data[ key ] + old_val = self.__getattribute__( key ) + if new_val == old_val: + continue + + self.__setattr__( key, new_val ) + changed[ key ] = new_val + + return changed + + +class DatasetCollectionInstance( object, HasName ): + """ + """ + def __init__( + self, + collection=None, + deleted=False, + ): + # Relationships + self.collection = collection + # Since deleted property is shared between history and dataset collections, + # it could be on either table - some places in the code however it is convient + # it is on instance instead of collection. + self.deleted = deleted + + @property + def state( self ): + return self.collection.state + + def display_name( self ): + return self.get_display_name() + + def _base_to_dict( self, view ): + return dict( + id=self.id, + name=self.name, + collection_type=self.collection.collection_type, + type="collection", # contents type (distinguished from file or folder (in case of library)) + ) + + def set_from_dict( self, new_data ): + """ + Set object attributes to the values in dictionary new_data limiting + to only those keys in dict_element_visible_keys. + + Returns a dictionary of the keys, values that have been changed. + """ + # precondition: keys are proper, values are parsed and validated + changed = self.collection.set_from_dict( new_data ) + + # unknown keys are ignored here + for key in [ k for k in new_data.keys() if k in self.editable_keys ]: + new_val = new_data[ key ] + old_val = self.__getattribute__( key ) + if new_val == old_val: + continue + + self.__setattr__( key, new_val ) + changed[ key ] = new_val + + return changed + + +class HistoryDatasetCollectionAssociation( DatasetCollectionInstance, Dictifiable ): + """ Associates a DatasetCollection with a History. """ + editable_keys = ( 'name', 'deleted', 'visible' ) + + def __init__( + self, + id=None, + hid=None, + collection=None, + history=None, + name=None, + deleted=False, + visible=True, + copied_from_history_dataset_collection_association=None, + implicit_output_name=None, + implicit_input_collections=[], + ): + super( HistoryDatasetCollectionAssociation, self ).__init__( + collection=collection, + deleted=deleted, + ) + self.id = id + self.hid = hid + self.history = history + self.name = name + self.visible = visible + self.copied_from_history_dataset_collection_association = copied_from_history_dataset_collection_association + self.implicit_output_name = implicit_output_name + self.implicit_input_collections = implicit_input_collections + + @property + def history_content_type( self ): + return "dataset_collection" + + def to_dict( self, view='collection' ): + dict_value = dict( + hid=self.hid, + history_id=self.history.id, + history_content_type=self.history_content_type, + visible=self.visible, + deleted=self.deleted, + **self._base_to_dict(view=view) + ) + return dict_value + + def add_implicit_input_collection( self, name, history_dataset_collection ): + self.implicit_input_collections.append( ImplicitlyCreatedDatasetCollectionInput( name, history_dataset_collection) ) + + def find_implicit_input_collection( self, name ): + matching_collection = None + for implicit_input_collection in self.implicit_input_collections: + if implicit_input_collection.name == name: + matching_collection = implicit_input_collection.input_dataset_collection + break + return matching_collection + + def copy( self ): + """ + Create a copy of this history dataset collection association. Copy + underlying collection. + """ + hdca = HistoryDatasetCollectionAssociation( + hid=self.hid, + collection=self.collection.copy(), + visible=self.visible, + deleted=self.deleted, + name=self.name, + copied_from_history_dataset_collection_association=self, + ) + + object_session( self ).add( hdca ) + object_session( self ).flush() + return hdca + + +class LibraryDatasetCollectionAssociation( DatasetCollectionInstance, Dictifiable ): + """ Associates a DatasetCollection with a library folder. """ + editable_keys = ( 'name', 'deleted' ) + + def __init__( + self, + id=None, + collection=None, + name=None, + deleted=False, + folder=None, + ): + super(LibraryDatasetCollectionAssociation, self).__init__( + collection=collection, + deleted=deleted, + ) + self.id = id + self.folder = folder + self.name = name + + def to_dict( self, view='collection' ): + dict_value = dict( + folder_id=self.folder.id, + **self._base_to_dict(view=view) + ) + return dict_value + + +class DatasetCollectionElement( object, Dictifiable ): + """ Associates a DatasetInstance (hda or ldda) with a DatasetCollection. """ + # actionable dataset id needs to be available via API... + dict_collection_visible_keys = ( 'id', 'element_type', 'element_index', 'element_identifier' ) + dict_element_visible_keys = ( 'id', 'element_type', 'element_index', 'element_identifier' ) + + def __init__( + self, + id=None, + collection=None, + element=None, + element_index=None, + element_identifier=None, + ): + if isinstance(element, HistoryDatasetAssociation): + self.hda = element + #self.instance_type = 'hda' + elif isinstance(element, LibraryDatasetDatasetAssociation): + self.ldda = element + #self.instance_type = 'ldda' + elif isinstance( element, DatasetCollection ): + self.child_collection = element + else: + raise AttributeError( 'Unknown element type provided: %s' % type( element ) ) + + self.id = id + self.collection = collection + self.element_index = element_index + self.element_identifier = element_identifier or str(element_index) + + @property + def element_type( self ): + if self.hda: + return "hda" + elif self.ldda: + return "ldda" + elif self.child_collection: + #TOOD: Rename element_type to element_type. + return "dataset_collection" + else: + raise Exception( "Unknown element instance type" ) + + @property + def is_collection( self ): + return self.element_type == "dataset_collection" + + @property + def element_object( self ): + if self.hda: + return self.hda + elif self.ldda: + return self.ldda + elif self.child_collection: + return self.child_collection + else: + raise Exception( "Unknown element instance type" ) + + @property + def dataset_instance( self ): + element_object = self.element_object + if isinstance( element_object, DatasetCollection ): + raise AttributeError( "Nested collection has no associated dataset_instance." ) + return element_object + + @property + def dataset( self ): + return self.dataset_instance.dataset + + def copy_to_collection( self, collection ): + new_element = DatasetCollectionElement( + element=self.element_object, + collection=collection, + element_index=self.element_index, + element_identifier=self.element_identifier, + ) + return new_element + + class Event( object ): def __init__( self, message=None, history=None, user=None, galaxy_session=None ): self.history = history @@ -3520,6 +3890,15 @@ class VisualizationTagAssociation ( ItemTagAssociation ): pass + +class HistoryDatasetCollectionTagAssociation( ItemTagAssociation ): + pass + + +class LibraryDatasetCollectionTagAssociation( ItemTagAssociation ): + pass + + class ToolTagAssociation( ItemTagAssociation ): def __init__( self, id=None, user=None, tool_id=None, tag_id=None, user_tname=None, value=None ): self.id = id @@ -3550,6 +3929,15 @@ class VisualizationAnnotationAssociation( object ): pass + +class HistoryDatasetCollectionAnnotationAssociation( object ): + pass + + +class LibraryDatasetCollectionAnnotationAssociation( object ): + pass + + # Item rating classes. class ItemRatingAssociation( object ): @@ -3583,6 +3971,17 @@ def set_item( self, visualization ): self.visualization = visualization + +class HistoryDatasetCollectionRatingAssociation( ItemRatingAssociation ): + def set_item( self, dataset_collection ): + self.dataset_collection = dataset_collection + + +class LibraryDatasetCollectionRatingAssociation( ItemRatingAssociation ): + def set_item( self, dataset_collection ): + self.dataset_collection = dataset_collection + + #Data Manager Classes class DataManagerHistoryAssociation( object ): def __init__( self, id=None, history=None, user=None ): This diff is so big that we needed to truncate the remainder. https://bitbucket.org/galaxy/galaxy-central/commits/45fd6f57a7a6/ Changeset: 45fd6f57a7a6 User: nsoranzo Date: 2014-05-07 11:32:54 Summary: Merged galaxy/galaxy-central into default Affected #: 10 files diff -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 -r 45fd6f57a7a66298e4c9b5dd468f73c71225209f lib/galaxy/webapps/galaxy/api/configuration.py --- a/lib/galaxy/webapps/galaxy/api/configuration.py +++ b/lib/galaxy/webapps/galaxy/api/configuration.py @@ -9,6 +9,7 @@ import logging log = logging.getLogger( __name__ ) + class ConfigurationController( BaseAPIController ): # config attributes viewable by non-admin users EXPOSED_USER_OPTIONS = [ @@ -20,6 +21,7 @@ 'logo_url', 'terms_url', 'allow_user_dataset_purge', + 'use_remote_user' ] # config attributes viewable by admin users EXPOSED_ADMIN_OPTIONS = [ diff -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 -r 45fd6f57a7a66298e4c9b5dd468f73c71225209f lib/galaxy/webapps/galaxy/api/workflows.py --- a/lib/galaxy/webapps/galaxy/api/workflows.py +++ b/lib/galaxy/webapps/galaxy/api/workflows.py @@ -168,8 +168,8 @@ from_history_id = payload.get( 'from_history_id' ) history = self.get_history( trans, from_history_id, check_ownership=False, check_accessible=True ) job_ids = map( trans.security.decode_id, payload.get( 'job_ids', [] ) ) - dataset_ids = map( trans.security.decode_id, payload.get( 'dataset_ids', [] ) ) - dataset_collection_ids = map( trans.security.decode_id, payload.get( 'dataset_collection_ids', [] ) ) + dataset_ids = payload.get( 'dataset_ids', [] ) + dataset_collection_ids = payload.get( 'dataset_collection_ids', [] ) workflow_name = payload[ 'workflow_name' ] stored_workflow = extract_workflow( trans=trans, diff -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 -r 45fd6f57a7a66298e4c9b5dd468f73c71225209f lib/galaxy/workflow/extract.py --- a/lib/galaxy/workflow/extract.py +++ b/lib/galaxy/workflow/extract.py @@ -87,7 +87,8 @@ for hid in dataset_collection_ids: step = model.WorkflowStep() step.type = 'data_collection_input' - step.tool_inputs = dict( name="Input Dataset Collection" ) + collection_type = summary.collection_types[ hid ] + step.tool_inputs = dict( name="Input Dataset Collection", collection_type=collection_type ) hid_to_output_pair[ hid ] = ( step, 'output' ) steps.append( step ) # Tool steps @@ -167,6 +168,8 @@ self.warnings = set() self.jobs = odict() self.implicit_map_jobs = [] + self.collection_types = {} + self.__summarize() def __summarize( self ): @@ -177,7 +180,9 @@ implicit_outputs = [] for content in self.history.active_contents: if content.history_content_type == "dataset_collection": + hid = content.hid content = self.__original_hdca( content ) + self.collection_types[ hid ] = content.collection.collection_type if not content.implicit_output_name: job = DatasetCollectionCreationJob( content ) self.jobs[ job ] = [ ( None, content ) ] diff -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 -r 45fd6f57a7a66298e4c9b5dd468f73c71225209f test/api/helpers.py --- a/test/api/helpers.py +++ b/test/api/helpers.py @@ -1,3 +1,5 @@ +from operator import itemgetter + import time import json import StringIO @@ -12,6 +14,35 @@ DEFAULT_HISTORY_TIMEOUT = 5 # Secs to wait on history to turn ok +def skip_without_tool( tool_id ): + """ Decorate an API test method as requiring a specific tool, + have nose skip the test case is the tool is unavailable. + """ + + def method_wrapper( method ): + + def get_tool_ids( api_test_case ): + index = api_test_case.galaxy_interactor.get( "tools", data=dict(in_panel=False) ) + tools = index.json() + # In panels by default, so flatten out sections... + tool_ids = map( itemgetter( "id" ), tools ) + return tool_ids + + def wrapped_method( api_test_case, *args, **kwargs ): + if tool_id not in get_tool_ids( api_test_case ): + from nose.plugins.skip import SkipTest + raise SkipTest( ) + + return method( api_test_case, *args, **kwargs ) + + # Must preserve method name so nose can detect and report tests by + # name. + wrapped_method.__name__ = method.__name__ + return wrapped_method + + return method_wrapper + + # Deprecated mixin, use dataset populator instead. # TODO: Rework existing tests to target DatasetPopulator in a setup method instead. class TestsDatasets: @@ -84,8 +115,8 @@ class WorkflowPopulator( object ): # Impulse is to make this a Mixin, but probably better as an object. - def __init__( self, api_test_case ): - self.api_test_case = api_test_case + def __init__( self, galaxy_interactor ): + self.galaxy_interactor = galaxy_interactor def load_workflow( self, name, content=workflow_str, add_pja=False ): workflow = json.loads( content ) @@ -111,7 +142,7 @@ workflow=json.dumps( workflow ), **create_kwds ) - upload_response = self.api_test_case._post( "workflows/upload", data=data ) + upload_response = self.galaxy_interactor.post( "workflows/upload", data=data ) uploaded_workflow_id = upload_response.json()[ "id" ] return uploaded_workflow_id @@ -189,6 +220,99 @@ return show().json() +class DatasetCollectionPopulator( object ): + + def __init__( self, galaxy_interactor ): + self.galaxy_interactor = galaxy_interactor + self.dataset_populator = DatasetPopulator( galaxy_interactor ) + + def create_list_from_pairs( self, history_id, pairs ): + element_identifiers = [] + for i, pair in enumerate( pairs ): + element_identifiers.append( dict( + name="test%d" % i, + src="hdca", + id=pair + ) ) + + payload = dict( + instance_type="history", + history_id=history_id, + element_identifiers=json.dumps(element_identifiers), + collection_type="list:paired", + ) + return self.__create( payload ) + + def create_pair_in_history( self, history_id, **kwds ): + payload = self.create_pair_payload( + history_id, + instance_type="history", + **kwds + ) + return self.__create( payload ) + + def create_list_in_history( self, history_id, **kwds ): + payload = self.create_list_payload( + history_id, + instance_type="history", + **kwds + ) + return self.__create( payload ) + + def create_list_payload( self, history_id, **kwds ): + return self.__create_payload( history_id, identifiers_func=self.list_identifiers, collection_type="list", **kwds ) + + def create_pair_payload( self, history_id, **kwds ): + return self.__create_payload( history_id, identifiers_func=self.pair_identifiers, collection_type="paired", **kwds ) + + def __create_payload( self, history_id, identifiers_func, collection_type, **kwds ): + contents = None + if "contents" in kwds: + contents = kwds[ "contents" ] + del kwds[ "contents" ] + + if "element_identifiers" not in kwds: + kwds[ "element_identifiers" ] = json.dumps( identifiers_func( history_id, contents=contents ) ) + + payload = dict( + history_id=history_id, + collection_type=collection_type, + **kwds + ) + return payload + + def pair_identifiers( self, history_id, contents=None ): + hda1, hda2 = self.__datasets( history_id, count=2, contents=contents ) + + element_identifiers = [ + dict( name="left", src="hda", id=hda1[ "id" ] ), + dict( name="right", src="hda", id=hda2[ "id" ] ), + ] + return element_identifiers + + def list_identifiers( self, history_id, contents=None ): + hda1, hda2, hda3 = self.__datasets( history_id, count=3, contents=contents ) + element_identifiers = [ + dict( name="data1", src="hda", id=hda1[ "id" ] ), + dict( name="data2", src="hda", id=hda2[ "id" ] ), + dict( name="data3", src="hda", id=hda3[ "id" ] ), + ] + return element_identifiers + + def __create( self, payload ): + create_response = self.galaxy_interactor.post( "dataset_collections", data=payload ) + return create_response + + def __datasets( self, history_id, count, contents=None ): + datasets = [] + for i in xrange( count ): + new_kwds = {} + if contents: + new_kwds[ "content" ] = contents[ i ] + datasets.append( self.dataset_populator.new_dataset( history_id, **new_kwds ) ) + return datasets + + def wait_on_state( state_func, assert_ok=False, timeout=5 ): delta = .1 iteration = 0 diff -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 -r 45fd6f57a7a66298e4c9b5dd468f73c71225209f test/api/test_dataset_collections.py --- /dev/null +++ b/test/api/test_dataset_collections.py @@ -0,0 +1,113 @@ +from base import api +import json +from .helpers import DatasetPopulator +from .helpers import DatasetCollectionPopulator + + +class DatasetCollectionApiTestCase( api.ApiTestCase ): + + def setUp( self ): + super( DatasetCollectionApiTestCase, self ).setUp() + self.dataset_populator = DatasetPopulator( self.galaxy_interactor ) + self.dataset_collection_populator = DatasetCollectionPopulator( self.galaxy_interactor ) + self.history_id = self.dataset_populator.new_history() + + def test_create_pair_from_history( self ): + payload = self.dataset_collection_populator.create_pair_payload( + self.history_id, + instance_type="history", + ) + create_response = self._post( "dataset_collections", payload ) + dataset_collection = self._check_create_response( create_response ) + returned_datasets = dataset_collection[ "elements" ] + assert len( returned_datasets ) == 2, dataset_collection + + def test_create_list_from_history( self ): + element_identifiers = self.dataset_collection_populator.list_identifiers( self.history_id ) + + payload = dict( + instance_type="history", + history_id=self.history_id, + element_identifiers=json.dumps(element_identifiers), + collection_type="list", + ) + + create_response = self._post( "dataset_collections", payload ) + dataset_collection = self._check_create_response( create_response ) + returned_datasets = dataset_collection[ "elements" ] + assert len( returned_datasets ) == 3, dataset_collection + + def test_create_list_of_existing_pairs( self ): + pair_payload = self.dataset_collection_populator.create_pair_payload( + self.history_id, + instance_type="history", + ) + pair_create_response = self._post( "dataset_collections", pair_payload ) + dataset_collection = self._check_create_response( pair_create_response ) + hdca_id = dataset_collection[ "id" ] + + element_identifiers = [ + dict( name="test1", src="hdca", id=hdca_id ) + ] + + payload = dict( + instance_type="history", + history_id=self.history_id, + element_identifiers=json.dumps(element_identifiers), + collection_type="list", + ) + create_response = self._post( "dataset_collections", payload ) + dataset_collection = self._check_create_response( create_response ) + returned_collections = dataset_collection[ "elements" ] + assert len( returned_collections ) == 1, dataset_collection + + def test_create_list_of_new_pairs( self ): + pair_identifiers = self.dataset_collection_populator.pair_identifiers( self.history_id ) + element_identifiers = [ dict( + src="new_collection", + name="test_pair", + collection_type="paired", + element_identifiers=pair_identifiers, + ) ] + payload = dict( + collection_type="list:paired", + instance_type="history", + history_id=self.history_id, + name="nested_collecion", + element_identifiers=json.dumps( element_identifiers ), + ) + create_response = self._post( "dataset_collections", payload ) + dataset_collection = self._check_create_response( create_response ) + assert dataset_collection[ "collection_type" ] == "list:paired" + returned_collections = dataset_collection[ "elements" ] + assert len( returned_collections ) == 1, dataset_collection + pair_1_element = returned_collections[ 0 ] + self._assert_has_keys( pair_1_element, "element_index" ) + pair_1_object = pair_1_element[ "object" ] + self._assert_has_keys( pair_1_object, "collection_type", "elements" ) + self.assertEquals( pair_1_object[ "collection_type" ], "paired" ) + pair_elements = pair_1_object[ "elements" ] + assert len( pair_elements ) == 2 + pair_1_element_1 = pair_elements[ 0 ] + assert pair_1_element_1[ "element_index" ] == 0 + + def test_hda_security( self ): + element_identifiers = self.dataset_collection_populator.pair_identifiers( self.history_id ) + + with self._different_user( ): + history_id = self.dataset_populator.new_history() + payload = dict( + instance_type="history", + history_id=history_id, + element_identifiers=json.dumps(element_identifiers), + collection_type="paired", + ) + + create_response = self._post( "dataset_collections", payload ) + self._assert_status_code_is( create_response, 403 ) + + def _check_create_response( self, create_response ): + self._assert_status_code_is( create_response, 200 ) + dataset_collection = create_response.json() + self._assert_has_keys( dataset_collection, "elements", "url", "name", "collection_type" ) + return dataset_collection diff -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 -r 45fd6f57a7a66298e4c9b5dd468f73c71225209f test/api/test_history_contents.py --- a/test/api/test_history_contents.py +++ b/test/api/test_history_contents.py @@ -3,7 +3,7 @@ from .helpers import TestsDatasets from .helpers import LibraryPopulator -from .test_dataset_collections import DatasetCollectionPopulator +from .helpers import DatasetCollectionPopulator from base.interactor import ( put_request, delete_request, diff -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 -r 45fd6f57a7a66298e4c9b5dd468f73c71225209f test/api/test_search.py --- a/test/api/test_search.py +++ b/test/api/test_search.py @@ -7,7 +7,7 @@ class SearchApiTestCase( api.ApiTestCase ): def test_search_workflows( self ): - workflow_populator = WorkflowPopulator( self ) + workflow_populator = WorkflowPopulator( self.galaxy_interactor ) workflow_id = workflow_populator.simple_workflow( "test_for_search" ) search_response = self.__search( "select * from workflow" ) assert self.__has_result_with_name( search_response, "test_for_search (imported from API)" ), search_response.json() diff -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 -r 45fd6f57a7a66298e4c9b5dd468f73c71225209f test/api/test_tools.py --- a/test/api/test_tools.py +++ b/test/api/test_tools.py @@ -2,7 +2,8 @@ from base import api from operator import itemgetter from .helpers import DatasetPopulator -from .test_dataset_collections import DatasetCollectionPopulator +from .helpers import DatasetCollectionPopulator +from .helpers import skip_without_tool class ToolsTestCase( api.ApiTestCase ): @@ -51,8 +52,8 @@ result_content = self._upload_and_get_content( table ) self.assertEquals( result_content, table ) + @skip_without_tool( "cat1" ) def test_run_cat1( self ): - self.__skip_without_tool( "cat1" ) # Run simple non-upload tool with an input data parameter. history_id = self.dataset_populator.new_history() new_dataset = self.dataset_populator.new_dataset( history_id, content='Cat1Test' ) @@ -66,8 +67,8 @@ output1_content = self._get_content( history_id, dataset=output1 ) self.assertEqual( output1_content.strip(), "Cat1Test" ) + @skip_without_tool( "cat1" ) def test_run_cat1_with_two_inputs( self ): - self.__skip_without_tool( "cat1" ) # Run tool with an multiple data parameter and grouping (repeat) history_id = self.dataset_populator.new_history() new_dataset1 = self.dataset_populator.new_dataset( history_id, content='Cat1Test' ) @@ -83,8 +84,8 @@ output1_content = self._get_content( history_id, dataset=output1 ) self.assertEqual( output1_content.strip(), "Cat1Test\nCat2Test" ) + @skip_without_tool( "cat1" ) def test_multirun_cat1( self ): - self.__skip_without_tool( "cat1" ) history_id = self.dataset_populator.new_history() new_dataset1 = self.dataset_populator.new_dataset( history_id, content='123' ) new_dataset2 = self.dataset_populator.new_dataset( history_id, content='456' ) @@ -104,8 +105,8 @@ self.assertEquals( output1_content.strip(), "123" ) self.assertEquals( output2_content.strip(), "456" ) + @skip_without_tool( "cat1" ) def test_multirun_in_repeat( self ): - self.__skip_without_tool( "cat1" ) history_id = self.dataset_populator.new_history() new_dataset1 = self.dataset_populator.new_dataset( history_id, content='123' ) new_dataset2 = self.dataset_populator.new_dataset( history_id, content='456' ) @@ -127,8 +128,8 @@ self.assertEquals( output1_content.strip(), "Common\n123" ) self.assertEquals( output2_content.strip(), "Common\n456" ) + @skip_without_tool( "cat1" ) def test_multirun_on_multiple_inputs( self ): - self.__skip_without_tool( "cat1" ) history_id = self.dataset_populator.new_history() new_dataset1 = self.dataset_populator.new_dataset( history_id, content='123' ) new_dataset2 = self.dataset_populator.new_dataset( history_id, content='456' ) @@ -153,8 +154,8 @@ assert "123\n0ab" in outputs_contents assert "456\n0ab" in outputs_contents + @skip_without_tool( "cat1" ) def test_map_over_collection( self ): - self.__skip_without_tool( "cat1" ) history_id = self.dataset_populator.new_history() hdca_id = self.__build_pair( history_id, [ "123", "456" ] ) inputs = { @@ -167,7 +168,6 @@ self.assertEquals( len( jobs ), 2 ) self.assertEquals( len( outputs ), 2 ) self.assertEquals( len( implicit_collections ), 1 ) - self.dataset_populator.wait_for_history( history_id, assert_ok=True ) output1 = outputs[ 0 ] output2 = outputs[ 1 ] @@ -176,8 +176,8 @@ self.assertEquals( output1_content.strip(), "123" ) self.assertEquals( output2_content.strip(), "456" ) + @skip_without_tool( "cat1" ) def test_map_over_nested_collections( self ): - self.__skip_without_tool( "cat1" ) history_id = self.dataset_populator.new_history() hdca_id = self.__build_nested_list( history_id ) inputs = { @@ -204,6 +204,7 @@ first_object_left_element = first_object[ "elements" ][ 0 ] self.assertEquals( outputs[ 0 ][ "id" ], first_object_left_element[ "object" ][ "id" ] ) + @skip_without_tool( "cat1" ) def test_map_over_two_collections( self ): history_id = self.dataset_populator.new_history() hdca1_id = self.__build_pair( history_id, [ "123", "456" ] ) @@ -222,8 +223,8 @@ self.assertEquals( output1_content.strip(), "123\n789" ) self.assertEquals( output2_content.strip(), "456\n0ab" ) + @skip_without_tool( "cat1" ) def test_cannot_map_over_incompatible_collections( self ): - self.__skip_without_tool( "cat1" ) history_id = self.dataset_populator.new_history() hdca1_id = self.__build_pair( history_id, [ "123", "456" ] ) hdca2_id = self.dataset_collection_populator.create_list_in_history( history_id ).json()[ "id" ] @@ -236,8 +237,8 @@ # on server. assert run_response.status_code >= 400 + @skip_without_tool( "multi_data_param" ) def test_reduce_collections( self ): - self.__skip_without_tool( "multi_data_param" ) history_id = self.dataset_populator.new_history() hdca1_id = self.__build_pair( history_id, [ "123", "456" ] ) hdca2_id = self.dataset_collection_populator.create_list_in_history( history_id ).json()[ "id" ] @@ -258,8 +259,8 @@ assert output1_content.strip() == "123\n456" assert len( output2_content.strip().split("\n") ) == 3, output2_content + @skip_without_tool( "collection_paired_test" ) def test_subcollection_mapping( self ): - self.__skip_without_tool( "collection_paired_test" ) history_id = self.dataset_populator.new_history() hdca_list_id = self.__build_nested_list( history_id ) inputs = { @@ -285,7 +286,8 @@ return self._run_outputs( self._run( tool_id, history_id, inputs ) ) def _run_outputs( self, create_response ): - self._assert_status_code_is( create_response, 200, assert_ok=True )[ 'outputs' ] + self._assert_status_code_is( create_response, 200 ) + return create_response.json()[ 'outputs' ] def _run_cat1( self, history_id, inputs, assert_ok=False ): return self._run( 'cat1', history_id, inputs, assert_ok=assert_ok ) @@ -334,11 +336,6 @@ tool_ids = map( itemgetter( "id" ), tools ) return tool_ids - def __skip_without_tool( self, tool_id ): - from nose.plugins.skip import SkipTest - if tool_id not in self.__tool_ids( ): - raise SkipTest( ) - def __build_nested_list( self, history_id ): hdca1_id = self.__build_pair( history_id, [ "123", "456" ] ) hdca2_id = self.__build_pair( history_id, [ "789", "0ab" ] ) diff -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 -r 45fd6f57a7a66298e4c9b5dd468f73c71225209f test/api/test_workflows.py --- a/test/api/test_workflows.py +++ b/test/api/test_workflows.py @@ -1,8 +1,11 @@ from base import api from json import dumps +from json import loads import time -from .helpers import TestsDatasets from .helpers import WorkflowPopulator +from .helpers import DatasetPopulator +from .helpers import DatasetCollectionPopulator +from .helpers import skip_without_tool from base.interactor import delete_request # requests like delete @@ -12,11 +15,13 @@ # - Allow post to workflows/<workflow_id>/run in addition to posting to # /workflows with id in payload. # - Much more testing obviously, always more testing. -class WorkflowsApiTestCase( api.ApiTestCase, TestsDatasets ): +class WorkflowsApiTestCase( api.ApiTestCase ): def setUp( self ): super( WorkflowsApiTestCase, self ).setUp() - self.workflow_populator = WorkflowPopulator( self ) + self.workflow_populator = WorkflowPopulator( self.galaxy_interactor ) + self.dataset_populator = DatasetPopulator( self.galaxy_interactor ) + self.dataset_collection_populator = DatasetCollectionPopulator( self.galaxy_interactor ) def test_delete( self ): workflow_id = self.workflow_populator.simple_workflow( "test_delete" ) @@ -58,6 +63,7 @@ first_input = downloaded_workflow[ "steps" ][ "0" ][ "inputs" ][ 0 ] assert first_input[ "name" ] == "WorkflowInput1" + @skip_without_tool( "cat1" ) def test_run_workflow( self ): workflow = self.workflow_populator.load_workflow( name="test_for_run" ) workflow_request, history_id = self._setup_workflow_run( workflow ) @@ -65,19 +71,20 @@ # something like that. run_workflow_response = self._post( "workflows", data=workflow_request ) self._assert_status_code_is( run_workflow_response, 200 ) - self._wait_for_history( history_id, assert_ok=True ) + self.dataset_populator.wait_for_history( history_id, assert_ok=True ) + @skip_without_tool( "cat1" ) def test_extract_from_history( self ): workflow = self.workflow_populator.load_workflow( name="test_for_extract" ) workflow_request, history_id = self._setup_workflow_run( workflow ) contents_response = self._get( "histories/%s/contents" % history_id ) self._assert_status_code_is( contents_response, 200 ) - hda_ids = map( lambda c: c[ "id" ], contents_response.json() ) + hda_ids = map( lambda c: c[ "hid" ], contents_response.json() ) run_workflow_response = self._post( "workflows", data=workflow_request ) self._assert_status_code_is( run_workflow_response, 200 ) - self._wait_for_history( history_id, assert_ok=True ) + self.dataset_populator.wait_for_history( history_id, assert_ok=True ) data = dict( history_id=history_id, tool_id="cat1" ) jobs_response = self._get( "jobs", data=data ) self._assert_status_code_is( jobs_response, 200 ) @@ -90,26 +97,66 @@ job_ids=dumps( [ cat1_job_id ] ), workflow_name="test import from history", ) - run_workflow_response = self._post( "workflows", data=create_from_data ) - self._assert_status_code_is( run_workflow_response, 200 ) + create_workflow_response = self._post( "workflows", data=create_from_data ) + self._assert_status_code_is( create_workflow_response, 200 ) - new_workflow_id = run_workflow_response.json()[ "id" ] + new_workflow_id = create_workflow_response.json()[ "id" ] download_response = self._get( "workflows/%s/download" % new_workflow_id ) self._assert_status_code_is( download_response, 200 ) downloaded_workflow = download_response.json() self.assertEquals( downloaded_workflow[ "name" ], "test import from history" ) assert len( downloaded_workflow[ "steps" ] ) == 3 + @skip_without_tool( "collection_paired_test" ) + def test_extract_workflows_with_dataset_collections( self ): + history_id = self.dataset_populator.new_history() + hdca = self.dataset_collection_populator.create_pair_in_history( history_id ).json() + hdca_id = hdca[ "id" ] + inputs = { + "f1": dict( src="hdca", id=hdca_id ) + } + payload = self.dataset_populator.run_tool_payload( + tool_id="collection_paired_test", + inputs=inputs, + history_id=history_id, + ) + tool_response = self._post( "tools", data=payload ) + self._assert_status_code_is( tool_response, 200 ) + job_id = tool_response.json()[ "jobs" ][ 0 ][ "id" ] + self.dataset_populator.wait_for_history( history_id, assert_ok=True ) + create_from_data = dict( + from_history_id=history_id, + dataset_collection_ids=dumps( [ hdca[ "hid" ] ] ), + job_ids=dumps( [ job_id ] ), + workflow_name="test import from history", + ) + create_workflow_response = self._post( "workflows", data=create_from_data ) + self._assert_status_code_is( create_workflow_response, 200 ) + create_workflow_response.json()[ "id" ] + + new_workflow_id = create_workflow_response.json()[ "id" ] + download_response = self._get( "workflows/%s/download" % new_workflow_id ) + self._assert_status_code_is( download_response, 200 ) + downloaded_workflow = download_response.json() + assert len( downloaded_workflow[ "steps" ] ) == 2 + collection_steps = [ s for s in downloaded_workflow[ "steps" ].values() if s[ "type" ] == "data_collection_input" ] + assert len( collection_steps ) == 1 + collection_step = collection_steps[ 0 ] + collection_step_state = loads( collection_step[ "tool_state" ] ) + self.assertEquals( collection_step_state[ "collection_type" ], u"paired" ) + + @skip_without_tool( "random_lines1" ) def test_run_replace_params_by_tool( self ): workflow_request, history_id = self._setup_random_x2_workflow( "test_for_replace_tool_params" ) workflow_request[ "parameters" ] = dumps( dict( random_lines1=dict( num_lines=5 ) ) ) run_workflow_response = self._post( "workflows", data=workflow_request ) self._assert_status_code_is( run_workflow_response, 200 ) - self._wait_for_history( history_id, assert_ok=True ) + self.dataset_populator.wait_for_history( history_id, assert_ok=True ) # Would be 8 and 6 without modification self.__assert_lines_hid_line_count_is( history_id, 2, 5 ) self.__assert_lines_hid_line_count_is( history_id, 3, 5 ) + @skip_without_tool( "random_lines1" ) def test_run_replace_params_by_steps( self ): workflow_request, history_id = self._setup_random_x2_workflow( "test_for_replace_step_params" ) workflow_summary_response = self._get( "workflows/%s" % workflow_request[ "workflow_id" ] ) @@ -120,7 +167,7 @@ workflow_request[ "parameters" ] = params run_workflow_response = self._post( "workflows", data=workflow_request ) self._assert_status_code_is( run_workflow_response, 200 ) - self._wait_for_history( history_id, assert_ok=True ) + self.dataset_populator.wait_for_history( history_id, assert_ok=True ) # Would be 8 and 6 without modification self.__assert_lines_hid_line_count_is( history_id, 2, 8 ) self.__assert_lines_hid_line_count_is( history_id, 3, 5 ) @@ -136,6 +183,7 @@ pja = pjas[ 0 ] self._assert_has_keys( pja, "action_type", "output_name", "action_arguments" ) + @skip_without_tool( "cat1" ) def test_invocation_usage( self ): workflow = self.workflow_populator.load_workflow( name="test_usage" ) workflow_request, history_id = self._setup_workflow_run( workflow ) @@ -159,6 +207,7 @@ for step in usage_details[ "steps" ].values(): self._assert_has_keys( step, "workflow_step_id", "order_index" ) + @skip_without_tool( "cat1" ) def test_post_job_action( self ): """ Tests both import and execution of post job actions. """ @@ -166,7 +215,7 @@ workflow_request, history_id = self._setup_workflow_run( workflow ) run_workflow_response = self._post( "workflows", data=workflow_request ) self._assert_status_code_is( run_workflow_response, 200 ) - self._wait_for_history( history_id, assert_ok=True ) + self.dataset_populator.wait_for_history( history_id, assert_ok=True ) time.sleep(.1) # Give another little bit of time for rename (needed?) contents = self._get( "histories/%s/contents" % history_id ).json() # loading workflow with add_pja=True causes workflow output to be @@ -183,9 +232,9 @@ step_1 = key if label == "WorkflowInput2": step_2 = key - history_id = self._new_history() - hda1 = self._new_dataset( history_id, content="1 2 3" ) - hda2 = self._new_dataset( history_id, content="4 5 6" ) + history_id = self.dataset_populator.new_history() + hda1 = self.dataset_populator.new_dataset( history_id, content="1 2 3" ) + hda2 = self.dataset_populator.new_dataset( history_id, content="4 5 6" ) workflow_request = dict( history="hist_id=%s" % history_id, workflow_id=uploaded_workflow_id, @@ -201,9 +250,9 @@ uploaded_workflow_id = self.workflow_populator.create_workflow( workflow ) workflow_inputs = self._workflow_inputs( uploaded_workflow_id ) key = workflow_inputs.keys()[ 0 ] - history_id = self._new_history() + history_id = self.dataset_populator.new_history() ten_lines = "\n".join( map( str, range( 10 ) ) ) - hda1 = self._new_dataset( history_id, content=ten_lines ) + hda1 = self.dataset_populator.new_dataset( history_id, content=ten_lines ) workflow_request = dict( history="hist_id=%s" % history_id, workflow_id=uploaded_workflow_id, diff -r ef7bd8c935bc2a01cf51dff8daa36e8ed7ac8ae6 -r 45fd6f57a7a66298e4c9b5dd468f73c71225209f test/functional/api/test_dataset_collections.py --- a/test/functional/api/test_dataset_collections.py +++ /dev/null @@ -1,206 +0,0 @@ -from base import api -import json -from .helpers import DatasetPopulator - - -# TODO: Move into helpers with rest of populators -class DatasetCollectionPopulator( object ): - - def __init__( self, galaxy_interactor ): - self.galaxy_interactor = galaxy_interactor - self.dataset_populator = DatasetPopulator( galaxy_interactor ) - - def create_list_from_pairs( self, history_id, pairs ): - element_identifiers = [] - for i, pair in enumerate( pairs ): - element_identifiers.append( dict( - name="test%d" % i, - src="hdca", - id=pair - ) ) - - payload = dict( - instance_type="history", - history_id=history_id, - element_identifiers=json.dumps(element_identifiers), - collection_type="list:paired", - ) - return self.__create( payload ) - - def create_pair_in_history( self, history_id, **kwds ): - payload = self.create_pair_payload( - history_id, - instance_type="history", - **kwds - ) - return self.__create( payload ) - - def create_list_in_history( self, history_id, **kwds ): - payload = self.create_list_payload( - history_id, - instance_type="history", - **kwds - ) - return self.__create( payload ) - - def create_list_payload( self, history_id, **kwds ): - return self.__create_payload( history_id, identifiers_func=self.list_identifiers, collection_type="list", **kwds ) - - def create_pair_payload( self, history_id, **kwds ): - return self.__create_payload( history_id, identifiers_func=self.pair_identifiers, collection_type="paired", **kwds ) - - def __create_payload( self, history_id, identifiers_func, collection_type, **kwds ): - contents = None - if "contents" in kwds: - contents = kwds[ "contents" ] - del kwds[ "contents" ] - - if "element_identifiers" not in kwds: - kwds[ "element_identifiers" ] = json.dumps( identifiers_func( history_id, contents=contents ) ) - - payload = dict( - history_id=history_id, - collection_type=collection_type, - **kwds - ) - return payload - - def pair_identifiers( self, history_id, contents=None ): - hda1, hda2 = self.__datasets( history_id, count=2, contents=contents ) - - element_identifiers = [ - dict( name="left", src="hda", id=hda1[ "id" ] ), - dict( name="right", src="hda", id=hda2[ "id" ] ), - ] - return element_identifiers - - def list_identifiers( self, history_id, contents=None ): - hda1, hda2, hda3 = self.__datasets( history_id, count=3, contents=contents ) - element_identifiers = [ - dict( name="data1", src="hda", id=hda1[ "id" ] ), - dict( name="data2", src="hda", id=hda2[ "id" ] ), - dict( name="data3", src="hda", id=hda3[ "id" ] ), - ] - return element_identifiers - - def __create( self, payload ): - create_response = self.galaxy_interactor.post( "dataset_collections", data=payload ) - return create_response - - def __datasets( self, history_id, count, contents=None ): - datasets = [] - for i in xrange( count ): - new_kwds = {} - if contents: - new_kwds[ "content" ] = contents[ i ] - datasets.append( self.dataset_populator.new_dataset( history_id, **new_kwds ) ) - return datasets - - -class DatasetCollectionApiTestCase( api.ApiTestCase ): - - def setUp( self ): - super( DatasetCollectionApiTestCase, self ).setUp() - self.dataset_populator = DatasetPopulator( self.galaxy_interactor ) - self.dataset_collection_populator = DatasetCollectionPopulator( self.galaxy_interactor ) - self.history_id = self.dataset_populator.new_history() - - def test_create_pair_from_history( self ): - payload = self.dataset_collection_populator.create_pair_payload( - self.history_id, - instance_type="history", - ) - create_response = self._post( "dataset_collections", payload ) - dataset_collection = self._check_create_response( create_response ) - returned_datasets = dataset_collection[ "elements" ] - assert len( returned_datasets ) == 2, dataset_collection - - def test_create_list_from_history( self ): - element_identifiers = self.dataset_collection_populator.list_identifiers( self.history_id ) - - payload = dict( - instance_type="history", - history_id=self.history_id, - element_identifiers=json.dumps(element_identifiers), - collection_type="list", - ) - - create_response = self._post( "dataset_collections", payload ) - dataset_collection = self._check_create_response( create_response ) - returned_datasets = dataset_collection[ "elements" ] - assert len( returned_datasets ) == 3, dataset_collection - - def test_create_list_of_existing_pairs( self ): - pair_payload = self.dataset_collection_populator.create_pair_payload( - self.history_id, - instance_type="history", - ) - pair_create_response = self._post( "dataset_collections", pair_payload ) - dataset_collection = self._check_create_response( pair_create_response ) - hdca_id = dataset_collection[ "id" ] - - element_identifiers = [ - dict( name="test1", src="hdca", id=hdca_id ) - ] - - payload = dict( - instance_type="history", - history_id=self.history_id, - element_identifiers=json.dumps(element_identifiers), - collection_type="list", - ) - create_response = self._post( "dataset_collections", payload ) - dataset_collection = self._check_create_response( create_response ) - returned_collections = dataset_collection[ "elements" ] - assert len( returned_collections ) == 1, dataset_collection - - def test_create_list_of_new_pairs( self ): - pair_identifiers = self.dataset_collection_populator.pair_identifiers( self.history_id ) - element_identifiers = [ dict( - src="new_collection", - name="test_pair", - collection_type="paired", - element_identifiers=pair_identifiers, - ) ] - payload = dict( - collection_type="list:paired", - instance_type="history", - history_id=self.history_id, - name="nested_collecion", - element_identifiers=json.dumps( element_identifiers ), - ) - create_response = self._post( "dataset_collections", payload ) - dataset_collection = self._check_create_response( create_response ) - assert dataset_collection[ "collection_type" ] == "list:paired" - returned_collections = dataset_collection[ "elements" ] - assert len( returned_collections ) == 1, dataset_collection - pair_1_element = returned_collections[ 0 ] - self._assert_has_keys( pair_1_element, "element_index" ) - pair_1_object = pair_1_element[ "object" ] - self._assert_has_keys( pair_1_object, "collection_type", "elements" ) - self.assertEquals( pair_1_object[ "collection_type" ], "paired" ) - pair_elements = pair_1_object[ "elements" ] - assert len( pair_elements ) == 2 - pair_1_element_1 = pair_elements[ 0 ] - assert pair_1_element_1[ "element_index" ] == 0 - - def test_hda_security( self ): - element_identifiers = self.dataset_collection_populator.pair_identifiers( self.history_id ) - - with self._different_user( ): - history_id = self.dataset_populator.new_history() - payload = dict( - instance_type="history", - history_id=history_id, - element_identifiers=json.dumps(element_identifiers), - collection_type="paired", - ) - - create_response = self._post( "dataset_collections", payload ) - self._assert_status_code_is( create_response, 403 ) - - def _check_create_response( self, create_response ): - self._assert_status_code_is( create_response, 200 ) - dataset_collection = create_response.json() - self._assert_has_keys( dataset_collection, "elements", "url", "name", "collection_type" ) - return dataset_collection https://bitbucket.org/galaxy/galaxy-central/commits/f58a4a9185e1/ Changeset: f58a4a9185e1 User: nsoranzo Date: 2014-05-07 11:39:06 Summary: Workflow API: Add documentation for dataset_collection_ids param. Affected #: 1 file diff -r 45fd6f57a7a66298e4c9b5dd468f73c71225209f -r f58a4a9185e133a987d032e4563c043477790a71 lib/galaxy/webapps/galaxy/api/workflows.py --- a/lib/galaxy/webapps/galaxy/api/workflows.py +++ b/lib/galaxy/webapps/galaxy/api/workflows.py @@ -143,13 +143,16 @@ :param from_history_id: Id of history to extract a workflow from. Either workflow_id, installed_repository_file or from_history_id must be specified :type from_history_id: str - :param job_ids: If from_history_id is set - optional list of jobs to include when extracting workflow from history. + :param job_ids: If from_history_id is set - optional list of jobs to include when extracting a workflow from history :type job_ids: str - :param dataset_ids: If from_history_id is set - optional list of HDA ids corresponding to workflow inputs when extracting workflow from history. + :param dataset_ids: If from_history_id is set - optional list of HDA ids corresponding to workflow inputs when extracting a workflow from history :type dataset_ids: str - :param workflow_name: If from_history_id is set - name of the workflow to create + :param dataset_collection_ids: If from_history_id is set - optional list of HDCA ids corresponding to workflow inputs when extracting a workflow from history + :type dataset_collection_ids: str + + :param workflow_name: If from_history_id is set - name of the workflow to create when extracting a workflow from history :type workflow_name: str """ https://bitbucket.org/galaxy/galaxy-central/commits/83531e44b244/ Changeset: 83531e44b244 User: jmchilton Date: 2014-05-07 18:43:03 Summary: Merge pull request #383. Affected #: 1 file diff -r 9e502242af28473660ee70ee0f98345f3604e137 -r 83531e44b2445dad9a4cf2c262e210c12498d9bd lib/galaxy/webapps/galaxy/api/workflows.py --- a/lib/galaxy/webapps/galaxy/api/workflows.py +++ b/lib/galaxy/webapps/galaxy/api/workflows.py @@ -1,9 +1,9 @@ -from __future__ import absolute_import - """ API operations for Workflows """ +from __future__ import absolute_import + import logging from sqlalchemy import desc, or_ from galaxy import exceptions @@ -12,7 +12,6 @@ from galaxy.web import _future_expose_api as expose_api from galaxy.web.base.controller import BaseAPIController, url_for, UsesStoredWorkflowMixin from galaxy.web.base.controller import UsesHistoryMixin -from galaxy.workflow.modules import module_factory from galaxy.workflow.run import invoke from galaxy.workflow.run import WorkflowRunConfig from galaxy.workflow.extract import extract_workflow @@ -120,79 +119,85 @@ workflow will be created for this user. Otherwise, workflow_id must be specified and this API method will cause a workflow to execute. - :param installed_repository_file The path of a workflow to import. Either workflow_id or installed_repository_file must be specified + :param installed_repository_file The path of a workflow to import. Either workflow_id, installed_repository_file or from_history_id must be specified :type installed_repository_file str - :param workflow_id: an existing workflow id. Either workflow_id or installed_repository_file must be specified + :param workflow_id: An existing workflow id. Either workflow_id, installed_repository_file or from_history_id must be specified :type workflow_id: str - :param parameters: See _update_step_parameters() + :param parameters: If workflow_id is set - see _update_step_parameters() :type parameters: dict - :param ds_map: A dictionary mapping each input step id to a dictionary with 2 keys: 'src' (which can be 'ldda', 'ld' or 'hda') and 'id' (which should be the id of a LibraryDatasetDatasetAssociation, LibraryDataset or HistoryDatasetAssociation respectively) + :param ds_map: If workflow_id is set - a dictionary mapping each input step id to a dictionary with 2 keys: 'src' (which can be 'ldda', 'ld' or 'hda') and 'id' (which should be the id of a LibraryDatasetDatasetAssociation, LibraryDataset or HistoryDatasetAssociation respectively) :type ds_map: dict - :param no_add_to_history: if present in the payload with any value, the input datasets will not be added to the selected history + :param no_add_to_history: If workflow_id is set - if present in the payload with any value, the input datasets will not be added to the selected history :type no_add_to_history: str - :param history: Either the name of a new history or "hist_id=HIST_ID" where HIST_ID is the id of an existing history + :param history: If workflow_id is set - optional history where to run the workflow, either the name of a new history or "hist_id=HIST_ID" where HIST_ID is the id of an existing history. If not specified, the workflow will be run a new unnamed history :type history: str - :param replacement_params: A dictionary used when renaming datasets + :param replacement_params: If workflow_id is set - an optional dictionary used when renaming datasets :type replacement_params: dict - :param from_history_id: Id of history to extract a workflow from. Should not be used with worfklow_id or installed_repository_file. + :param from_history_id: Id of history to extract a workflow from. Either workflow_id, installed_repository_file or from_history_id must be specified :type from_history_id: str - :param job_ids: If from_history_id is set - this should be a list of jobs to include when extracting workflow from history. + :param job_ids: If from_history_id is set - optional list of jobs to include when extracting a workflow from history :type job_ids: str - :param dataset_ids: If from_history_id is set - this should be a list of HDA ids corresponding to workflow inputs when extracting workflow from history. + :param dataset_ids: If from_history_id is set - optional list of HDA ids corresponding to workflow inputs when extracting a workflow from history :type dataset_ids: str + + :param dataset_collection_ids: If from_history_id is set - optional list of HDCA ids corresponding to workflow inputs when extracting a workflow from history + :type dataset_collection_ids: str + + :param workflow_name: If from_history_id is set - name of the workflow to create when extracting a workflow from history + :type workflow_name: str """ - # Pull parameters out of payload. - workflow_id = payload.get('workflow_id', None) - param_map = payload.get('parameters', {}) - ds_map = payload.get('ds_map', {}) + if len( set( ['workflow_id', 'installed_repository_file', 'from_history_id'] ).intersection( payload ) ) > 1: + trans.response.status = 403 + return "Only one among 'workflow_id', 'installed_repository_file', 'from_history_id' must be specified" + + if 'installed_repository_file' in payload: + workflow_controller = trans.webapp.controllers[ 'workflow' ] + result = workflow_controller.import_workflow( trans=trans, + cntrller='api', + **payload) + return result + + if 'from_history_id' in payload: + from_history_id = payload.get( 'from_history_id' ) + history = self.get_history( trans, from_history_id, check_ownership=False, check_accessible=True ) + job_ids = map( trans.security.decode_id, payload.get( 'job_ids', [] ) ) + dataset_ids = payload.get( 'dataset_ids', [] ) + dataset_collection_ids = payload.get( 'dataset_collection_ids', [] ) + workflow_name = payload[ 'workflow_name' ] + stored_workflow = extract_workflow( + trans=trans, + user=trans.get_user(), + history=history, + job_ids=job_ids, + dataset_ids=dataset_ids, + dataset_collection_ids=dataset_collection_ids, + workflow_name=workflow_name, + ) + item = stored_workflow.to_dict( value_mapper={ 'id': trans.security.encode_id } ) + item[ 'url' ] = url_for( 'workflow', id=item[ 'id' ] ) + return item + + workflow_id = payload.get( 'workflow_id', None ) + if not workflow_id: + trans.response.status = 403 + return "Either workflow_id, installed_repository_file or from_history_id must be specified" + + # Pull other parameters out of payload. + param_map = payload.get( 'parameters', {} ) + ds_map = payload.get( 'ds_map', {} ) add_to_history = 'no_add_to_history' not in payload history_param = payload.get('history', '') - # Get/create workflow. - if not workflow_id: - # create new - if 'installed_repository_file' in payload: - workflow_controller = trans.webapp.controllers[ 'workflow' ] - result = workflow_controller.import_workflow( trans=trans, - cntrller='api', - **payload) - return result - if 'from_history_id' in payload: - from_history_id = payload.get( 'from_history_id' ) - history = self.get_history( trans, from_history_id, check_ownership=False, check_accessible=True ) - job_ids = map( trans.security.decode_id, payload.get( "job_ids", [] ) ) - dataset_ids = payload.get( "dataset_ids", [] ) - dataset_collection_ids = payload.get( "dataset_collection_ids", [] ) - workflow_name = payload[ "workflow_name" ] - stored_workflow = extract_workflow( - trans=trans, - user=trans.get_user(), - history=history, - job_ids=job_ids, - dataset_ids=dataset_ids, - dataset_collection_ids=dataset_collection_ids, - workflow_name=workflow_name, - ) - item = stored_workflow.to_dict( value_mapper={ "id": trans.security.encode_id } ) - item[ 'url' ] = url_for( 'workflow', id=item[ "id" ] ) - return item - - trans.response.status = 403 - return "Either workflow_id or installed_repository_file must be specified" - if 'installed_repository_file' in payload: - trans.response.status = 403 - return "installed_repository_file may not be specified with workflow_id" - # Get workflow + accessibility check. stored_workflow = trans.sa_session.query(self.app.model.StoredWorkflow).get( trans.security.decode_id(workflow_id)) https://bitbucket.org/galaxy/galaxy-central/commits/a014c2d841a8/ Changeset: a014c2d841a8 User: jmchilton Date: 2014-05-07 18:49:59 Summary: Re-update documentation for fixes in a44bea8. Affected #: 1 file diff -r 83531e44b2445dad9a4cf2c262e210c12498d9bd -r a014c2d841a8490f017086c687997a191f3a602f lib/galaxy/webapps/galaxy/api/workflows.py --- a/lib/galaxy/webapps/galaxy/api/workflows.py +++ b/lib/galaxy/webapps/galaxy/api/workflows.py @@ -146,10 +146,10 @@ :param job_ids: If from_history_id is set - optional list of jobs to include when extracting a workflow from history :type job_ids: str - :param dataset_ids: If from_history_id is set - optional list of HDA ids corresponding to workflow inputs when extracting a workflow from history + :param dataset_ids: If from_history_id is set - optional list of HDA `hid`s corresponding to workflow inputs when extracting a workflow from history :type dataset_ids: str - :param dataset_collection_ids: If from_history_id is set - optional list of HDCA ids corresponding to workflow inputs when extracting a workflow from history + :param dataset_collection_ids: If from_history_id is set - optional list of HDCA `hid`s corresponding to workflow inputs when extracting a workflow from history :type dataset_collection_ids: str :param workflow_name: If from_history_id is set - name of the workflow to create when extracting a workflow from history Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: jmchilton: Simpler, more powerful subcollection mapping.
by commits-noreply＠bitbucket.org 07 May '14

07 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/9e502242af28/ Changeset: 9e502242af28 User: jmchilton Date: 2014-05-07 18:29:51 Summary: Simpler, more powerful subcollection mapping. Collections can be mapped over 'data' parameters and sufficiently nested collections can be mapped over 'data_collection' parameters (for instance a list of 5 pairs can be supplied to a tool taking in a pair and 5 jobs will be executed). I (perhaps poorly) term these concepts collection mapping and subcollection mapping. Prior to this changeset - the API for doing collection 'mapping' and 'subcollection mapping' was somewhat more divergent and the tool execution code explicitly forbid doing both kinds of mappings in the same tool execution even if the effective collections could be matched (e.g. it could not map a 'data' parameter over a list of 5 datasets and a pair parameter a list of 5 pairs in the same execution). This changeset should remedy this - as long as the effective collection mappings can match up such jobs should be possible. The workflow editor (and I think runner) already thought this was possible, so this changeset reduces the tool-workflow impedance mismatch - an existing problem exacerbated by recent dataset collections introduction. This all needs much more testing - test workflows execute this way, functional test of a tool execution that combines collection mapping and subcollection mapping, etc.... Affected #: 4 files diff -r c5b736d3e4e5027391a04fd187e2f9d2aa4ebe46 -r 9e502242af28473660ee70ee0f98345f3604e137 lib/galaxy/tools/__init__.py --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -1925,7 +1925,7 @@ # Fixed set of input parameters may correspond to any number of jobs. # Expand these out to individual parameters for given jobs (tool # executions). - expanded_incomings, collection_info = expand_meta_parameters( trans, incoming, self.inputs ) + expanded_incomings, collection_info = expand_meta_parameters( trans, self, incoming ) if not expanded_incomings: raise exceptions.MessageException( "Tool execution failed, trying to run a tool over an empty collection." ) diff -r c5b736d3e4e5027391a04fd187e2f9d2aa4ebe46 -r 9e502242af28473660ee70ee0f98345f3604e137 lib/galaxy/tools/parameters/basic.py --- a/lib/galaxy/tools/parameters/basic.py +++ b/lib/galaxy/tools/parameters/basic.py @@ -2021,7 +2021,7 @@ self._ensure_selection( field ) return field - def _get_select_dataset_collection_field( self, trans, history, multiple=False, suffix="|__subcollection_multirun__", value=None, other_values=None ): + def _get_select_dataset_collection_field( self, trans, history, multiple=False, suffix="|__collection_multirun__", value=None, other_values=None ): field_name = "%s%s" % ( self.name, suffix ) field = form_builder.SelectField( field_name, multiple, None, self.refresh_on_change, refresh_on_change_values=self.refresh_on_change_values ) dataset_matcher = DatasetMatcher( trans, self, value, other_values ) diff -r c5b736d3e4e5027391a04fd187e2f9d2aa4ebe46 -r 9e502242af28473660ee70ee0f98345f3604e137 lib/galaxy/tools/parameters/meta.py --- a/lib/galaxy/tools/parameters/meta.py +++ b/lib/galaxy/tools/parameters/meta.py @@ -9,7 +9,7 @@ log = logging.getLogger( __name__ ) -def expand_meta_parameters( trans, incoming, inputs ): +def expand_meta_parameters( trans, tool, incoming ): """ Take in a dictionary of raw incoming parameters and expand to a list of expanded incoming parameters (one set of parameters per tool @@ -34,26 +34,23 @@ def collection_classifier( input_key ): multirun_key = "%s|__collection_multirun__" % input_key if multirun_key in incoming: - encoded_hdc_id = incoming[ multirun_key ] - hdc_id = trans.app.security.decode_id( encoded_hdc_id ) - hdc = trans.sa_session.query( model.HistoryDatasetCollectionAssociation ).get( hdc_id ) - collections_to_match.add( input_key, hdc ) - hdas = hdc.collection.dataset_instances - return permutations.input_classification.MATCHED, hdas - else: - return permutations.input_classification.SINGLE, incoming[ input_key ] - - def subcollection_classifier( input_key ): - multirun_key = "%s|__subcollection_multirun__" % input_key - if multirun_key in incoming: incoming_val = incoming[ multirun_key ] - # value will be "hdca_id|subcollection_type" - encoded_hdc_id, subcollection_type = incoming_val.split( "|", 1 ) + # If subcollectin multirun of data_collection param - value will + # be "hdca_id|subcollection_type" else it will just be hdca_id + if "|" in incoming_val: + encoded_hdc_id, subcollection_type = incoming_val.split( "|", 1 ) + else: + encoded_hdc_id = incoming_val + subcollection_type = None hdc_id = trans.app.security.decode_id( encoded_hdc_id ) hdc = trans.sa_session.query( model.HistoryDatasetCollectionAssociation ).get( hdc_id ) collections_to_match.add( input_key, hdc, subcollection_type=subcollection_type ) - subcollection_elements = subcollections.split_dataset_collection_instance( hdc, subcollection_type ) - return permutations.input_classification.MATCHED, subcollection_elements + if subcollection_type is not None: + subcollection_elements = subcollections.split_dataset_collection_instance( hdc, subcollection_type ) + return permutations.input_classification.MATCHED, subcollection_elements + else: + hdas = hdc.collection.dataset_instances + return permutations.input_classification.MATCHED, hdas else: return permutations.input_classification.SINGLE, incoming[ input_key ] @@ -72,26 +69,17 @@ multirun_found = False collection_multirun_found = False - subcollection_multirun_found = False for key, value in incoming.iteritems(): multirun_found = try_replace_key( key, "|__multirun__" ) or multirun_found collection_multirun_found = try_replace_key( key, "|__collection_multirun__" ) or collection_multirun_found - subcollection_multirun_found = try_replace_key( key, "|__subcollection_multirun__" ) or subcollection_multirun_found - if sum( [ 1 if f else 0 for f in [ multirun_found, collection_multirun_found, subcollection_multirun_found ] ] ) > 1: + if sum( [ 1 if f else 0 for f in [ multirun_found, collection_multirun_found ] ] ) > 1: # In theory doable, but to complicated for a first pass. message = "Cannot specify parallel execution across both multiple datasets and dataset collections." raise exceptions.ToolMetaParameterException( message ) if multirun_found: return permutations.expand_multi_inputs( incoming_template, classifier ), None - elif subcollection_multirun_found: - expanded_incomings = permutations.expand_multi_inputs( incoming_template, subcollection_classifier ) - if collections_to_match.has_collections(): - collection_info = trans.app.dataset_collections_service.match_collections( collections_to_match ) - else: - collection_info = None - return expanded_incomings, collection_info else: expanded_incomings = permutations.expand_multi_inputs( incoming_template, collection_classifier ) if collections_to_match.has_collections(): diff -r c5b736d3e4e5027391a04fd187e2f9d2aa4ebe46 -r 9e502242af28473660ee70ee0f98345f3604e137 test/api/test_tools.py --- a/test/api/test_tools.py +++ b/test/api/test_tools.py @@ -264,7 +264,7 @@ history_id = self.dataset_populator.new_history() hdca_list_id = self.__build_nested_list( history_id ) inputs = { - "f1|__subcollection_multirun__": "%s|paired" % hdca_list_id + "f1|__collection_multirun__": "%s|paired" % hdca_list_id } # Following wait not really needed - just getting so many database # locked errors with sqlite. Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: 2 new changesets
by commits-noreply＠bitbucket.org 07 May '14

07 May '14

2 new commits in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/fff767c478fb/ Changeset: fff767c478fb User: dannon Date: 2014-05-07 17:48:11 Summary: Merge stable. Affected #: 1 file https://bitbucket.org/galaxy/galaxy-central/commits/c5b736d3e4e5/ Changeset: c5b736d3e4e5 User: dannon Date: 2014-05-07 17:49:14 Summary: Merge. Affected #: 2 files diff -r fff767c478fbce151720d49729d5444f4f85cd28 -r c5b736d3e4e5027391a04fd187e2f9d2aa4ebe46 lib/galaxy/exceptions/__init__.py --- a/lib/galaxy/exceptions/__init__.py +++ b/lib/galaxy/exceptions/__init__.py @@ -104,16 +104,16 @@ class Conflict( MessageException ): status_code = 409 - err_code = error_codes.CONFLICT + err_code = error_codes.CONFLICT class InconsistentDatabase ( MessageException ): status_code = 500 - err_code = error_codes.INCONSISTENT_DATABASE + err_code = error_codes.INCONSISTENT_DATABASE class InternalServerError ( MessageException ): status_code = 500 - err_code = error_codes.INTERNAL_SERVER_ERROR + err_code = error_codes.INTERNAL_SERVER_ERROR class NotImplemented ( MessageException ): status_code = 501 - err_code = error_codes.NOT_IMPLEMENTED + err_code = error_codes.NOT_IMPLEMENTED diff -r fff767c478fbce151720d49729d5444f4f85cd28 -r c5b736d3e4e5027391a04fd187e2f9d2aa4ebe46 lib/galaxy/webapps/galaxy/buildapp.py --- a/lib/galaxy/webapps/galaxy/buildapp.py +++ b/lib/galaxy/webapps/galaxy/buildapp.py @@ -192,7 +192,6 @@ webapp.mapper.connect( 'workflow_usage', '/api/workflows/{workflow_id}/usage', controller='workflows', action='workflow_usage', conditions=dict(method=['GET'])) webapp.mapper.connect( 'workflow_usage_contents', '/api/workflows/{workflow_id}/usage/{usage_id}', controller='workflows', action='workflow_usage_contents', conditions=dict(method=['GET'])) - # ============================ # ===== AUTHENTICATE API ===== # ============================ @@ -248,8 +247,8 @@ parent_resources=dict( member_name='library', collection_name='libraries' ) ) _add_item_extended_metadata_controller( webapp, - name_prefix="library_dataset_", - path_prefix='/api/libraries/:library_id/contents/:library_content_id' ) + name_prefix="library_dataset_", + path_prefix='/api/libraries/:library_id/contents/:library_content_id' ) # ======================= # ===== FOLDERS API ===== Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: martenson: pep8 fixes
by commits-noreply＠bitbucket.org 07 May '14

07 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/e8336f4c485b/ Changeset: e8336f4c485b User: martenson Date: 2014-05-07 17:47:28 Summary: pep8 fixes Affected #: 2 files diff -r a9e4f0ea892c8b7da3f1e468195dc1d29914fd38 -r e8336f4c485b0eedd75b9185f78ff5acaca7a8b3 lib/galaxy/exceptions/__init__.py --- a/lib/galaxy/exceptions/__init__.py +++ b/lib/galaxy/exceptions/__init__.py @@ -104,16 +104,16 @@ class Conflict( MessageException ): status_code = 409 - err_code = error_codes.CONFLICT + err_code = error_codes.CONFLICT class InconsistentDatabase ( MessageException ): status_code = 500 - err_code = error_codes.INCONSISTENT_DATABASE + err_code = error_codes.INCONSISTENT_DATABASE class InternalServerError ( MessageException ): status_code = 500 - err_code = error_codes.INTERNAL_SERVER_ERROR + err_code = error_codes.INTERNAL_SERVER_ERROR class NotImplemented ( MessageException ): status_code = 501 - err_code = error_codes.NOT_IMPLEMENTED + err_code = error_codes.NOT_IMPLEMENTED diff -r a9e4f0ea892c8b7da3f1e468195dc1d29914fd38 -r e8336f4c485b0eedd75b9185f78ff5acaca7a8b3 lib/galaxy/webapps/galaxy/buildapp.py --- a/lib/galaxy/webapps/galaxy/buildapp.py +++ b/lib/galaxy/webapps/galaxy/buildapp.py @@ -192,7 +192,6 @@ webapp.mapper.connect( 'workflow_usage', '/api/workflows/{workflow_id}/usage', controller='workflows', action='workflow_usage', conditions=dict(method=['GET'])) webapp.mapper.connect( 'workflow_usage_contents', '/api/workflows/{workflow_id}/usage/{usage_id}', controller='workflows', action='workflow_usage_contents', conditions=dict(method=['GET'])) - # ============================ # ===== AUTHENTICATE API ===== # ============================ @@ -248,8 +247,8 @@ parent_resources=dict( member_name='library', collection_name='libraries' ) ) _add_item_extended_metadata_controller( webapp, - name_prefix="library_dataset_", - path_prefix='/api/libraries/:library_id/contents/:library_content_id' ) + name_prefix="library_dataset_", + path_prefix='/api/libraries/:library_id/contents/:library_content_id' ) # ======================= # ===== FOLDERS API ===== Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: dannon: Force an absolute path for the temp directory in the local runner
by commits-noreply＠bitbucket.org 07 May '14

07 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/3534f67acd2a/ Changeset: 3534f67acd2a Branch: stable User: dannon Date: 2014-05-07 15:45:53 Summary: Force an absolute path for the temp directory in the local runner Affected #: 1 file diff -r 61b4a993d39c5341afea857859221a4f5b94334f -r 3534f67acd2a4f84899335326ac2cc8bcf05201d lib/galaxy/jobs/runners/local.py --- a/lib/galaxy/jobs/runners/local.py +++ b/lib/galaxy/jobs/runners/local.py @@ -39,7 +39,7 @@ #Set TEMP if a valid temp value is not already set if not ( 'TMPDIR' in self._environ or 'TEMP' in self._environ or 'TMP' in self._environ ): - self._environ[ 'TEMP' ] = tempfile.gettempdir() + self._environ[ 'TEMP' ] = os.path.abspath(tempfile.gettempdir()) super( LocalJobRunner, self ).__init__( app, nworkers ) self._init_worker_threads() Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: dannon: Force an absolute path for the temp directory in the local runner
by commits-noreply＠bitbucket.org 07 May '14

07 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/a9e4f0ea892c/ Changeset: a9e4f0ea892c User: dannon Date: 2014-05-07 15:45:53 Summary: Force an absolute path for the temp directory in the local runner Affected #: 1 file diff -r ac95389c7eb502345d0321c54b750f134c5e6ba9 -r a9e4f0ea892c8b7da3f1e468195dc1d29914fd38 lib/galaxy/jobs/runners/local.py --- a/lib/galaxy/jobs/runners/local.py +++ b/lib/galaxy/jobs/runners/local.py @@ -39,7 +39,7 @@ #Set TEMP if a valid temp value is not already set if not ( 'TMPDIR' in self._environ or 'TEMP' in self._environ or 'TMP' in self._environ ): - self._environ[ 'TEMP' ] = tempfile.gettempdir() + self._environ[ 'TEMP' ] = os.path.abspath(tempfile.gettempdir()) super( LocalJobRunner, self ).__init__( app, nworkers ) self._init_worker_threads() Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0