app.api.task_api#

Definition of task API endpoints.

Attributes#

Classes#

Task reorder model.

Functions#

`create_task`(...)	Create a new acquisition task.
`create_task_from_template`(...)	Create a new acquisition task from a template.
`get_task`(]) → scanhub_libraries.models.AcquisitionTaskOut)	Get an existing task.
`get_all_protocol_tasks`(...)	Get all tasks of a protocol.
`get_all_task_templates`(...)	Get all task templates.
`delete_task`(]) → None)	Delete a task.
`reorder_tasks`(]) → None)	Reorder tasks by updating their position.
`update_task`(...)	Update an existing task.
`update_task_status`(...)	Update only the status of a task (called by Dagster sensors).

Module Contents#

app.api.task_api.task_router#

class app.api.task_api.TaskReorder#

Bases: pydantic.BaseModel

Task reorder model.

task_ids: list[uuid.UUID]#

async app.api.task_api.create_task(payload: scanhub_libraries.models.BaseAcquisitionTask, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) → scanhub_libraries.models.AcquisitionTaskOut#: Create a new acquisition task.

async app.api.task_api.create_task_from_template(protocol_id: uuid.UUID, template_id: uuid.UUID, new_task_is_template: bool, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) → scanhub_libraries.models.AcquisitionTaskOut#: Create a new acquisition task from a template.

async app.api.task_api.get_task(task_id: uuid.UUID | str, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) → scanhub_libraries.models.AcquisitionTaskOut#: Get an existing task.

async app.api.task_api.get_all_protocol_tasks(protocol_id: uuid.UUID | str, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) → list[scanhub_libraries.models.AcquisitionTaskOut]#: Get all tasks of a protocol.

async app.api.task_api.get_all_task_templates(user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) → list[scanhub_libraries.models.AcquisitionTaskOut]#: Get all task templates.

async app.api.task_api.delete_task(task_id: uuid.UUID | str, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) → None#: Delete a task.

async app.api.task_api.reorder_tasks(payload: TaskReorder, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) → None#: Reorder tasks by updating their position.

async app.api.task_api.update_task(task_id: uuid.UUID | str, payload: scanhub_libraries.models.BaseAcquisitionTask, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) → scanhub_libraries.models.AcquisitionTaskOut#: Update an existing task.

async app.api.task_api.update_task_status(task_id: uuid.UUID | str, status: str, user: Annotated[scanhub_libraries.models.User, Depends(get_current_user)]) → scanhub_libraries.models.AcquisitionTaskOut#: Update only the status of a task (called by Dagster sensors).