server.web.routes.v3.cli.task
=============================

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

.. autoapi-nested-parse::

   Routes for Tasks.



Attributes
----------

.. autoapisummary::

   server.web.routes.v3.cli.task.logger


Functions
---------

.. autoapisummary::

   server.web.routes.v3.cli.task.get_task_status
   server.web.routes.v3.cli.task.get_task_subdag
   server.web.routes.v3.cli.task.parse_request_data
   server.web.routes.v3.cli.task.validate_workflow_for_update
   server.web.routes.v3.cli.task.create_response
   server.web.routes.v3.cli.task.update_task_statuses
   server.web.routes.v3.cli.task.get_task_dependencies
   server.web.routes.v3.cli.task.get_tasks_recursive
   server.web.routes.v3.cli.task.get_task_resource_usage
   server.web.routes.v3.cli.task.get_downstream_tasks
   server.web.routes.v3.cli.task.get_task_details
   server.web.routes.v3.cli.task.get_task_details_viz


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

.. py:data:: logger

.. py:function:: get_task_status(task_ids: Optional[Union[int, list[int]]] = Query(...), status: Optional[Union[str, list[str]]] = Query(None), db: sqlalchemy.orm.Session = Depends(get_db)) -> jobmon.server.web.schemas.task.TaskStatusResponse

   Get the status of a task.


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


   Used to get the sub dag of a given task.

   It returns a list of sub tasks as well as a list of sub nodes.


.. py:function:: parse_request_data(data: Dict) -> tuple[str, bool, Union[List[int], str], str]

   Parse and validate request data.


.. py:function:: validate_workflow_for_update(task_ids: List[int], session: sqlalchemy.orm.Session) -> str

   Validate workflow status for task updates.

   Validates that:

   - All tasks belong to the same workflow
   - The workflow status allows updates (FAILED, DONE, ABORTED, HALTED) OR
   - All downstream tasks are in valid states (G, I, Q) for non-terminal workflows

   :param task_ids: List of task IDs to validate
   :param session: Database session

   :returns: The workflow status if validation passes

   :raises InvalidUsage: If validation fails with detailed error message


.. py:function:: create_response(new_status: str) -> starlette.responses.JSONResponse

   Create a JSON HTTP response indicating a successful status update.

   :param new_status: The new status that was applied.
   :type new_status: str

   :returns: A FastAPI JSONResponse object with a success message.
   :rtype: JSONResponse


.. py:function:: update_task_statuses(request: fastapi.Request, db: sqlalchemy.orm.Session = Depends(get_db)) -> Any
   :async:


   Update the status of the tasks.

   Description:

   - When ``workflow_id='all'``, it updates all tasks in the workflow with
     ``recursive=False``. This improves performance.
   - When ``recursive=True``, it updates the tasks and its dependencies all
     the way up or down the DAG.
   - When ``recursive=False``, it updates only the tasks in the task_ids list.
   - Validates workflow status before proceeding with updates.
   - After updating the tasks, it checks the workflow status and updates it.


.. py:function:: get_task_dependencies(task_id: int, db: sqlalchemy.orm.Session = Depends(get_db)) -> jobmon.server.web.schemas.task.TaskDependenciesResponse

   Get task's downstream and upstream tasks and their status.


.. py:function:: get_tasks_recursive(direction: str, request: fastapi.Request, db: sqlalchemy.orm.Session = Depends(get_db)) -> jobmon.server.web.schemas.task.TasksRecursiveResponse
   :async:


   Get all input task_ids'.

   Either downstream or upsteam tasks based on direction;
   return all recursive(including input set) task_ids in the defined direction.


.. py:function:: get_task_resource_usage(task_id: int, db: sqlalchemy.orm.Session = Depends(get_db)) -> jobmon.server.web.schemas.task.TaskResourceUsageResponse

   Return the resource usage for a given Task ID.


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


   Get only the direct downstreams of a task.


.. py:function:: get_task_details(task_id: int, db: sqlalchemy.orm.Session = Depends(get_db)) -> jobmon.server.web.schemas.task.TaskInstanceDetailsResponse

   Get information about TaskInstances associated with specific Task ID.


.. py:function:: get_task_details_viz(task_id: int, db: sqlalchemy.orm.Session = Depends(get_db)) -> jobmon.server.web.schemas.task.TaskDetailsResponse

   Get status of Task from Task ID.