Skip to content

mecfs_bio.build_system.runner.simple_runner

Classes:

Attributes:

logger module-attribute 

logger = structlog.get_logger()

SimpleRunner

Simple wrapper class that orchestrates an execution of the workflow build system with a topological scheduler and verifying trace rebuilder

info_store: path at which to store persistent cash for build-system internal information asset_root: root under which to create the asset store, and the default root for any subtree not claimed by path_remap tracer: algorithm uses to calculate verifying traces of assets. An example would be a hashing algorithm. changing this forces all assets to be rebuilt post_execute: hook invoked after each task materialization. Defaults to running gc.collect() and logging RSS, which keeps cycle-held memory from accumulating across long pipeline runs. Pass noop_post_execute_hook from mecfs_bio.build_system.rebuilder.sandboxed_execute to disable. path_remap: rules sending selected store subtrees to other filesystems. Empty by default, which keeps the whole store under asset_root. See the remapping_meta_to_path module for what is worth remapping and what is not.

Methods:

  • run –

    Targets: the ultimate targets we aim to produce. All transitive dependencies of these targets will either be rebuilt, or fetched (determined according to that status of their trace)

Attributes:

asset_root instance-attribute 

asset_root: Path

info_store instance-attribute 

info_store: Path

meta_to_path property 

meta_to_path: MetaToPath

path_remap class-attribute instance-attribute 

path_remap: tuple[PathRemapRule, ...] = ()

post_execute class-attribute instance-attribute 

post_execute: Callable[[Task], None] = (
    gc_collect_post_execute_hook
)

tracer class-attribute instance-attribute 

tracer: Tracer = SimpleHasher.md5_hasher()

run 

run(
    targets: list[Task],
    must_rebuild_transitive: Sequence[Task] = tuple(),
    incremental_save: bool = True,
    settings: TopologicalSchedulerSettings = TopologicalSchedulerSettings(),
) -> Mapping[AssetId, Asset]

Targets: the ultimate targets we aim to produce. All transitive dependencies of these targets will either be rebuilt, or fetched (determined according to that status of their trace) must_rebuild_transitive: list of tasks that the rebuilder will be forced to rebuild, regardless of the status of their trace. - This is particularly useful when you have changed the code that generates and asset, and so want it and its dependees to be regenerated. returns: mapping from asset id to file system information for all assets that were built or retrieved as part of the execution of the scheduler

Raises RemapRootUnavailableError, before any work is scheduled, if a configured path_remap root is not present.

Source code in mecfs_bio/build_system/runner/simple_runner.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def run(
    self,
    targets: list[Task],
    must_rebuild_transitive: Sequence[Task] = tuple(),
    incremental_save: bool = True,
    settings: TopologicalSchedulerSettings = TopologicalSchedulerSettings(),
) -> Mapping[AssetId, Asset]:
    """
    Targets: the ultimate targets we aim to produce.  All transitive dependencies of these targets will either be rebuilt, or fetched (determined according to that status of their trace)
    must_rebuild_transitive: list of tasks that the rebuilder will be forced to rebuild, regardless of the status of their trace.
       - This is particularly useful when you have changed the code that generates and asset, and so want it and its dependees to be regenerated.
    returns:
    mapping from asset id to file system information for all assets that were built or retrieved as part of the execution of the scheduler

    Raises RemapRootUnavailableError, before any work is scheduled, if a configured
    path_remap root is not present.
    """
    check_remap_roots_available(self.path_remap)
    warn_on_unmigrated_assets(self.asset_root, self.path_remap)
    if self.info_store.is_file():
        info = VerifyingTraceInfo.deserialize(self.info_store)
    else:
        info = VerifyingTraceInfo.empty()
    if incremental_save:
        logger.debug("incremental save is enabled")
    rebuilder = VerifyingTraceRebuilder(
        tracer=self.tracer, post_execute=self.post_execute
    )
    wf = make_wf(downloader=RobustWFDownloader())
    meta_to_path = self.meta_to_path
    tasks = find_tasks(targets)
    must_rebuild_graph = dependees_of_targets_from_tasks(
        tasks=tasks,
        targets=[task.asset_id for task in must_rebuild_transitive],
    )
    info = attrs.evolve(info, must_rebuild=set(must_rebuild_graph.nodes))
    info = info_purge(info, tasks=tasks)
    incremental_save_path = self.info_store if incremental_save else None
    store, info = topological(
        rebuilder=rebuilder,
        tasks=tasks,
        info=info,
        wf=wf,
        meta_to_path=meta_to_path,
        targets=[target.asset_id for target in targets],
        incremental_save_path=incremental_save_path,
        settings=settings,
    )
    info.serialize(self.info_store)
    _print_target_locs(store, targets=targets)
    return store