Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.tools.actions.history_imp_exp

import datetime
import logging
import os
import tempfile

from galaxy.job_execution.setup import create_working_directory_for_job
from galaxy.model.base import transaction
from galaxy.tools.actions import ToolAction
from galaxy.tools.imp_exp import (
    JobExportHistoryArchiveWrapper,
    JobImportHistoryArchiveWrapper,
)
from galaxy.util import ready_name_for_url

log = logging.getLogger(__name__)


[docs]class ImportHistoryToolAction(ToolAction): """Tool action used for importing a history to an archive.""" produces_real_jobs = True
[docs] def execute(self, tool, trans, incoming=None, set_output_hid=False, overwrite=True, history=None, **kwargs): # # Create job. # incoming = incoming or {} trans.check_user_activation() job = trans.app.model.Job() job.galaxy_version = trans.app.config.version_major session = trans.get_galaxy_session() job.session_id = session and session.id if history: history_id = history.id elif trans.history: history_id = trans.history.id else: history_id = None job.history_id = history_id job.tool_id = tool.id job.user_id = trans.user.id start_job_state = job.state # should be job.states.NEW job.state = ( job.states.WAITING ) # we need to set job state to something other than NEW, or else when tracking jobs in db it will be picked up before we have added input / output parameters trans.sa_session.add(job) with transaction(trans.sa_session): # ensure job.id are available trans.sa_session.commit() # # Setup job and job wrapper. # # Add association for keeping track of job, history relationship. # Use abspath because mkdtemp() does not, contrary to the documentation, # always return an absolute path. archive_dir = os.path.abspath(tempfile.mkdtemp()) jiha = trans.app.model.JobImportHistoryArchive(job=job, archive_dir=archive_dir) trans.sa_session.add(jiha) job_wrapper = JobImportHistoryArchiveWrapper(trans.app, job) job_wrapper.setup_job(jiha, incoming["__ARCHIVE_SOURCE__"], incoming["__ARCHIVE_TYPE__"]) # # Add parameters to job_parameter table. # # Set additional parameters. incoming["__DEST_DIR__"] = jiha.archive_dir for name, value in tool.params_to_strings(incoming, trans.app).items(): job.add_parameter(name, value) job.state = start_job_state # job inputs have been configured, restore initial job state return job, {}
[docs]class ExportHistoryToolAction(ToolAction): """Tool action used for exporting a history to an archive.""" produces_real_jobs = True
[docs] def execute(self, tool, trans, incoming=None, set_output_hid=False, overwrite=True, history=None, **kwargs): trans.check_user_activation() # # Get history to export. # incoming = incoming or {} history = None for name, value in incoming.items(): if isinstance(value, trans.app.model.History): history_param_name = name history = value del incoming[history_param_name] break if not history: raise Exception("There is no history to export.") # # Create the job and output dataset objects # job = trans.app.model.Job() job.galaxy_version = trans.app.config.version_major session = trans.get_galaxy_session() job.session_id = session and session.id history_id = history.id job.history_id = history_id job.tool_id = tool.id if trans.user: # If this is an actual user, run the job as that individual. Otherwise we're running as guest. job.user_id = trans.user.id job.state = ( job.states.WAITING ) # we need to set job state to something other than NEW, or else when tracking jobs in db it will be picked up before we have added input / output parameters trans.sa_session.add(job) compressed = incoming["compress"] exporting_to_uri = "directory_uri" in incoming if not exporting_to_uri: # see comment below about how this should be transitioned to occuring in a # job handler or detached MQ-driven thread jeha = trans.app.model.JobExportHistoryArchive.create_for_history( history, job, trans.sa_session, trans.app.object_store, compressed ) store_directory = jeha.temp_directory else: # creating a job directory in the web thread is bad (it is slow, bypasses # dynamic objectstore assignment, etc..) but it is arguably less bad than # creating a dataset (like above for dataset export case). # ensure job.id is available with transaction(trans.sa_session): trans.sa_session.commit() job_directory = create_working_directory_for_job(trans.app.object_store, job) store_directory = os.path.join(job_directory, "working", "_object_export") os.makedirs(store_directory) # # Setup job and job wrapper. # cmd_line = f"--galaxy-version '{job.galaxy_version}'" if compressed: cmd_line += " -G" cmd_line = f"{cmd_line} {store_directory}" # # Add parameters to job_parameter table. # # Set additional parameters. incoming["__HISTORY_TO_EXPORT__"] = history.id incoming["__EXPORT_HISTORY_COMMAND_INPUTS_OPTIONS__"] = cmd_line if exporting_to_uri: directory_uri = incoming["directory_uri"] file_name = incoming.get("file_name") if file_name is None: hname = ready_name_for_url(history.name) human_timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S") if compressed: extension = ".tar.gz" else: extension = ".tar" file_name = f"Galaxy-History-{hname}-{human_timestamp}.{extension}" file_name = os.path.basename(os.path.abspath(file_name)) sep = "" if directory_uri.endswith("/") else "/" incoming["__EXPORT_TO_URI__"] = f"{directory_uri}{sep}{file_name}" for name, value in tool.params_to_strings(incoming, trans.app).items(): job.add_parameter(name, value) with transaction(trans.sa_session): trans.sa_session.commit() job_wrapper = JobExportHistoryArchiveWrapper(trans.app, job.id) job_wrapper.setup_job( history, store_directory, include_hidden=incoming["include_hidden"], include_deleted=incoming["include_deleted"], compressed=compressed, user=trans.user, ) return job, {}