Yas Mah

03/03/2022, 1:57 PM
Hello 🙂 which possibilities are there to access the result of a task and use it in an other operation with the flow, which is not a task:
def get_access_paths(base_path:Path):
    return base_path

with Flow("flow") as flow:
    base_path = Parameter("base_path", default=pathlib.Path(__file__).parent.parent.resolve())
    data_access = get_access_paths(base_path)
    files = [str(Path.joinpath(data_access, x)) for x in data_access.glob('*') if x.is_file()]

    input = Parameter("input", default=files)

Kevin Kho

03/03/2022, 2:04 PM
Hi @Yas Mah, 🙂 I think the thinking is the other way around. You need to make the non tasks into tasks to defer their execution to run time. Otherwise they will run while the Flow (DAG) is being constructed.
🙏 1

Yas Mah

03/03/2022, 2:49 PM
Thank you @Kevin Kho, it makes sense!