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.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"