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 logging
import os
import tempfile
from galaxy.tools.actions import ToolAction
from galaxy.tools.imp_exp import JobExportHistoryArchiveWrapper
from galaxy.util.odict import odict
log = logging.getLogger(__name__)
[docs]class ImportHistoryToolAction(ToolAction):
"""Tool action used for importing a history to an archive. """
[docs] def execute(self, tool, trans, incoming={}, set_output_hid=False, overwrite=True, history=None, **kwargs):
#
# Create job.
#
job = trans.app.model.Job()
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)
trans.sa_session.flush() # ensure job.id are available
#
# 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)
#
# 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
job.set_handler(tool.get_job_handler(None))
trans.sa_session.flush()
# Queue the job for execution
trans.app.job_manager.job_queue.put(job.id, tool.id)
trans.log_event("Added import history job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id)
return job, odict()
[docs]class ExportHistoryToolAction(ToolAction):
"""Tool action used for exporting a history to an archive. """
[docs] def execute(self, tool, trans, incoming={}, set_output_hid=False, overwrite=True, history=None, **kwargs):
#
# Get history to export.
#
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()
session = trans.get_galaxy_session()
job.session_id = session and session.id
if history:
history_id = history.id
else:
history_id = trans.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
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)
# Create dataset that will serve as archive.
archive_dataset = trans.app.model.Dataset()
trans.sa_session.add(archive_dataset)
trans.sa_session.flush() # ensure job.id and archive_dataset.id are available
trans.app.object_store.create(archive_dataset) # set the object store id, create dataset (if applicable)
#
# Setup job and job wrapper.
#
# Add association for keeping track of job, history, archive relationship.
jeha = trans.app.model.JobExportHistoryArchive(job=job, history=history,
dataset=archive_dataset,
compressed=incoming['compress'])
trans.sa_session.add(jeha)
job_wrapper = JobExportHistoryArchiveWrapper(job)
cmd_line = job_wrapper.setup_job(trans, jeha, include_hidden=incoming['include_hidden'],
include_deleted=incoming['include_deleted'])
#
# Add parameters to job_parameter table.
#
# Set additional parameters.
incoming['__HISTORY_TO_EXPORT__'] = history.id
incoming['__EXPORT_HISTORY_COMMAND_INPUTS_OPTIONS__'] = cmd_line
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
job.set_handler(tool.get_job_handler(None))
trans.sa_session.flush()
# Queue the job for execution
trans.app.job_manager.job_queue.put(job.id, tool.id)
trans.log_event("Added export history job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id)
return job, odict()