Fix double id encoding in summarize_job_parameters

This commit is contained in:
mvdbeek
2024-01-08 11:33:47 +01:00
parent 0f22f79155
commit a7d00a9148
4 changed files with 15 additions and 13 deletions
+10 -10
View File
@@ -236,7 +236,7 @@ class JobManager:
)
return self.job_lock()
def get_accessible_job(self, trans, decoded_job_id):
def get_accessible_job(self, trans, decoded_job_id) -> Job:
job = trans.sa_session.get(Job, decoded_job_id)
if job is None:
raise ObjectNotFound()
@@ -619,7 +619,7 @@ class JobSearch:
return None
def view_show_job(trans, job, full: bool) -> typing.Dict:
def view_show_job(trans, job: Job, full: bool) -> typing.Dict:
is_admin = trans.user_is_admin
job_dict = job.to_dict("element", system_details=is_admin)
if trans.app.config.expose_dataset_path and "command_line" not in job_dict:
@@ -873,7 +873,7 @@ def summarize_destination_params(trans, job):
return destination_params
def summarize_job_parameters(trans, job):
def summarize_job_parameters(trans, job: Job):
"""Produce a dict-ified version of job parameters ready for tabular rendering.
Precondition: the caller has verified the job is accessible to the user
@@ -943,14 +943,14 @@ def summarize_job_parameters(trans, job):
elif input.type == "data" or input.type == "data_collection":
value = []
for element in listify(param_values[input.name]):
encoded_id = trans.security.encode_id(element.id)
element_id = element.id
if isinstance(element, model.HistoryDatasetAssociation):
hda = element
value.append({"src": "hda", "id": encoded_id, "hid": hda.hid, "name": hda.name})
value.append({"src": "hda", "id": element_id, "hid": hda.hid, "name": hda.name})
elif isinstance(element, model.DatasetCollectionElement):
value.append({"src": "dce", "id": encoded_id, "name": element.element_identifier})
value.append({"src": "dce", "id": element_id, "name": element.element_identifier})
elif isinstance(element, model.HistoryDatasetCollectionAssociation):
value.append({"src": "hdca", "id": encoded_id, "hid": element.hid, "name": element.name})
value.append({"src": "hdca", "id": element_id, "hid": element.hid, "name": element.name})
else:
raise Exception(
f"Unhandled data input parameter type encountered {element.__class__.__name__}"
@@ -1011,7 +1011,7 @@ def summarize_job_parameters(trans, job):
return {
"parameters": parameters,
"has_parameter_errors": has_parameter_errors,
"outputs": summarize_job_outputs(job=job, tool=tool, params=params_objects, security=trans.security),
"outputs": summarize_job_outputs(job=job, tool=tool, params=params_objects),
}
@@ -1026,7 +1026,7 @@ def get_output_name(tool, output, params):
pass
def summarize_job_outputs(job: model.Job, tool, params, security):
def summarize_job_outputs(job: model.Job, tool, params):
outputs = defaultdict(list)
output_labels = {}
possible_outputs = (
@@ -1046,7 +1046,7 @@ def summarize_job_outputs(job: model.Job, tool, params, security):
outputs[output_name].append(
{
"label": label,
"value": {"src": src, "id": security.encode_id(getattr(output_association, attribute))},
"value": {"src": src, "id": getattr(output_association, attribute)},
}
)
return outputs
+2 -1
View File
@@ -46,6 +46,7 @@ from galaxy.managers.jobs import (
summarize_job_parameters,
)
from galaxy.managers.licenses import LicensesManager
from galaxy.model import Job
from galaxy.model.item_attrs import get_item_annotation_str
from galaxy.model.orm.now import now
from galaxy.schema import PdfDocumentType
@@ -655,7 +656,7 @@ class ToBasicMarkdownDirectiveHandler(GalaxyInternalMarkdownDirectiveHandler):
markdown += f"| {title} | {value} |\n"
return (markdown, True)
def handle_job_parameters(self, line, job):
def handle_job_parameters(self, line, job: Job):
markdown = """
| Input Parameter | Value |
|-----------------|-------|
+1 -1
View File
@@ -4820,7 +4820,7 @@ class DatasetInstance(RepresentById, UsesCreateAndUpdateTime, _HasTable):
return _source_dataset_chain(self, [])
@property
def creating_job(self):
def creating_job(self) -> Optional[Job]:
# TODO this should work with `return self.dataset.job` (revise failing unit tests)
creating_job_associations = None
if self.creating_job_associations:
+2 -1
View File
@@ -17,6 +17,7 @@ from galaxy.managers.jobs import (
JobSearch,
view_show_job,
)
from galaxy.model import Job
from galaxy.schema.fields import DecodedDatabaseIdField
from galaxy.schema.jobs import JobAssociation
from galaxy.schema.schema import JobIndexQueryPayload
@@ -107,7 +108,7 @@ class JobsService(ServiceBase):
job_id: Optional[int] = None,
dataset_id: Optional[int] = None,
hda_ldda: str = "hda",
):
) -> Optional[Job]:
if job_id is not None:
return self.job_manager.get_accessible_job(trans, decoded_job_id=job_id)
elif dataset_id is not None: