lucian
06/29/2025, 5:01 AMMarvin
06/29/2025, 5:01 AMMarvin
06/29/2025, 5:02 AMpython
from prefect import flow, task
from prefect.tasks.core.defaults import cache_policy
@task(cache_policy=cache_policy(
key_storage="/path/to/cache/directory" # This acts as your cache prefix
))
def my_cached_task():
return "some result"
2. Using a remote storage block (like S3):
python
from prefect import flow, task
from prefect.tasks.core.defaults import cache_policy
from prefect_aws import S3Bucket
# First, ensure you have configured your S3 block
s3_block = S3Bucket.load("my-block-name")
@task(cache_policy=cache_policy(
key_storage=s3_block,
prefix="my-custom-prefix/" # Optional prefix within the storage location
))
def my_cached_task():
return "some result"
The cache key will be automatically generated based on:
- The task's inputs
- The task's code definition
- The flow run or task run ID
These components are hashed together to create a unique cache key, and the prefix/storage location you specify determines where these cache records are stored.
Remember that caching is particularly useful for:
- Expensive computations that you want to avoid repeating
- Tasks that frequently produce the same results for the same inputs
- Optimizing workflow execution time
For more details on caching configuration and best practices, you can refer to the Prefect caching documentation.