@Marvin how can I cache a task's result for reuse during a flow run and clear them after the flow has completed, such that the results aren't stored permanently anywhere. I'm using Prefect 2.

You can use Prefect 2 to cache a task's result during a flow run with the

cache_key_fn

and

cache_expiration

parameters. Here's an example code snippet to illustrate:

python
from prefect import flow, task
from datetime import timedelta

@task(cache_key_fn=lambda *args, **kwargs: "my_cache_key", cache_expiration=timedelta(seconds=30))
def cached_task(param):
    return param * 2

@flow
def my_flow():
    res1 = cached_task(2)
    res2 = cached_task(2)
    return res1, res2

if __name__ == "__main__":
    my_flow()

In this example, the

cached_task

will return the cached result if it is executed with the same parameter value within the next 30 seconds. Be sure to adjust the expiration to meet your requirements. To clear these results after the flow has completed without storing them permanently, Prefect automatically manages cache expiration based on the time you specify, preventing long-term storage. For more details, see the relevant Prefect documentation and code examples.