task_interface

TaskOutput

Bases: BaseModel

Model for task output.

ATTRIBUTE	DESCRIPTION
image_list_updates	List of image-list elements to be created or updated. TYPE: list[SingleImageTaskOutput]
image_list_removals	List of Zarr URLs to be removed from the dataset image list. TYPE: list[ZarrUrlStr]

Source code in fractal_server/runner/v2/task_interface.py

class TaskOutput(BaseModel):
    """
    Model for task output.

    Attributes:
        image_list_updates:
            List of image-list elements to be created or updated.
        image_list_removals:
            List of Zarr URLs to be removed from the dataset image list.
    """

    model_config = ConfigDict(extra="forbid")

    image_list_updates: list[SingleImageTaskOutput] = Field(
        default_factory=list
    )
    image_list_removals: list[ZarrUrlStr] = Field(default_factory=list)

    def check_zarr_urls_are_unique(self) -> None:
        zarr_urls = [img.zarr_url for img in self.image_list_updates]
        zarr_urls.extend(self.image_list_removals)
        if len(zarr_urls) != len(set(zarr_urls)):
            duplicates = [
                zarr_url
                for zarr_url in set(zarr_urls)
                if zarr_urls.count(zarr_url) > 1
            ]
            msg = (
                "TaskOutput "
                f"({len(self.image_list_updates)} image_list_updates and "
                f"{len(self.image_list_removals)} image_list_removals) "
                "has non-unique zarr_urls:"
            )
            for duplicate in duplicates:
                msg = f"{msg}\n{duplicate}"
            raise ValueError(msg)