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

"""
API operations provenance
"""
import logging
from galaxy import web
from galaxy.web.base.controller import BaseAPIController
from paste.httpexceptions import HTTPNotImplemented, HTTPBadRequest
from galaxy import managers

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.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.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.expose_api def create(self, trans, tag_name, payload=None, **kwd): payload = payload or {} raise HTTPNotImplemented()
[docs] @web.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"