Source code for girder_worker.docker.transforms.girder

import errno
import os
import shutil
from girder_worker_utils.transform import Transform
from girder_worker_utils.transforms.girder_io import (
    GirderUploadJobArtifact,
    GirderClientTransform,
    GirderUploadToFolder,
    GirderUploadToItem
)
from . import TemporaryVolume, Connect, NamedOutputPipe, _maybe_transform


[docs]class ProgressPipe(Transform): """ This can be used to stream progress information out of a running docker container as part of a docker_run task. For a usage example, see :ref:`docker-progress`. :param name: The filename, which will be a named pipe open for reading from the host. :type name: str :param volume: The bind mount volume where the underlying named pipe will reside. :type volume: :py:class:`girder_worker.docker.transforms.BindMountVolume` """ def __init__(self, name='.girder_progress', volume=TemporaryVolume.default): self.name = name self._volume = volume def transform(self, task=None, **kwargs): from girder_worker.docker.stream_adapter import JobProgressAdapter self._volume.transform(**kwargs) # TODO What do we do is job_manager is None? When can it me None? job_mgr = task.job_manager # For now we are using JobProgressAdapter which is part of core, we should # probably add a copy to the docker package to break to reference to core. return Connect(NamedOutputPipe(self.name, volume=self._volume), JobProgressAdapter(job_mgr)).transform(**kwargs)
[docs]class GirderFileIdToStream(GirderClientTransform): """ This can be used to stream a Girder file into a docker container. See :ref:`docker-run-streaming-input` for example usage. :param _id: The Girder file ID. :type _id: str or ObjectId """ def __init__(self, _id, **kwargs): super(GirderFileIdToStream, self).__init__(**kwargs) self.file_id = _id def transform(self, **kwargs): from girder_worker.docker.io.girder import ( GirderFileStreamReader ) return GirderFileStreamReader(self.gc, self.file_id)
[docs]class GirderFileIdToVolume(GirderClientTransform): """ This can be used to pass a Girder file into a docker container. It downloads the file to a bind mounted volume, and returns the container path of the file. :param _id: The Girder file ID. :type _id: str or ObjectId :param volume: The bind mount volume where the file will reside. :type volume: :py:class:`girder_worker.docker.transforms.BindMountVolume` :param filename: Alternate name for the file. Default is to use the name from Girder. :type filename: str """ def __init__(self, _id, volume=TemporaryVolume.default, filename=None, **kwargs): super(GirderFileIdToVolume, self).__init__(**kwargs) self._file_id = str(_id) self._volume = volume self._filename = filename self._file_path = None def _create_file_path(self, root): if self._filename is None: # If no filename is explicitly passed, we read the filename from Girder # and put it in its own directory named by its UUID. filename = self.gc.getFile(self._file_id)['name'] path = os.path.join(root, self._file_id) try: os.mkdir(path) except OSError as e: if e.errno != errno.EEXIST: raise return os.path.join(self._file_id, filename), os.path.join(path, filename) else: return self._filename, os.path.join(root, self._filename) def transform(self, **kwargs): self._volume.transform(**kwargs) dir = self._volume.host_path rel_path, self._file_path = self._create_file_path(dir) self.gc.downloadFile(self._file_id, self._file_path) # Return the path inside the container return os.path.join(self._volume.container_path, rel_path) def cleanup(self, **kwargs): if self._file_path is not None: shutil.rmtree(self._file_path, ignore_errors=True) def _repr_model_(self): if self._filename: template = u'<{module}.{cls}: File ID={id} -> "{fname}">' else: template = u'<{module}.{cls}: File ID={id}>' return template.format( module=self.__module__, cls=self.__class__.__name__, id=self._file_id, fname=self._filename)
[docs]class GirderFolderIdToVolume(GirderClientTransform): """ This can be used to pass a Girder folder into a docker container. It downloads the folder to a bind mounted volume, and returns the container path of the directory. :param _id: The Girder folder ID. :type _id: str or ObjectId :param volume: The bind mount volume where the directory will reside. :type volume: :py:class:`girder_worker.docker.transforms.BindMountVolume` :param folder_name: Alternate name for the directory. Default is to use the name from Girder. :type folder_name: str """ def __init__(self, _id, volume=TemporaryVolume.default, folder_name=None, **kwargs): super(GirderFolderIdToVolume, self).__init__(**kwargs) self._folder_id = str(_id) self._volume = volume self._folder_name = folder_name self._folder_path = None def _create_folder_path(self, root): if self._folder_name is None: # If no folder name is explicitly passed, we read the filename from Girder # and put it in its own directory named by its UUID. self._folder_name = self.gc.getFolder(self._folder_id)['name'] path = os.path.join(root, self._folder_id, self._folder_name) try: os.makedirs(path) except OSError as e: if e.errno != errno.EEXIST: raise return os.path.join(self._folder_id, self._folder_name), path def transform(self, **kwargs): self._volume.transform(**kwargs) dir = self._volume.host_path rel_path, self._folder_path = self._create_folder_path(dir) self.gc.downloadFolderRecursive(self._folder_id, self._folder_path) # Return the path inside the container return os.path.join(self._volume.container_path, rel_path) def cleanup(self, **kwargs): if self._folder_path is not None: shutil.rmtree(self._folder_path, ignore_errors=True) def _repr_model_(self): return '<%s.%s: Folder ID=%s -> "%s">' % ( self.__module__, self.__class__.__name__, self._folder_id, self._folder_path)
[docs]class GirderUploadVolumePathToItem(GirderUploadToItem): """ This transform uploads data in a bind mount volume to a Girder item. This should be used in ``girder_result_hooks`` to upload files produced by the task. :param volumepath: The location of the file to upload. :type volumepath: :py:class:`girder_worker.docker.transforms.VolumePath` :param item_id: The item ID in Girder. :type item_id: str or ObjectId :param delete_file: Whether to delete the file afterward. :type delete_file: bool """ def __init__(self, volumepath, item_id, delete_file=False, **kwargs): item_id = str(item_id) super(GirderUploadVolumePathToItem, self).__init__(item_id, delete_file, **kwargs) self._volumepath = volumepath # We ignore the "result" def transform(self, *args, **kwargs): path = _maybe_transform(self._volumepath, *args, **kwargs) return super(GirderUploadVolumePathToItem, self).transform(path)
[docs]class GirderUploadVolumePathToFolder(GirderUploadToFolder): """ This transform uploads data in a bind mount volume to a Girder folder. This should be used in ``girder_result_hooks`` to upload data produced by the task. :param volumepath: The location of the file or directory to upload. :type volumepath: :py:class:`girder_worker.docker.transforms.VolumePath` :param folder_id: The folder ID in Girder. :type folder_id: str or ObjectId :param delete_file: Whether to delete the data afterward. :type delete_file: bool """ def __init__(self, volumepath, folder_id, delete_file=False, **kwargs): super(GirderUploadVolumePathToFolder, self).__init__(str(folder_id), delete_file, **kwargs) self._volumepath = volumepath def transform(self, *args, **kwargs): path = _maybe_transform(self._volumepath, *args, **kwargs) return super(GirderUploadVolumePathToFolder, self).transform(path)
[docs]class GirderUploadVolumePathJobArtifact(GirderUploadJobArtifact): """ This transform can be used to upload artifacts produced during a docker task execution and attach them to the corresponding job in Girder. This can be useful for tracing and debugging jobs, or simply collecting intermediate information during job execution. If the passed in path does not exist, this is a no-op. :param volumepath: A volume path pointing to a mounted directory or file. If a directory, all files within the directory will be uploaded as artifacts to the job. If a file, just uploads the single file. If it does not exist, no action is performed. :type volumepath: :py:class:`girder_worker.docker.transforms.VolumePath` :param job_id: The job ID to attach the artifacts to. If calling this from Girder via ``docker_run.delay``, you will not need to set this, as it will be set automatically. :type job_id: str :param name: A name for the artifact. Only applies for single file paths. If not specified, will use the basename of the file. :type name: str :param upload_on_exception: If True, this transform will occur even if the docker task fails. This can be used to debug failed ``docker_run`` tasks. :type upload_on_exception: bool """ def __init__(self, volumepath, job_id=None, name=None, upload_on_exception=False, **kwargs): if job_id is not None: job_id = str(job_id) super(GirderUploadVolumePathJobArtifact, self).__init__(job_id, name, **kwargs) self._volumepath = volumepath self._upload_on_exception = upload_on_exception def transform(self, *args, **kwargs): path = _maybe_transform(self._volumepath, *args, **kwargs) return super(GirderUploadVolumePathJobArtifact, self).transform(path) def exception(self): if self._upload_on_exception: return self.transform()