opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

move more logic to the workflow service

Browse Source
This commit is contained in:
Martin Cech
2022-10-19 10:27:33 -07:00
parent efb8019f60
commit 26f7249b6c
2 changed files with 12 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/webapps/galaxy/api/workflows.py
+2 -7
View File
@@ -1241,7 +1241,6 @@ SkipStepCountsQueryParam: bool = Query(
class FastAPIWorkflows:
service: WorkflowsService = depends(WorkflowsService)
invocations_service: InvocationsService = depends(InvocationsService)
workflows_manager: WorkflowsManager = depends(WorkflowsManager)
@router.get(
"/api/workflows",
@@ -1412,9 +1411,7 @@ class FastAPIWorkflows:
trans: ProvidesUserContext = DependsOnTrans,
workflow_id: DecodedDatabaseIdField = StoredWorkflowIDPathParam,
):
workflow_to_delete = self.workflows_manager.get_stored_workflow(trans, workflow_id)
self.workflows_manager.check_security(trans, workflow_to_delete)
self.workflows_manager.delete(workflow_to_delete)
self.service.delete(trans, workflow_id)
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.post(
@@ -1426,9 +1423,7 @@ class FastAPIWorkflows:
trans: ProvidesUserContext = DependsOnTrans,
workflow_id: DecodedDatabaseIdField = StoredWorkflowIDPathParam,
):
workflow_to_undelete = self.workflows_manager.get_stored_workflow(trans, workflow_id)
self.workflows_manager.check_security(trans, workflow_to_undelete)
self.workflows_manager.undelete(workflow_to_undelete)
self.service.undelete(trans, workflow_id)
return Response(status_code=status.HTTP_204_NO_CONTENT)
# TODO: remove this endpoint after 23.1 release
lib/galaxy/webapps/galaxy/services/workflows.py
+10
View File
@@ -99,6 +99,16 @@ class WorkflowsService(ServiceBase):
return workflows, total_matches
return rval, total_matches
def delete(self, trans, workflow_id):
workflow_to_delete = self._workflows_manager.get_stored_workflow(trans, workflow_id)
self._workflows_manager.check_security(trans, workflow_to_delete)
self._workflows_manager.delete(workflow_to_delete)
def undelete(self, trans, workflow_id):
workflow_to_undelete = self._workflows_manager.get_stored_workflow(trans, workflow_id)
self._workflows_manager.check_security(trans, workflow_to_undelete)
self._workflows_manager.undelete(workflow_to_undelete)
def __get_full_shed_url(self, url):
for shed_url in self._tool_shed_registry.tool_sheds.values():
if url in shed_url: