perago.build_workspace_task_output#

perago.build_workspace_task_output(task, input_workspace, published_ref, raw_result)[源代码]#

Validate and wrap a completed workspace task output.

build_workspace_task_output combines a published workspace reference with a task body result. It is useful for tests and runtime integrations that already performed download, body execution, staging, and publication.

Parameters:

taskTaskDefinition: Loaded workspace task definition whose output model validates raw_result.
input_workspaceWorkspaceInput or mapping of str to Any: Original workspace input used for the attempt. The repository, branch, and ref type are preserved in the output.
published_refstr: Reference returned by workspace publication.
raw_resultobject: Value returned by the task callable. It is validated by the task output model with extra fields forbidden.

Returns:

dict of str to Any: Output wrapper containing "workspace" with the published ref and "result" with the validated task output.

Raises:

TaskInputError: If task is not a workspace task.
pydantic.ValidationError: If input_workspace or raw_result is invalid.

参数:

task (TaskDefinition)
input_workspace (WorkspaceInput | Mapping[str, Any])
published_ref (str)
raw_result (object)

返回类型:

dict[str, Any]

参见

run_workspace_task_attempt: Full workspace attempt execution flow.
WorkspaceInput.published_output: Derive the published workspace output model.
build_workspace_free_task_output: Build the workspace-free output wrapper.

Examples

>>> task_def = load_module_task("app.workers.features_build")
>>> output = build_workspace_task_output(
...     task_def,
...     {"repository": "song-000123", "branch": "main", "ref_type": "commit", "ref": "..."},
...     "published-ref",
...     {"row_count": 100, "feature_count": 24},
... )
>>> sorted(output)
['result', 'workspace']