Skip to content

mecfs_bio.build_system.runner.info_purge

Functions:

  • info_purge –

    For each Task stored in Tasks, check whether its dependencies as expressed in info are a subset of its declared task ependencies

Attributes:

logger module-attribute 

logger = get_logger()

info_purge 

info_purge(
    info: VerifyingTraceInfo, tasks: SimpleTasks
) -> VerifyingTraceInfo

For each Task stored in Tasks, check whether its dependencies as expressed in info are a subset of its declared task ependencies If this is not the case, purge it from info

Source code in mecfs_bio/build_system/runner/info_purge.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def info_purge(info: VerifyingTraceInfo, tasks: SimpleTasks) -> VerifyingTraceInfo:
    """
    For each Task stored in Tasks, check whether its dependencies as expressed in info  are a subset of its declared task ependencies
    If this is not the case, purge it from info
    """
    info_dict = deepcopy(info.trace_store)
    logger.debug("Looking for invalid dependencies")
    for task in tasks.values():
        if task.asset_id in info_dict:
            recorded_deps = _get_deps_from_trace_record(info_dict[task.asset_id])
            actual_deps = set([item.asset_id for item in task.deps])
            if not (recorded_deps <= actual_deps):
                logger.debug(
                    f"Dependencies of asset {task.asset_id} appear to be outdated.\n Recorded deps:{recorded_deps}.  Actual deps: {actual_deps}.  Purging from info store."
                )
                info_dict.pop(task.asset_id)
    logger.debug("Done search for invalid dependencies")
    return attrs.evolve(info, trace_store=info_dict)