@Marvin Is there a way to define a caching function to be based on the state of another task asper it only uses the value stored in its cache if the 1st task aligns with specific conditions e.g if there are 2 tasks: A and B, where task A has inputs and is cached using the task_input_hash but task B doesn’t have any input to its function so i want to cache it based on the state of task A - specifically if task A shows that it used the cache, then task B should equally use its cache but if task A doesn’t use its cache (the inputs are different), then task B should run and not use the previously saved cache

It seems that the information retrieved from the documents is not directly related to defining a caching function based on another task's state in Prefect. I would need more specific documentation on Prefect's caching mechanism and how to implement dependent caching between tasks. Since the information is not sufficient for your query, I will search for relevant GitHub issues to see if anyone else has discussed or implemented a similar requirement. Please hold on. From the GitHub issues search, I found some related discussions: 1. Add flow-level caching: Users discussed the desire to have flow-level caching behavior, similar to task-level caching using

cache_key_fn

. While this isn't directly related to your scenario, the pattern for custom caching and dependency could be inferred from this. 2. Removing flow_run clears everything except task_run_state_cache: Although this issue is about cleanup after deletion, it does imply the use of

task_run_state_cache

, which might be relevant for your caching logic. 3. Task's `refresh_cache` not working with flow retries: This issue discusses a bug related to

refresh_cache

not working with retries. It indicates that there are mechanisms to control cache behavior. Based on these issues, it's clear that Prefect has some features to manage cache at a task level. You might be able to achieve task dependency by using custom

cache_key_fn

functions that consider the state of other tasks or potentially by manipulating the

task_run_state_cache

directly. Unfortunately, no explicit solution for your specific requirement was found in the issues, and it may require some creativity in using caching functions along with Prefect's state handling. For your scenario, you can consider this approach: - Create a custom

cache_key_fn

for task B that checks the state of task A. - When task A runs, store its state in a way that it can be accessed by the caching function of task B. - In the caching function for B, apply logic to determine whether to use the cached value or not, depending on the state of task A. Note that this will likely involve creating a custom caching strategy and orchestrating the caching logic between tasks, and it might need specific handling in your Prefect flow's code. If this approach doesn't suit you, or you need more guidance, I recommend reaching out to the Prefect community or opening a discussion on the Prefect GitHub repository.