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.Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.
Powered by