Here’s a minimal Prefect 3.x example.

python
from prefect import flow, task
from prefect.cache_policies import INPUTS
from prefect.filesystems import RemoteFileSystem

rfs = RemoteFileSystem(basepath="<s3://my-bucket/prefect-results>")

@task(
    cache_policy=INPUTS,
    persist_result=True,
    result_storage=rfs,
    result_serializer="json",
)
def add_one(x: int):
    print(f"Running task for x={x}")
    return {"x": x, "y": x + 1}

@flow
def my_flow():
    a = add_one(1)
    b = add_one(1)  # cache hit: same input
    c = add_one(2)  # new result: different input
    return a, b, c

if __name__ == "__main__":
    my_flow()

How it works -

cache_policy=INPUTS

means the cache key is derived from the task inputs. - Calling

add_one(1)

twice produces the same cache key, so the second call can reuse the persisted result. - Calling

add_one(2)

produces a different cache key, so it runs again. A couple notes: -

result_serializer="json"

requires the task result to be JSON-serializable. -

RemoteFileSystem

works with fsspec-supported backends like S3, GCS, Azure, etc., assuming the appropriate dependencies/credentials are available. If you want, I can also give you: - the same example with

S3Bucket

/

GcsBucket

- a version using

cache_key_fn

- a fully runnable local example without cloud credentials Relevant docs: - Results and result persistence - Task caching

Yes — in Prefect 3.x, the usual way to bypass reading from cache is

refresh_cache=True

. Per-call override

python
from prefect import flow, task
from prefect.cache_policies import INPUTS
from prefect.filesystems import RemoteFileSystem

rfs = RemoteFileSystem(basepath="<s3://my-bucket/prefect-results>")

@task(
    cache_policy=INPUTS,
    persist_result=True,
    result_storage=rfs,
    result_serializer="json",
)
def add_one(x: int):
    print(f"Running task for x={x}")
    return {"x": x, "y": x + 1}

@flow
def my_flow():
    a = add_one(1)  # may read from cache

    b = add_one.with_options(refresh_cache=True)(1)
    # forces re-run, overwriting/refreshing cached result for that key

    c = add_one(1)  # may now read the refreshed cached result
    return a, b, c

Other options - Always refresh for that task

python
@task(
    cache_policy=INPUTS,
    persist_result=True,
    result_storage=rfs,
    result_serializer="json",
    refresh_cache=True,
)
def add_one(x: int):
    ...

- Refresh all tasks globally Set:

bash
PREFECT_TASKS_REFRESH_CACHE=true

- Disable caching entirely Set:

bash
PREFECT_TASKS_DISABLE_CACHING=true

Rule of thumb -

refresh_cache=True

= don’t read cached result; run task again and update cache -

disable_caching=true

= no cache read and no cache write Docs: - Task caching - Settings reference If you want, I can show the smallest possible example of “normally cached, but force refresh on one call only.”