Warning
This document is for an in-development version 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.job_execution.ports.view
import logging
from galaxy import (
exceptions,
model,
util
)
log = logging.getLogger(__name__)
[docs]class JobPortsView:
[docs] def register_container_information(self, job_id, **kwd):
job = self.__authorize_job_access(job_id, **kwd)
container_runtime = kwd.get("container_runtime")
log.info(kwd)
self._app.interactivetool_manager.configure_entry_points(job, container_runtime)
return {"message": "ok"}
# Copy/paste from JobFilesView - TODO: de-duplicate.
def __authorize_job_access(self, encoded_job_id, **kwargs):
key = "job_key"
if key not in kwargs:
error_message = "Job files action requires a valid '%s'." % key
raise exceptions.ObjectAttributeMissingException(error_message)
job_id = self._security.decode_id(encoded_job_id)
job_key = self._security.encode_id(job_id, kind="jobs_files")
if not util.safe_str_cmp(kwargs["job_key"], job_key):
raise exceptions.ItemAccessibilityException("Invalid job_key supplied.")
# Verify job is active. Don't update the contents of complete jobs.
sa_session = self._app.model.context.current
job = sa_session.query(model.Job).get(job_id)
if not job.running:
error_message = "Attempting to read or modify the files of a job that has already completed."
raise exceptions.ItemAccessibilityException(error_message)
return job
@property
def _security(self):
return self._app.security