Indeed The `LocalResultHandler` do

Question

Indeed < Andrew Schechtman Rook> The `LocalResultHandler` does store information without regard to the task in question however the state object as a result of each task run does keep track of the path written to You can then persist the task to file mapping separately any way you d like You could persist all of the state returned by `flow run ` or persist as state is available with a state handler An example of caching this for the whole flow using the `LocalResultHandler` + some state manipulation ``` with prefect Flow example result handler=LocalResultHandler dir= prefect as flow x = e y = y x z y flow state = flow run for task task state in state result items task name = slugify slugify task name filepath = task state result safe value value with open os path join prefect task name wb as fh fh write filepath ``` In this way in the ` prefect` directory you would have the a result file for each task as well as a tracking file for each task named by the task name ```tree prefect prefect prefect 2g3sboyi prefect ixeqv82p prefect km5ufh9y x y z``` Note I used slugify to encode the task name but you have several options here A good option if you have off task names is to set the `task slug` and use this to key off of If you wanted to do the same thing but in a task state handler ```def handler task old new if new is successful task name = slugify slugify task name filepath = new result safe value value with open os path join prefect task name w as fh fh write filepath ```

Andrew Schechtman-Rook · Answer

Thanks < Alex Goodman> I m trying to run your example and I m getting a `TypeError a bytes like object is required not NoResultType ` on the `write` command This is what I have I had to make some slight modifications to deal with a couple of other errors ```import prefect from prefect engine result handlers import LocalResultHandler prefect task def e return 5 prefect task def y input val return input val 3 with prefect Flow example result handler=LocalResultHandler dir= prefect as flow x = e y = y x flow state = flow run for task task state in flow state result items task name = slugify slugify task name filepath = task state result safe value value with open os path join prefect task name wb as fh fh write filepath ```

Alex Goodman · Answer

ahh my mistake change ` wb ` in the `open` clause to ` w `

Andrew Schechtman-Rook · Answer

`TypeError write argument must be str not NoResultType`

Alex Goodman · Answer

gotcha in my example all tasks returned results where in your flow there are some tasks with no returned results Before writing anything we ll have to take account for this

Andrew Schechtman-Rook · Answer

which of my tasks don t return a result both of my tasks have a `return`

Alex Goodman · Answer

let me run it locally to see what s up

Andrew Schechtman-Rook · Answer

thanks

Alex Goodman · Answer

Interesting I wasn t able to reproduce ```~ code in script cli poc venv374 python test py 2019 12 23 21 49 46 282 INFO prefect FlowRunner | Beginning Flow run for example lt snip gt 2019 12 23 21 49 46 371 DEBUG prefect FlowRunner | Flow example Handling state change from Running to Success ~ code in script cli poc venv374 ls al prefect total 32 drwxr xr x 6 user user 192 Dec 23 16 49 drwxr xr x 16 user user 512 Dec 23 16 48 rw r r 1 user user 56 Dec 23 16 49 e rw 1 user user 5 Dec 23 16 49 prefect qcf3c39e rw 1 user user 5 Dec 23 16 49 prefect wd24pmtb rw r r 1 user user 56 Dec 23 16 49 y ~ code in script cli poc venv374 ``` The only thing I changed was the `wb` i mentioned and a few imported `os` and `slugify`

Andrew Schechtman-Rook · Answer

so that s interesting

Andrew Schechtman-Rook · Answer

oh looks like I m on an older version of prefect maybe that s my problem

Alex Goodman · Answer

which version I can try to reproduce it

Andrew Schechtman-Rook · Answer

0 7 3

Andrew Schechtman-Rook · Answer

I m trying to pull the latest version from `conda forge` now

Alex Goodman · Answer

yup that s the problem I can update the example

Andrew Schechtman-Rook · Answer

don t worry about it I d rather update slightly smiling face

Alex Goodman · Answer

all good updating it usually best however if you find yourself unable to update a hold over would be ```for task task state in flow state result items task name = slugify slugify task name task state result store safe value filepath = task state result safe value value with open os path join prefect task name w as fh fh write filepath ```

Andrew Schechtman-Rook · Answer

would you be able to paste in your working version of the test script even after updating mine s still not working and I m starting to strongly suspect the problem is between my ears

Alex Goodman · Answer

nope looks it was my bad I may have been depending on features from master this will do you the trick ```import os import prefect from prefect engine result handlers import LocalResultHandler import slugify prefect task def e return 5 prefect task def y input val return input val 3 with prefect Flow example result handler=LocalResultHandler dir= prefect as flow x = e y = y x flow state = flow run for task task state in flow state result items task name = slugify slugify task name task state result store safe value filepath = task state result safe value value with open os path join prefect task name w as fh fh write filepath ```

Andrew Schechtman-Rook · Answer

yep that s working now

Andrew Schechtman-Rook · Answer

thanks this at least gives me something to start playing around with

Andrew Schechtman-Rook · Answer

In case anyone in the future is looking for this functionality here s what I got working ```import datetime import os import prefect from prefect engine result handlers import LocalResultHandler prefect task def generate data return 5 prefect task cache for=datetime timedelta minutes=1 def manipulate data 1 input data return input data 3 prefect task def manipulate data 2 input data print input data 2 return input data 2 def cache to result handler task old state new state if old state is successful and new state is cached old state result store safe value if old state is pending and new state is cached new state result = new state result safe value to result return new state with prefect Flow example result handler=LocalResultHandler dir= prefect as flow data 1 = generate data data 2 = manipulate data 1 data 1 data 3 = manipulate data 2 data 2 flow state = flow run task runner state handlers= cache to result handler for task name caches in prefect context caches items for cache in caches cache result value = None flow state 2 = flow run task runner state handlers= cache to result handler ``` First there s a state handler that does 2 things for state changes from successful to cached it invokes the `safe result` method to trigger the `result handler` s `write` method for state changes from pending to cached it inflates the safe result by invoking the `to result` method and uses that to replace the cached result Second at the end of the first flow a quick loop replaces all cached result values with `None` so that the results won t take up any space if persisted to disk It seems to get the job done although I m not super psyched about it because manually editing the cache post flow seems really kludgey