@Marvin I am defining the cache policy for a task that is composed of a function in which body I am calling a couple of other utility functions I wrote. I want to use the cache in case the task has been called on the same inputs and its source code and the source of the functions called by it has not been modified. I set

@task(cache_policy=INPUTS+TASK_SOURCE)

, but this way if the source of any of the functions called inside the task changed the cached result is used instead of rerunning the task. How can I solve this?

To ensure that your task's cache policy considers changes in both the task's source code and the source code of any utility functions it calls, you can use a custom cache key function. Prefect allows for custom cache policy logic through cache key functions, which can be tailored to your specific needs. Here's a concise approach to achieve this: 1. Define a Custom Cache Key Function: - This function should compute a unique cache key based on the task's inputs, its source code, and the source code of the utility functions it calls. 2. Use the Custom Cache Key Function in the Task: - Set your task's cache policy to use this custom function. Here's a basic example to guide you:

python
from prefect import task, TaskRunContext

def custom_cache_key(context: TaskRunContext, inputs: dict):
    # Compute a unique hash based on inputs and source codes
    task_source_hash = hash(context.task.fn.__code__.co_code)
    utility_function_hash = hash(your_utility_function.__code__.co_code)
    return (task_source_hash, utility_function_hash, hash(tuple(inputs.items())))

@task(cache_key_fn=custom_cache_key)
def your_task(input_1, input_2):
    # Call your utility functions
    result = your_utility_function(input_1) + another_utility_function(input_2)
    return result

In this example,

custom_cache_key

generates a cache key that includes hashes of the source code of the task and the utility functions, as well as the task inputs. Adjust the function to include any other utility functions you use. For further reading on cache policies and custom cache key functions, you may find the Prefect documentation on task caching helpful.