KG
04/03/2024, 3:04 PMMarvin
04/03/2024, 3:05 PMpython
import asyncio
from prefect import flow
async def some_async_function():
await asyncio.sleep(1)
return "I'm an async function"
@flow(log_prints=True) # send `print` statements as INFO logs
async def my_async_workflow():
print("I can call any python code here, including prefect tasks or flows")
await some_async_function()
if __name__ == "__main__":
asyncio.run(my_async_workflow()) # run it like a normal async python function
You might want to look into Prefect's state handlers or the section about managing flow runs for more information.
Unfortunately, I wasn’t able to locate a specific code example for your case via Prefect’s documentation or resources. If easier solutions don't emerge, let me know and we can endure the tedious work of searching GitHub issues for similar cases or concoct a solution from scratch. Heaven forbid we must resort to optimism and creativity, though I suppose even in a universe as unyieldingly vast as ours, such atrocities might occur.KG
04/03/2024, 3:17 PMMarvin
04/03/2024, 3:17 PMKG
04/03/2024, 4:09 PMMarvin
04/03/2024, 4:10 PMon_cancellation
argument of a task would be executed when a task fails due to a CancelledError
. However, in line with the behavior of on_cancellation
, it's designed to trigger when a task is canceled intentionally, rather than when it fails due to an exception.
Typically, the on_failure
hook is used to execute a function when a task encounters an exception and fails, whereas the on_cancellation
hook is intended to be used when a task run is canceled (intentionally terminated before completion).
Based on the traceback you provided, the error indicates that the task was canceled (via CancelledError
). This could happen if you have logic to cancel the task under certain conditions or due to an infrastructure issue causing the cancellation. Therefore, if you have
set the on_cancellation
hook, it should be executed when the task is canceled as indicated by the error you've encountered.
It is worth noting that Prefect's documentation mentions that they are working on improved cancellation handling and you can enable this experimental feature using the PREFECT_EXPERIMENTAL_ENABLE_ENHANCED_CANCELLATION
configuration on your worker or agents.
For further specifics, it might be helpful to examine the Prefect documentation related to task lifecycle hooks for understanding how on_cancellation
behaves, as well as the section on enhanced cancellation if you face issues with the current cancellation experience.