protocol_manager.dal.task_dal#

Data access layer (DAL) between fastapi endpoint and sql database.

Functions#

`add_task_data`(→ app.db.postgres.AcquisitionTask)	Add new acquisition task to database.
`get_task_data`(→ app.db.postgres.AcquisitionTask \| None)	Get acquisition task by id.
`get_all_task_data`(→ list[app.db.postgres.AcquisitionTask])	Get all acquisition tasks assigned to a protocol, ordered by position.
`get_all_task_template_data`(...)	Get all acquisition task templates.
`delete_task_data`(→ bool)	Delete acquisition task by id.
`update_task_data`(→ app.db.postgres.AcquisitionTask \| None)	Update existing acquisition task in database.
`update_task_status`(...)	Update only the status field of an acquisition task.
`reorder_tasks_data`(→ bool)	Update the position of multiple tasks.

async protocol_manager.dal.task_dal.add_task_data(payload: scanhub_libraries.models.BaseAcquisitionTask, creator: str) → app.db.postgres.AcquisitionTask#: Add new acquisition task to database.

async protocol_manager.dal.task_dal.get_task_data(task_id: uuid.UUID) → app.db.postgres.AcquisitionTask | None#: Get acquisition task by id.

async protocol_manager.dal.task_dal.get_all_task_data(protocol_id: uuid.UUID) → list[app.db.postgres.AcquisitionTask]#: Get all acquisition tasks assigned to a protocol, ordered by position.

async protocol_manager.dal.task_dal.get_all_task_template_data() → list[app.db.postgres.AcquisitionTask]#: Get all acquisition task templates.

async protocol_manager.dal.task_dal.delete_task_data(task_id: uuid.UUID) → bool#: Delete acquisition task by id.

async protocol_manager.dal.task_dal.update_task_data(task_id: uuid.UUID, payload: scanhub_libraries.models.BaseAcquisitionTask) → app.db.postgres.AcquisitionTask | None#: Update existing acquisition task in database.

async protocol_manager.dal.task_dal.update_task_status(task_id: uuid.UUID, status: str) → app.db.postgres.AcquisitionTask | None#: Update only the status field of an acquisition task.

async protocol_manager.dal.task_dal.reorder_tasks_data(task_ids: list[uuid.UUID]) → bool#: Update the position of multiple tasks.