Romain Lefebvre
07/30/2025, 10:13 AMMarvin
07/30/2025, 10:13 AMMarvin
07/30/2025, 10:14 AMpython
from prefect import task
from prefect.tasks import NO_CACHE
@task(cache_policy=NO_CACHE)
def check_computation_status():
# Your API call code here
return status
This solution is perfect for your use case because:
1. Setting cache_policy=NO_CACHE
ensures the task runs every time it's called
2. It will allow your polling mechanism to actually check the API status every 5 seconds instead of returning cached results
If you're currently using a task without any decorator parameters, just add the cache_policy=NO_CACHE
parameter to your existing decorator.
For reference, here's a complete example of how your polling logic might look:
python
from prefect import task, flow
from prefect.tasks import NO_CACHE
import time
@task(cache_policy=NO_CACHE)
def check_computation_status():
# Your API call here
response = api.get_status() # Replace with your actual API call
return response.status
@flow
def monitor_computation():
while True:
status = check_computation_status()
if status == "succeeded":
break
time.sleep(5)
return "Computation completed successfully"
This will ensure that each status check actually calls the API instead of returning cached results.