Update LWR client through LWR changeset 51f097d.

Add RemoteTransferAction - to allow LWR server to stage files up and down by downloading/uploading them. Still need to augment Galaxy to support such an endpoint for job-related files.

lib/galaxy/jobs/runners/lwr_client/action_mapper.py

@@ -1,4 +1,6 @@
 from json import load
+from os import makedirs
+from os.path import exists
 from os.path import abspath
 from os.path import dirname
 from os.path import join
@@ -11,6 +13,9 @@ import galaxy.util
 from galaxy.util.bunch import Bunch
 from .util import directory_files
 from .util import unique_path_prefix
+from .transport import get_file
+from .transport import post_file
+
 
 DEFAULT_MAPPED_ACTION = 'transfer'  # Not really clear to me what this should be, exception?
 DEFAULT_PATH_MAPPER_TYPE = 'prefix'
@@ -67,7 +72,7 @@ class FileActionMapper(object):
     ...     f = NamedTemporaryFile(delete=False)
     ...     f.write(config_contents.encode('UTF-8'))
     ...     f.close()
-    ...     mock_client = Bunch(default_file_action=default_action, action_config_path=f.name)
+    ...     mock_client = Bunch(default_file_action=default_action, action_config_path=f.name, files_endpoint=None)
     ...     mapper = FileActionMapper(mock_client)
     ...     mapper = FileActionMapper(config=mapper.to_dict()) # Serialize and deserialize it to make sure still works
     ...     unlink(f.name)
@@ -125,10 +130,12 @@ class FileActionMapper(object):
             config = self.__client_to_config(client)
         self.default_action = config.get("default_action", "transfer")
         self.mappers = mappers_from_dicts(config.get("paths", []))
+        self.files_endpoint = config.get("files_endpoint", None)
 
     def to_dict(self):
         return dict(
             default_action=self.default_action,
+            files_endpoint=self.files_endpoint,
             paths=map(lambda m: m.to_dict(), self.mappers)
         )
 
@@ -139,6 +146,7 @@ class FileActionMapper(object):
         else:
             config = dict()
         config["default_action"] = client.default_file_action
+        config["files_endpoint"] = client.files_endpoint
         return config
 
     def __load_action_config(self, path):
@@ -166,7 +174,9 @@ class FileActionMapper(object):
             message_template = "Unknown action_type encountered %s while trying to map path %s"
             message_args = (action_type, path)
             raise Exception(message_template % message_args)
-        return action_class(path, file_lister=file_lister)
+        action = action_class(path, file_lister=file_lister)
+        self.__process_action(action, type)
+        return action
 
     def unstructured_mappers(self):
         """ Return mappers that will map 'unstructured' files (i.e. go beyond
@@ -174,6 +184,20 @@ class FileActionMapper(object):
         """
         return filter(lambda m: path_type.UNSTRUCTURED in m.path_types, self.mappers)
 
+    def __process_action(self, action, file_type):
+        """ Extension point to populate extra action information after an
+        action has been created.
+        """
+        if action.action_type == "remote_transfer":
+            url_base = self.files_endpoint
+            if not url_base:
+                raise Exception("Attempted to use remote_transfer action with defining a files_endpoint")
+            if "?" not in url_base:
+                url_base = "%s?" % url_base
+            # TODO: URL encode path.
+            url = "%s&path=%s&file_type=%s" % (url_base, action.path, file_type)
+            action.url = url
+
 
 class BaseAction(object):
 
@@ -233,7 +257,7 @@ class RemoteCopyAction(BaseAction):
     staging = STAGING_ACTION_REMOTE
 
     def to_dict(self):
-        return dict(path=self.path, action_type=RemoteCopyAction.action_type)
+        return dict(path=self.path, action_type=self.action_type)
 
     @classmethod
     def from_dict(cls, action_dict):
@@ -242,6 +266,41 @@ class RemoteCopyAction(BaseAction):
     def write_to_path(self, path):
         galaxy.util.copy_to_path(open(self.path, "rb"), path)
 
