Hi all I have a problem with flow again

Question

bral · Answer

from prefect import task Flow case from random import randint from prefect schedules import IntervalSchedule from datetime import timedelta task def is running return False def get files return i for i in range 0 randint 0 15 task def print lst print lst print len lst task def read file file return True task def preprocess file file return True task def save file file return True schedule = IntervalSchedule interval=timedelta minutes=1 with Flow process schedule=schedule as flow condition = is running with case condition False as cond files = get files print files for file in files 0 3 r = read file file p = preprocess file file s = save file file r set downstream p p set downstream s flow run

bral · Answer

Every time variable files is same Why

Alex Cano · Answer

If you re running the script and waiting until the schedule kicks it off my guess is because the ```files = get files ``` line only gets executed once and thus never changes If you re wanting it to change you can wrap it within a task and return it

Jim Crist-Harif · Answer

You re calling `get files` as a function not a task there s no ` task` decorator on it So `get files` runs at flow build time while the rest of your functions run at flow run time This means that the value of `files` is frozen in your flow and won t update each run

bral · Answer

Thanks I added decorator and how iterate over files

Jim Crist-Harif · Answer

You ll want to look at prefect s mapping concept <https docs prefect io core concepts mapping html prefect approach> The loop over files needs to be encoded as a mapped task or a series of mapped tasks rather than an explicit for loop

Jim Crist-Harif · Answer

See also <https docs prefect io core idioms mapping html>

bral · Answer

< Jim Crist Harif> thanks