server.web.routes.v3.cli.workflow
=================================

.. py:module:: server.web.routes.v3.cli.workflow

.. autoapi-nested-parse::

   Routes for Workflow.



Attributes
----------

.. autoapisummary::

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


Functions
---------

.. autoapisummary::

   server.web.routes.v3.cli.workflow.get_workflow_validation_status
   server.web.routes.v3.cli.workflow._check_downstream_tasks_status
   server.web.routes.v3.cli.workflow.get_workflow_tasks
   server.web.routes.v3.cli.workflow.get_workflow_user_validation
   server.web.routes.v3.cli.workflow.get_workflow_run_for_workflow_reset
   server.web.routes.v3.cli.workflow.reset_workflow
   server.web.routes.v3.cli.workflow.get_workflow_status
   server.web.routes.v3.cli.workflow.get_workflow_status_viz
   server.web.routes.v3.cli.workflow.workflows_by_user_form
   server.web.routes.v3.cli.workflow.task_details_by_wf_id
   server.web.routes.v3.cli.workflow.wf_details_by_wf_id


Module Contents
---------------

.. py:data:: logger

.. py:function:: get_workflow_validation_status(request: fastapi.Request, db: sqlalchemy.orm.Session = Depends(get_db)) -> jobmon.server.web.schemas.workflow.WorkflowValidationResponse
   :async:


   Check if workflow is valid.


.. py:function:: _check_downstream_tasks_status(session: sqlalchemy.orm.Session, task_ids: List[int], workflow_id: int, dag_id: int) -> bool

   Check if all downstream tasks are in valid states (G, I, Q).

   :param session: Database session
   :param task_ids: List of task IDs to check downstream tasks for
   :param workflow_id: Workflow ID
   :param dag_id: DAG ID

   :returns: True if all downstream tasks are in valid states, False otherwise


.. py:function:: get_workflow_tasks(workflow_id: int, limit: int, status: Optional[List[str]] = Query(None), db: sqlalchemy.orm.Session = Depends(get_db)) -> jobmon.server.web.schemas.workflow.WorkflowTasksResponse

   Get the tasks for a given workflow.


.. py:function:: 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.


.. py:function:: 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:
       1. The last workflow_run of the current workflow must be in error state.
       2. This last workflow_run must have been started by the input username.
       3. This last workflow_run is in status 'E'


.. py:function:: reset_workflow(workflow_id: int, request: fastapi.Request, db: sqlalchemy.orm.Session = Depends(get_db)) -> starlette.responses.JSONResponse
   :async:


   Update the workflow's status, all its tasks' statuses to 'G'.


.. py:function:: get_workflow_status(workflow_id: Optional[Union[int, str, List[Union[int, str]]]] = Query(None), limit: Optional[int] = Query(None), user: Optional[List[str]] = Query(None), db: sqlalchemy.orm.Session = Depends(get_db)) -> jobmon.server.web.schemas.workflow.WorkflowStatusResponse

   Get the status of the workflow.


.. py:function:: 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.


.. py:function:: workflows_by_user_form(user: Optional[str] = Query(None), tool: Optional[str] = Query(None), wf_name: Optional[str] = Query(None), wf_args: Optional[str] = Query(None), wf_attribute_value: Optional[str] = Query(None), wf_attribute_key: Optional[str] = Query(None), wf_id: Optional[str] = Query(None), date_submitted: Optional[str] = Query(None), date_submitted_end: Optional[str] = Query(None), status: Optional[str] = Query(None), db: sqlalchemy.orm.Session = Depends(get_db)) -> jobmon.server.web.schemas.workflow.WorkflowOverviewResponse

   Fetch associated workflows and workflow runs by username.


.. py:function:: 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.


.. py:function:: 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.