def run_task_non_parallel(
*,
images: list[dict[str, Any]],
zarr_dir: str,
task: TaskV2,
wftask: WorkflowTaskV2,
workflow_dir_local: Path,
workflow_dir_remote: Path,
runner: BaseRunner,
get_runner_config: GetRunnerConfigType,
dataset_id: int,
history_run_id: int,
task_type: Literal[TaskType.NON_PARALLEL, TaskType.CONVERTER_NON_PARALLEL],
user_id: int,
) -> tuple[dict[int, SubmissionOutcome], int]:
"""
This runs server-side (see `executor` argument)
"""
if task_type not in [
TaskType.NON_PARALLEL,
TaskType.CONVERTER_NON_PARALLEL,
]:
raise ValueError(f"Invalid {task_type=} for `run_task_non_parallel`.")
# Get TaskFiles object
task_files = TaskFiles(
root_dir_local=workflow_dir_local,
root_dir_remote=workflow_dir_remote,
task_order=wftask.order,
task_name=wftask.task.name,
component="",
prefix=SUBMIT_PREFIX,
)
runner_config = get_runner_config(
shared_config=runner.shared_config,
wftask=wftask,
which_type="non_parallel",
tot_tasks=1,
)
function_kwargs = {
"zarr_dir": zarr_dir,
**(wftask.args_non_parallel or {}),
}
if task_type == TaskType.NON_PARALLEL:
function_kwargs["zarr_urls"] = [img["zarr_url"] for img in images]
# Database History operations
with next(get_sync_db()) as db:
if task_type == TaskType.NON_PARALLEL:
zarr_urls = function_kwargs["zarr_urls"]
elif task_type == TaskType.CONVERTER_NON_PARALLEL:
zarr_urls = []
history_unit = HistoryUnit(
history_run_id=history_run_id,
status=HistoryUnitStatus.SUBMITTED,
logfile=task_files.log_file_local,
zarr_urls=zarr_urls,
)
db.add(history_unit)
db.commit()
db.refresh(history_unit)
logger.debug(
"[run_task_non_parallel] Created `HistoryUnit` with "
f"{history_run_id=}."
)
history_unit_id = history_unit.id
bulk_upsert_image_cache_fast(
db=db,
list_upsert_objects=[
dict(
workflowtask_id=wftask.id,
dataset_id=dataset_id,
zarr_url=zarr_url,
latest_history_unit_id=history_unit_id,
)
for zarr_url in history_unit.zarr_urls
],
)
result, exception = runner.submit(
base_command=task.command_non_parallel,
workflow_task_order=wftask.order,
workflow_task_id=wftask.task_id,
task_name=wftask.task.name,
parameters=function_kwargs,
task_type=task_type,
task_files=task_files,
history_unit_id=history_unit_id,
config=runner_config,
user_id=user_id,
)
positional_index = 0
num_tasks = 1
outcome = {
positional_index: _process_task_output(
result=result,
exception=exception,
)
}
# NOTE: Here we don't have to handle the
# `outcome[0].exception is not None` branch, since for non_parallel
# tasks it was already handled within submit
if outcome[0].invalid_output:
with next(get_sync_db()) as db:
update_history_unit_no_commit(
history_unit_id=history_unit_id,
status=HistoryUnitStatus.FAILED,
db_sync=db,
)
db.commit()
return outcome, num_tasks