"""
API operations provenance
"""
import logging
from paste.httpexceptions import (
HTTPBadRequest,
HTTPNotImplemented,
)
from galaxy import web
from galaxy.managers.hdas import HDAManager
from galaxy.util import string_as_bool
from . import (
BaseGalaxyAPIController,
depends,
)
log = logging.getLogger(__name__)
[docs]class BaseProvenanceController(BaseGalaxyAPIController):
""" """
[docs] @web.legacy_expose_api
def index(self, trans, **kwd):
follow = string_as_bool(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):
raise HTTPNotImplemented()
[docs] @web.legacy_expose_api
def create(self, trans, tag_name, payload=None, **kwd):
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"
hda_manager: HDAManager = depends(HDAManager)
[docs]class LDDAProvenanceController(BaseProvenanceController):
controller_name = "ldda_provenance"
provenance_item_class = "LibraryDatasetDatasetAssociation"
provenance_item_id = "library_content_id"
hda_manager: HDAManager = depends(HDAManager)