Skip to content

mecfs_bio.build_system.rebuilder.tracking_sandboxed_execute

Functions:

tracking_sandboxed_execute 

tracking_sandboxed_execute(
    task: Task,
    meta_to_path: MetaToPath,
    wf: WF,
    fetch: Fetch,
    post_execute: Callable[
        [Task], None
    ] = noop_post_execute_hook,
) -> tuple[Asset, list[tuple[AssetId, Asset]]]
Source code in mecfs_bio/build_system/rebuilder/tracking_sandboxed_execute.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def tracking_sandboxed_execute(
    task: Task,
    meta_to_path: MetaToPath,
    wf: WF,
    fetch: Fetch,
    post_execute: Callable[[Task], None] = noop_post_execute_hook,
) -> tuple[Asset, list[tuple[AssetId, Asset]]]:
    deps: list[tuple[AssetId, Asset]] = []

    class TrackingFetch(Fetch):
        def __call__(self, asset_id: AssetId) -> Asset:
            a = fetch(asset_id)
            deps.append((asset_id, a))
            return a

    result = sandboxed_execute(
        task=task,
        meta_to_path=meta_to_path,
        wf=wf,
        fetch=TrackingFetch(),
        post_execute=post_execute,
    )
    return result, deps