+    def write_from_path(self, lwr_path):
+        destination = self.path
+        parent_directory = dirname(destination)
+        if not exists(parent_directory):
+            makedirs(parent_directory)
+        with open(lwr_path, "rb") as f:
+            galaxy.util.copy_to_path(f, destination)
+
+
+class RemoteTransferAction(BaseAction):
+    """ This action indicates the LWR server should copy the file before
+    execution via direct file system copy. This is like a CopyAction, but
+    it indicates the action should occur on the LWR server instead of on
+    the client.
+    """
+    action_type = "remote_transfer"
+    staging = STAGING_ACTION_REMOTE
+
+    def __init__(self, path, file_lister=None, url=None):
+        super(RemoteTransferAction, self).__init__(path, file_lister=file_lister)
+        self.url = url
+
+    def to_dict(self):
+        return dict(path=self.path, action_type=self.action_type, url=self.url)
+
+    @classmethod
+    def from_dict(cls, action_dict):
+        return RemoteTransferAction(path=action_dict["path"], url=action_dict["url"])
+
+    def write_to_path(self, path):
+        get_file(self.url, path)
+
+    def write_from_path(self, lwr_path):
+        post_file(self.url, lwr_path)
+
 
 class MessageAction(object):
     """ Sort of pseudo action describing "files" store in memory and
@@ -275,7 +334,8 @@ class MessageAction(object):
     def write_to_path(self, path):
         open(path, "w").write(self.contents)
 
-DICTIFIABLE_ACTION_CLASSES = [RemoteCopyAction, MessageAction]
+
+DICTIFIABLE_ACTION_CLASSES = [RemoteCopyAction, RemoteTransferAction, MessageAction]
 
 
 def from_dict(action_dict):
@@ -402,8 +462,20 @@ class FileLister(object):
 
 DEFAULT_FILE_LISTER = FileLister(dict(depth=0))
 
-ACTION_CLASSES = [NoneAction, TransferAction, CopyAction, RemoteCopyAction]
+ACTION_CLASSES = [
+    NoneAction,
+    TransferAction,
+    CopyAction,
+    RemoteCopyAction,
+    RemoteTransferAction
+]
 actions = dict([(clazz.action_type, clazz) for clazz in ACTION_CLASSES])
 
 
-__all__ = [FileActionMapper, path_type, from_dict, MessageAction]
+__all__ = [
+    FileActionMapper,
+    path_type,
+    from_dict,
+    MessageAction,
+    RemoteTransferAction,  # For testing
+]

lib/galaxy/jobs/runners/lwr_client/client.py

@@ -39,6 +39,7 @@ class BaseJobClient(object):
             )
         else:
             job_directory = None
+        self.files_endpoint = destination_params.get("files_endpoint", None)
         self.job_directory = job_directory
 
         self.default_file_action = self.destination_params.get("default_file_action", "transfer")

lib/galaxy/jobs/runners/lwr_client/manager.py

@@ -59,8 +59,9 @@ class ClientManager(object):
             self.client_class = JobClient
             self.extra_client_kwds = {}
 
-    def get_client(self, destination_params, job_id):
+    def get_client(self, destination_params, job_id, **kwargs):
         destination_params = _parse_destination_params(destination_params)
+        destination_params.update(**kwargs)
         job_manager_interface_class = self.job_manager_interface_class
         job_manager_interface_args = dict(destination_params=destination_params, **self.job_manager_interface_args)
         job_manager_interface = job_manager_interface_class(**job_manager_interface_args)
@@ -112,8 +113,9 @@ class MessageQueueClientManager(object):
     def __nonzero__(self):
         return self.active
 
-    def get_client(self, destination_params, job_id):
+    def get_client(self, destination_params, job_id, **kwargs):
         destination_params = _parse_destination_params(destination_params)
+        destination_params.update(**kwargs)
         return MessageJobClient(destination_params, job_id, self)
 
 

lib/galaxy/jobs/runners/lwr_client/staging/up.py

@@ -381,7 +381,7 @@ class TransferTracker(object):
 
     def register_rewrite(self, local_path, remote_path, type, force=False):
         action = self.__action(local_path, type)
-        if action.action_type in ['transfer', 'copy'] or force:
+        if action.staging_needed or force:
             self.file_renames[local_path] = remote_path
 
     def rewrite_input_paths(self):

lib/galaxy/jobs/runners/lwr_client/transport/__init__.py

@@ -23,4 +23,9 @@ def __get_transport_type(transport_type, os_module):
             transport_type = 'curl'
     return transport_type
 
-__all__ = [get_transport]
+# TODO: Provide urllib implementation if these unavailable,
+# also explore a requests+poster option.
+from .curl import get_file
+from .curl import post_file
+
+__all__ = [get_transport, get_file, post_file]

lib/galaxy/jobs/runners/lwr_client/transport/curl.py

@@ -4,7 +4,7 @@ except ImportError:
     from io import StringIO
 try:
     from pycurl import Curl
-except:
+except ImportError:
     pass
 from os.path import getsize
 
@@ -16,9 +16,9 @@ PYCURL_UNAVAILABLE_MESSAGE = \
 class PycurlTransport(object):
 
     def execute(self, url, data=None, input_path=None, output_path=None):
-        buf = self._open_output(output_path)
+        buf = _open_output(output_path)
         try:
-            c = self._new_curl_object()
+            c = _new_curl_object()
             c.setopt(c.URL, url.encode('ascii'))
             c.setopt(c.WRITEFUNCTION, buf.write)
             if input_path:
@@ -37,11 +37,33 @@ class PycurlTransport(object):
         finally:
             buf.close()
 
-    def _new_curl_object(self):
-        try:
-            return Curl()
-        except NameError:
-            raise ImportError(PYCURL_UNAVAILABLE_MESSAGE)
 
-    def _open_output(self, output_path):
-        return open(output_path, 'wb') if output_path else StringIO()
+def post_file(url, path):
+    c = _new_curl_object()
+    c.setopt(c.URL, url.encode('ascii'))
+    c.setopt(c.HTTPPOST, [("file", (c.FORM_FILE, path.encode('ascii')))])
+    c.perform()
+
+
+def get_file(url, path):
+    buf = _open_output(path)
+    try:
+        c = _new_curl_object()
+        c.setopt(c.URL, url.encode('ascii'))
+        c.setopt(c.WRITEFUNCTION, buf.write)
+        c.perform()
+    finally:
+        buf.close()
+
+
+def _open_output(output_path):
+    return open(output_path, 'wb') if output_path else StringIO()
+
+
+def _new_curl_object():
+    try:
+        return Curl()
+    except NameError:
+        raise ImportError(PYCURL_UNAVAILABLE_MESSAGE)
+
+___all__ = [PycurlTransport, post_file, get_file]