import datetime

from prefect import task, Flow, Parameter
from prefect.engine.cache_validators import all_parameters
from prefect.engine.result_handlers import LocalResultHandler

local_cache = LocalResultHandler(dir='~/data/prefect_cache')
long_time = datetime.timedelta(weeks=52)


@task(result_handler=local_cache, checkpoint=True, state_handlers=[checkpoint], cache_validator=all_parameters,
      cache_for=long_time, trigger=trigger)
def fetch(x):
    return x

@task(result_handler=local_cache, checkpoint=True, state_handlers=[checkpoint], cache_validator=all_parameters,
      cache_for=long_time, trigger=trigger)
def inc(y):
    return y + 1

with Flow('test') as flow:
    n = Parameter('n')
    f = fetch(n)
    i = inc(f)

if __name__ == '__main__':
    flow.run(n=1)

Cached