Skip to content

mecfs_bio.build_system.rebuilder.sandboxed_execute

Functions:

noop_post_execute_hook 

noop_post_execute_hook(task: Task) -> None

Default post-execute hook: do nothing.

Source code in mecfs_bio/build_system/rebuilder/sandboxed_execute.py 
25
26
def noop_post_execute_hook(task: Task) -> None:
    """Default post-execute hook: do nothing."""

sandboxed_execute 

sandboxed_execute(
    task: Task,
    meta_to_path: MetaToPath,
    wf: WF,
    fetch: Fetch,
    post_execute: Callable[
        [Task], None
    ] = noop_post_execute_hook,
) -> Asset

Execute a task in a temporary directory, then move its result to the final location determined by the task's metadata.

post_execute is invoked with the task after the temp directory is cleaned up and the asset has been moved into place. The default is a no-op.

Source code in mecfs_bio/build_system/rebuilder/sandboxed_execute.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def sandboxed_execute(
    task: Task,
    meta_to_path: MetaToPath,
    wf: WF,
    fetch: Fetch,
    post_execute: Callable[[Task], None] = noop_post_execute_hook,
) -> Asset:
    """
    Execute a task in a temporary directory, then move its result to the final location
    determined by the task's metadata.

    `post_execute` is invoked with the task after the temp directory is cleaned up and
    the asset has been moved into place. The default is a no-op.
    """
    meta = task.meta
    target_path = meta_to_path(meta)
    with tempfile.TemporaryDirectory() as temp_dir:
        temp_path = Path(temp_dir) / "temp"
        temp_path.mkdir(parents=True, exist_ok=True)
        result = task.execute(scratch_dir=temp_path, fetch=fetch, wf=wf)
        target_path.parent.mkdir(exist_ok=True, parents=True)
        result = _move_asset(result, target_path)
        logger.debug(f"Saved asset {task.asset_id} to {target_path}.")
    post_execute(task)
    return result