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.webapps.galaxy.api.tool_entry_points

""" API for asynchronous job running mechanisms can use to fetch or put files
related to running and queued jobs.
import logging

from galaxy import (
from galaxy.managers.context import ProvidesUserContext
from galaxy.model import (
from galaxy.structured_app import StructuredApp
from galaxy.web import expose_api_anonymous_and_sessionless
from . import BaseGalaxyAPIController

log = logging.getLogger(__name__)

[docs]class ToolEntryPointsAPIController(BaseGalaxyAPIController):
[docs] def __init__(self, app: StructuredApp): self.app = app self.interactivetool_manager = app.interactivetool_manager
[docs] @expose_api_anonymous_and_sessionless def index(self, trans: ProvidesUserContext, running=False, job_id=None, **kwd): """ * GET /api/entry_points Returns tool entry point information. Currently passing a job_id parameter is required, as this becomes more general that won't be needed. :type job_id: string :param job_id: Encoded job id :type running: boolean :param running: filter to only include running job entry points. :rtype: list :returns: list of entry point dictionaries. """ running = util.asbool(running) if job_id is None and not running: raise exceptions.RequestParameterInvalidException("Currently this API must passed a job id or running=true") if job_id is not None and running: raise exceptions.RequestParameterInvalidException( "Currently this API must passed only a job id or running=true" ) if job_id is not None: job = trans.sa_session.query(Job).get(self.decode_id(job_id)) if not self.interactivetool_manager.can_access_job(trans, job): raise exceptions.ItemAccessibilityException() entry_points = job.interactivetool_entry_points if running: entry_points = self.interactivetool_manager.get_nonterminal_for_user_by_trans(trans) rval = [] for entry_point in entry_points: as_dict = self.encode_all_ids(trans, entry_point.to_dict(), True) target = self.interactivetool_manager.target_if_active(trans, entry_point) if target: as_dict["target"] = target rval.append(as_dict) return rval
[docs] @expose_api_anonymous_and_sessionless def access_entry_point(self, trans: ProvidesUserContext, id, **kwd): """ * GET /api/entry_points/{id}/access Return the URL target described by the entry point. :type id: string :param id: Encoded entry point id :rtype: dictionary :returns: dictionary containing target for interactivetool entry point """ # Because of auto id encoding needed for link from grid, the item.id keyword must be 'id' if not id: raise exceptions.RequestParameterMissingException("Must supply entry point ID.") entry_point_id = self.decode_id(id) return {"target": self.interactivetool_manager.access_entry_point_target(trans, entry_point_id)}
[docs] @expose_api_anonymous_and_sessionless def stop_entry_point(self, trans: ProvidesUserContext, id, **kwds): """ DELETE /api/entry_points/{id} """ if not id: raise exceptions.RequestParameterMissingException("Must supply entry point id") try: entry_point_id = self.decode_id(id) entry_point = trans.sa_session.query(InteractiveToolEntryPoint).get(entry_point_id) except Exception: raise exceptions.RequestParameterInvalidException("entry point '{id}' invalid") if self.app.interactivetool_manager.can_access_entry_point(trans, entry_point): self.app.interactivetool_manager.stop(trans, entry_point) else: raise exceptions.ItemAccessibilityException(f"entry point '{id}' is not accessible")