(self, compiled)
| 195 | return (plan, task.output_list(), task.workspace_type()) |
| 196 | |
| 197 | def _run_compiled(self, compiled): |
| 198 | plan, output_list, workspace_type = compiled |
| 199 | |
| 200 | # make sure the output blobs belong to the parent workspace |
| 201 | outputs = [] |
| 202 | for name in output_list.names(): |
| 203 | self._ws.create_blob(str(name)) |
| 204 | outputs.append(core.BlobReference(str(name))) |
| 205 | output_list.set_values(outputs, _fetch_func=self._fetch_output) |
| 206 | task_ws = ( |
| 207 | workspace.C.Workspace(self._ws) |
| 208 | if workspace_type == WorkspaceType.PRIVATE else self._ws) |
| 209 | with workspace.WorkspaceGuard(task_ws): |
| 210 | task_ws.run(plan) |
| 211 | |
| 212 | def _fetch_output(self, output): |
| 213 | return self._ws.blobs[str(output)].fetch() |
nothing calls this directly
no test coverage detected