Returning Manual State from a Task I want to run a scheduled

Question

Returning Manual State from a Task I want to run a scheduled flow that deals with an intermittent service and I want to handle the exceptions gracefully The common idiom seems to be raising `ValueError` to trigger a failure and thats how to manage the state of subsequent tasks or flows Here s the thing I don t want to see a stack trace unless it is some new failure that I need to troubleshoot For the expected exceptions I want to post an error the log and proceed with my flow according to logic I design I m having trouble returning a manual state of Cancelled concrete example and logs posted in thread Can anyone guide me toward causing the entire flow to be cancelled without raising an exception

merlin · Answer

This task will test if a network connection is available and is meant to halt the flow until the task gives up after X attempts I return a manual state of `Cancelled message = No connection available ` Unfortunately in the logs you can see a ValueError is raised by the `get state exception` call What I want is for the whole flow to be cancelled

merlin · Answer

```import requests from requests exceptions import ConnectionError from prefect import task flow get run logger from prefect orion schemas states import Cancelled task def wait for service X logger = get run logger <http logger info|logger info> Try connect to service for i in range 1 3600 try if requests get SERVICE URL ok return Service Connected except ConnectionError as ex latest exception = ex logger error f latest exception return Cancelled message=f Service connection failed after i attempts task def the rest of the tasks return This task represents all next tasks that shouldn t run flow def test flow state = wait for service X return state=True print f result state the rest of the tasks wait for= state test flow ```

merlin · Answer

And the resulting logs ```poetry run python src tasks wait service py 20 34 51 946 | INFO | prefect engine Created flow run cocky gibbon for flow test flow 20 34 52 009 | INFO | Flow run cocky gibbon Created task run wait for service X 828f1dbb 0 for task wait for service X 20 34 52 009 | INFO | Flow run cocky gibbon Executing wait for service X 828f1dbb 0 immediately 20 34 52 029 | INFO | Task run wait for service X 828f1dbb 0 Try connect to trino server 20 34 58 430 | INFO | Task run wait for service X 828f1dbb 0 Service connection failed after 2 attempts 20 34 58 432 | ERROR | Task run wait for service X 828f1dbb 0 HTTPSConnectionPool host= port=443 Max retries exceeded with url Caused by ConnectTimeoutError lt urllib3 connection HTTPSConnection object at 0x105b9b430 gt Connection to timed out connect timeout=1 20 34 58 432 | ERROR | Task run wait for service X 828f1dbb 0 response None request lt PreparedRequest GET gt 20 34 58 463 | ERROR | Task run wait for service X 828f1dbb 0 Finished in state Cancelled Service connection failed after 2 attempts result Cancelled Service connection failed after 2 attempts 20 34 58 484 | INFO | Flow run cocky gibbon Created task run the rest of the tasks daa0f3b5 0 for task the rest of the tasks 20 34 58 484 | INFO | Flow run cocky gibbon Executing the rest of the tasks daa0f3b5 0 immediately 20 34 58 510 | INFO | Task run the rest of the tasks daa0f3b5 0 This task represents all next tasks that shouldn t run 20 34 58 521 | INFO | Task run the rest of the tasks daa0f3b5 0 Finished in state Completed 20 34 58 534 | ERROR | Flow run cocky gibbon Finished in state Failed 0 2 states are not final Traceback most recent call last File HOME zillpo bssgeb src tasks wait service py line 39 in lt module gt test flow File HOME Library Caches pypoetry virtualenvs bssgeb 4hu7XBP5 py3 10 lib python3 10 site packages prefect flows py line 442 in call return enter flow run engine from flow call File HOME Library Caches pypoetry virtualenvs bssgeb 4hu7XBP5 py3 10 lib python3 10 site packages prefect engine py line 157 in enter flow run engine from flow call return anyio run begin run File HOME Library Caches pypoetry virtualenvs bssgeb 4hu7XBP5 py3 10 lib python3 10 site packages anyio core eventloop py line 70 in run return asynclib run func args backend options File HOME Library Caches pypoetry virtualenvs bssgeb 4hu7XBP5 py3 10 lib python3 10 site packages anyio backends asyncio py line 292 in run return native run wrapper debug=debug File opt homebrew Cellar python 3 10 3 10 8 Frameworks Python framework Versions 3 10 lib python3 10 asyncio runners py line 44 in run return loop run until complete main File opt homebrew Cellar python 3 10 3 10 8 Frameworks Python framework Versions 3 10 lib python3 10 asyncio base events py line 649 in run until complete return future result File HOME Library Caches pypoetry virtualenvs bssgeb 4hu7XBP5 py3 10 lib python3 10 site packages anyio backends asyncio py line 287 in wrapper return await func args File HOME Library Caches pypoetry virtualenvs bssgeb 4hu7XBP5 py3 10 lib python3 10 site packages prefect client utilities py line 47 in with injected client return await fn args kwargs File HOME Library Caches pypoetry virtualenvs bssgeb 4hu7XBP5 py3 10 lib python3 10 site packages prefect engine py line 237 in create then begin flow run return await state result fetch=True File HOME Library Caches pypoetry virtualenvs bssgeb 4hu7XBP5 py3 10 lib python3 10 site packages prefect states py line 74 in get state result raise await get state exception state File HOME Library Caches pypoetry virtualenvs bssgeb 4hu7XBP5 py3 10 lib python3 10 site packages prefect states py line 318 in get state exception raise ValueError ValueError Failed state result was an iterable of states but none were failed ```

