server.web.routes.v3.cli.workflow

Routes for Workflow.

Attributes

logger

Functions

`get_workflow_validation_status`(...)	Check if workflow is valid.
`get_workflow_tasks`(, db)	Get the tasks for a given workflow.
`get_workflow_requested_resources`(...)	Return the requested-resource clusters for a workflow.
`get_workflow_user_validation`(...)	Return all usernames associated with a given workflow_id's workflow runs.
`get_workflow_run_for_workflow_reset`(...)	Last workflow_run_id associated with a given workflow_id started by the username.
`reset_workflow`() → starlette.responses.JSONResponse)	Update the workflow's status, all its tasks' statuses to 'G'.
`get_workflow_status`(, limit, user, db)	Get the status of the workflow.
`get_workflow_status_viz`(, db)	Get the status of the workflows for GUI.
`workflows_by_user_form`(, tool, wf_name, wf_args, ...)	Fetch workflow overview with filtering.
`task_details_by_wf_id`(...)	Fetch Task details associated with Workflow ID and TaskTemplate name.
`wf_details_by_wf_id`(...)	Fetch name, args, dates, tool for a Workflow provided WF ID.

Module Contents

server.web.routes.v3.cli.workflow.logger

async server.web.routes.v3.cli.workflow.get_workflow_validation_status(request: fastapi.Request, db: sqlalchemy.orm.Session = Depends(get_db)) → jobmon.server.web.schemas.workflow.WorkflowValidationResponse: Check if workflow is valid.

server.web.routes.v3.cli.workflow.get_workflow_tasks(workflow_id: int, limit: int, status: List[str] | None = Query(None), db: sqlalchemy.orm.Session = Depends(get_db)) → jobmon.server.web.schemas.workflow.WorkflowTasksResponse: Get the tasks for a given workflow.

server.web.routes.v3.cli.workflow.get_workflow_requested_resources(workflow_id: int, db: sqlalchemy.orm.Session = Depends(get_db)) → jobmon.server.web.schemas.workflow.WorkflowRequestedResourcesResponse

Return the requested-resource clusters for a workflow.

One entry per unique (task_template, task_resources_id), with num_tasks, queue_name, and the raw requested_resources JSON blob. Available as soon as the workflow is bound — no TaskInstance required.

server.web.routes.v3.cli.workflow.get_workflow_user_validation(workflow_id: int, username: str, db: sqlalchemy.orm.Session = Depends(get_db)) → jobmon.server.web.schemas.workflow.WorkflowUserValidationResponse

Return all usernames associated with a given workflow_id’s workflow runs.

Used to validate permissions for a self-service request.

server.web.routes.v3.cli.workflow.get_workflow_run_for_workflow_reset(workflow_id: int, username: str, db: sqlalchemy.orm.Session = Depends(get_db)) → jobmon.server.web.schemas.workflow.WorkflowRunForResetResponse

Last workflow_run_id associated with a given workflow_id started by the username.

Used to validate for workflow_reset:

The last workflow_run of the current workflow must be in error state.
This last workflow_run must have been started by the input username.
This last workflow_run is in status ‘E’

async server.web.routes.v3.cli.workflow.reset_workflow(workflow_id: int, request: fastapi.Request, db: sqlalchemy.orm.Session = Depends(get_db)) → starlette.responses.JSONResponse: Update the workflow’s status, all its tasks’ statuses to ‘G’.

server.web.routes.v3.cli.workflow.get_workflow_status(workflow_id: int | str | List[int | str] | None = Query(None), limit: int | None = Query(None), user: List[str] | None = Query(None), db: sqlalchemy.orm.Session = Depends(get_db)) → jobmon.server.web.schemas.workflow.WorkflowStatusResponse: Get the status of the workflow.

server.web.routes.v3.cli.workflow.get_workflow_status_viz(workflow_ids: List[int] = Query(None), db: sqlalchemy.orm.Session = Depends(get_db)) → Any: Get the status of the workflows for GUI.

server.web.routes.v3.cli.workflow.workflows_by_user_form(user: str | None = Query(None), tool: str | None = Query(None), wf_name: str | None = Query(None), wf_args: str | None = Query(None), wf_attribute_value: str | None = Query(None), wf_attribute_key: str | None = Query(None), wf_id: str | None = Query(None), date_submitted: str | None = Query(None), date_submitted_end: str | None = Query(None), status: str | None = Query(None), user_exclude: List[str] | None = Query(None, alias='user!'), tool_exclude: List[str] | None = Query(None, alias='tool!'), status_exclude: List[str] | None = Query(None, alias='status!'), wf_name_contains: bool = Query(False), wf_args_contains: bool = Query(False), wf_attr: List[str] | None = Query(None), db: sqlalchemy.orm.Session = Depends(get_db)) → jobmon.server.web.schemas.workflow.WorkflowOverviewResponse

Fetch workflow overview with filtering.

wf_attr accepts repeated key:value pairs for multi-attribute AND filtering, e.g. ?wf_attr=project:fhs&wf_attr=release_id:7.

server.web.routes.v3.cli.workflow.task_details_by_wf_id(workflow_id: int, tt_name: str, db: sqlalchemy.orm.Session = Depends(get_db)) → jobmon.server.web.schemas.workflow.TaskTableResponse: Fetch Task details associated with Workflow ID and TaskTemplate name.

server.web.routes.v3.cli.workflow.wf_details_by_wf_id(workflow_id: int, db: sqlalchemy.orm.Session = Depends(get_db)) → List[jobmon.server.web.schemas.workflow.WorkflowDetailsItem]: Fetch name, args, dates, tool for a Workflow provided WF ID.