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

"""
API operations provenance
"""
import logging

from paste.httpexceptions import (
    HTTPBadRequest,
    HTTPNotImplemented
)

from galaxy import (
    managers,
    web
)
from galaxy.web.base.controller import BaseAPIController

log = logging.getLogger(__name__)


[docs]class BaseProvenanceController(BaseAPIController): """ """
[docs] def __init__(self, app): super(BaseProvenanceController, self).__init__(app) self.hda_manager = managers.hdas.HDAManager(app)
[docs] @web.legacy_expose_api def index(self, trans, **kwd): follow = kwd.get('follow', False) value = self._get_provenance(trans, self.provenance_item_class, kwd[self.provenance_item_id], follow) return value
[docs] @web.legacy_expose_api def show(self, trans, elem_name, **kwd): follow = kwd.get('follow', False) value = self._get_provenance(trans, self.provenance_item_class, kwd[self.provenance_item_id], follow) return value
[docs] @web.legacy_expose_api def create(self, trans, tag_name, payload=None, **kwd): payload = payload or {} raise HTTPNotImplemented()
[docs] @web.legacy_expose_api def delete(self, trans, tag_name, **kwd): raise HTTPBadRequest("Cannot Delete Provenance")
def _get_provenance(self, trans, item_class_name, item_id, follow=True): provenance_item = self.get_object(trans, item_id, item_class_name, check_ownership=False, check_accessible=False) if item_class_name == "HistoryDatasetAssociation": self.hda_manager.error_unless_accessible(provenance_item, trans.user) else: self.security_check(trans, provenance_item, check_accessible=True) out = self._get_record(trans, provenance_item, follow) return out def _get_record(self, trans, item, follow): if item is not None: if item.copied_from_library_dataset_dataset_association: item = item.copied_from_library_dataset_dataset_association job = item.creating_job if job is not None: return { "id": trans.security.encode_id(item.id), "uuid": (lambda uuid: str(uuid) if uuid else None)(item.dataset.uuid), "job_id": trans.security.encode_id(job.id), "tool_id": job.tool_id, "parameters": self._get_job_record(trans, job, follow), "stderr": job.stderr, "stdout": job.stdout, } else: return { "id": trans.security.encode_id(item.id), "uuid": (lambda uuid: str(uuid) if uuid else None)(item.dataset.uuid) } return None def _get_job_record(self, trans, job, follow): out = {} for p in job.parameters: out[p.name] = p.value for in_d in job.input_datasets: if not in_d.dataset: continue if follow: out[in_d.name] = self._get_record(trans, in_d.dataset, follow) else: out[in_d.name] = { "id": trans.security.encode_id(in_d.dataset.id), "uuid": (lambda uuid: str(uuid) if uuid else None)(in_d.dataset.dataset.uuid), } return out
[docs]class HDAProvenanceController(BaseProvenanceController): controller_name = "history_content_provenance" provenance_item_class = "HistoryDatasetAssociation" provenance_item_id = "history_content_id"
[docs]class LDDAProvenanceController(BaseProvenanceController): controller_name = "ldda_provenance" provenance_item_class = "LibraryDatasetDatasetAssociation" provenance_item_id = "library_content_id"