Khuyen Tran · Answer

It seems like what you need is a <https discourse prefect io t how can i set task level timeouts in prefect 2 601 2|task level timeout> Is that correct

Mason Menges · Answer

Hey < merlin> It looks Cancelled isn t being treated as a terminal state in your example which sounds like a bug would you be will to open an issue here and include your example for what it s worth running the same code returning a Completed Failed Crashed state behaves as I d expect it too

merlin · Answer

I will open an issue thanks for the reply It seems to be expected behavior though According to the aggregation rules for states here in the documentation below IMO those rules about combinations of states deserve more attention and probably a way for users to set a simple set of rules For example if a task get Cancelled and it is a dependency then the Cancelled state should roll up the chain and Cancel the whole flow depending on the remaining dependencies <https docs prefect io api ref prefect states prefect states get state exception|States in doc for get state exception> ```If not given a FAILED or CRASHED state this raise a value error If the state result is a state its exception will be returned If the state result is an iterable of states the exception of the first failure will be returned If the state result is a string a wrapper exception will be returned with the string as the message If the state result is null a wrapper exception will be returned with the state message attached If the state result is not of a known type a TypeError will be returned When a wrapper exception is returned the type will be FailedRun if the state type is FAILED or a CrashedRun if the state type is CRASHED ```

merlin · Answer

Yeah the more I think about it this rule about raising an error if not FAILED CRASHED needs a look At a minimum CANCELLED should be treated the same

merlin · Answer

gt Khuyen It seems like what you need is a task level timeout Is that correct Actually the task is going to retry a bunch of times at 5 min intervals Now that I know passing `Cancelled` back to the flow wont help I m returning `True False` which will guide the logical process in the calling flow ``` flow def test flow if VPN connected the rest of the tasks else return Cancelled message= VPN connection failed cancelling flow ``` So this works great the flow gets cancelled and shows up in the UI as a grey flow run to signify an expected result calling a flaky service The next day s job run will catch everything up so there is no need to troubleshoot a failure

merlin · Answer

< Mason Menges> There is an open issue with the same thing returning Cancelled from within a task raises TypeError

Zanie · Answer

We will tackle this when we add first class cancellation in the next month These rules do not handle cancellation right now because we do not have a cancellation mechanic yet You can use a named failed state for now if you would like e g `Failed name= Skipped message= This shouldn t run for reason `

merlin · Answer

Wild ok this is great BTW i love using Prefect

Zanie · Answer

You can even name it Cancelled if you want wink the state type is the primary determinant for orchestration behavior and names can be used to extend states to display information relevant to your organization

Zanie · Answer

Thanks We love hearing that slightly smiling face