hi @Marvin. Just one brief question. Can you give me a (perhaps simplest or shortest) working example of prefect that resuming the flow from the previous run? I tried several ways but it keeps redoing all the tasks all over again when restarted the flow. Thanks million!

Let's find the information you need. Could you confirm which version of Prefect you are using? It's important as Prefect 2.x and 3.x might handle this differently. If you don't know your version, that's okay too; I'll check both for you!

@Marvin 3.0

In Prefect 3.0, resuming a flow from a previous run without redoing tasks can be achieved using task caching. This allows you to prevent rerunning tasks if their inputs and outputs are unchanged, by storing their results. Here's a brief example to illustrate the concept:

python
from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta

# Define a task with caching enabled
@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(days=1))
def my_task(x):
    print(f"Running my_task with input: {x}")
    return x * 2

# Define a flow that uses the caching task
@flow
def my_flow(y):
    result = my_task(y)
    print(f"Flow result: {result}")

# Execute the flow with a specific input
my_flow(10)

In this example, the

my_task

function will only rerun if the input changes or the cache expires. This way, if you resume the flow with the same inputs, it won't redo the work unless necessary. For more details on caching and task configurations in Prefect 3.0, you can check out the official documentation here. If you need a reminder why it feels like you're already at yesterday's party when flows don't resume correctly, just remember Marvin... he wouldn't have cached anything either, but you can!