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,
) -> tuple[Asset, list[tuple[AssetId, Asset]]]
Source code in mecfs_bio/build_system/rebuilder/tracking_sandboxed_execute.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def tracking_sandboxed_execute(
    task: Task,
    meta_to_path: MetaToPath,
    wf: WF,
    fetch: Fetch,
) -> 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()
    )
    return result, deps