I think

all_parameter

is Prefect parameters, not task inputs.

Hi @Kevin Kho, caching is not working as I expect it with an S3Result. I’ve included a simple example below. Please can you see if anything is wrong or if I’ve misunderstood. Thank you. The result is written to S3. But each time I execute this script I get

Can't use cache because it is now invalid

and the task is run again. Environment variables

export PREFECT__LOGGING__LEVEL=DEBUG
export PREFECT__FLOWS__CHECKPOINTING=true

import datetime

from prefect import Flow, Parameter, context, task
from prefect.engine.cache_validators import all_inputs
from prefect.engine.results import S3Result

@task(
    cache_for=datetime.timedelta(hours=1),
    result=S3Result(bucket="my-bucket"),
    checkpoint=True,
)
def cache_this(x):

    # make it clear in the log if task is being executed or skipped
    <http://context.logger.info|context.logger.info>("\n\nTask was run!!!!!!!\n")

    return f"The value {x} was cached at {datetime.datetime.now()}"


with Flow("cache test") as flow:
    x = Parameter(name="x")
    cached_value = cache_this(x)

state = flow.run(parameters={"x": 42})
print(state.result[cached_value]._result.__dict__)

Looking at the content of what is serailized to S3 it is clear why this doesn’t work when running locally. This is all that was stored.

The value 42 was cached at 2021-07-15 10:17:26.133250

Since no Flow Parameters or task inputs were cached there’s clearly no way the cache_validators would have the information required to know if this cached value is valid or not. I was hoping the duration cache might work since the S3 object does have a timestamp. I notice if I run this flow in Prefect Server the cached value is used. I imagine this is only working because Prefect is storing meta-data about my flow run in it’s database. Is this correct? Which leads me back to my first question. When working locally it appears I have to run subsequent flow runs in the same python session so that flow metadata is available in memory despite the fact I’m caching results to external storage (S3). I had assumed switching to external storage would remove this requirement. I can see why now I was having some success with

target

. Since this doesn’t rely on flow metadata in memory or in the Prefect database. But the only control over validation is what I choose to include in the object path. Maybe in future the framework could append the required flow meta-data to my seralized result so that the built-in cache_validators can be used.

Are you using

flow.run()

in production? Or is this just for testing? The thing is, I’m not sure the metadata belongs in the

Result

in general because some users don’t serialize their results. Think like writing out a DataFrame as a csv or parquet, and then that file would be opened in a downstream process.