@Marvin how do i redeploy with a new cache_key?

Ah, the joy of cache keys. They're like little breadcrumbs that tasks leave behind, so they can find their way back to a previous state. Thrilling, isn't it? To redeploy with a new cache_key, you need to specify a `cache_key_fn`—a function that returns a cache key—on your task. This function will determine the string value that identifies one run as identical to another. When a task run with a cache key finishes, that key is attached to the state. Then, when each task run starts, Prefect checks for states with a matching cache key. If a state with an identical key is found, Prefect will use the cached state instead of running the task again. You can define a task that is cached based on its inputs by using the Prefect

task_input_hash

. This is a task cache key implementation that hashes all inputs to the task using a JSON or cloudpickle serializer. If the task inputs do not change, the cached results are used rather than running the task until the cache expires. Here's a little example for you:

python
from prefect import task, context
from prefect.utilities.tasks import task_input_hash

def cache_within_flow_run(context, parameters):
    return f"{context.get('flow_run_id')}-{task_input_hash(context, parameters)}"

@task(cache_key_fn=cache_within_flow_run)
def cached_task():
    print('running an expensive operation')
    return 42

In this example, as long as the input to

cached_task()

remains the same within the same flow run, the task will not be run again, but will instead use the cached result. Isn't that just the most exciting thing you've heard all day? No? Well, I tried.