Hello Team am new to prefect2 Am working on migrating flow f

Question

Hello Team am new to prefect2 Am working on migrating flow from prefect1 to prefect2

Mahesh · Answer

In prefect1 I did for looping dynamic tasks like below ``` task task run name= val def shell exec val with prefect context task run name=val ShellTask helper script= cd opt prefect shell= bash return all=True log stderr=True log stdout=True stream output=True command= venv prefect env bin python sql run py format val run with Flow flowname as flow sql list=get values tasks = shell exec val for val in sql list for i in range 1 len tasks tasks i set upstream tasks i 1 ``` I am trying same flow with prefect2 1 I need to get task name at run time 2 I have to for loop with dependency ``` flow name= exec def exec list1= run1 sql run2 sql run3 sql tasks = shell exec val for val in list1 for i in range 1 len tasks tasks i wait for tasks i 1 ``` 3 I have used `wait for` instead of `up stream` and stucked with below error ```08 57 59 052 | ERROR | Flow run vivacious hawk Finished in state Failed Flow run encountered an exception AttributeError coroutine object has no attribute wait for n Traceback most recent call last File flows test py line 31 in lt module gt exec File venv prefect2 lib python3 8 site packages prefect flows py line 442 in call return enter flow run engine from flow call File venv prefect2 lib python3 8 site packages prefect engine py line 157 in enter flow run engine from flow call return anyio run begin run File venv prefect2 lib python3 8 site packages anyio core eventloop py line 70 in run return asynclib run func args backend options File venv prefect2 lib python3 8 site packages anyio backends asyncio py line 292 in run return native run wrapper debug=debug File usr lib python3 8 asyncio runners py line 44 in run return loop run until complete main File usr lib python3 8 asyncio base events py line 616 in run until complete return future result File venv prefect2 lib python3 8 site packages anyio backends asyncio py line 287 in wrapper return await func args File venv prefect2 lib python3 8 site packages prefect client utilities py line 47 in with injected client return await fn args kwargs File venv prefect2 lib python3 8 site packages prefect engine py line 237 in create then begin flow run return await state result fetch=True File venv prefect2 lib python3 8 site packages prefect states py line 74 in get state result raise await get state exception state File venv prefect2 lib python3 8 site packages prefect engine py line 610 in orchestrate flow run result = await run sync flow call File venv prefect2 lib python3 8 site packages prefect utilities asyncutils py line 68 in run sync in worker thread return await anyio to thread run sync call cancellable=True File venv prefect2 lib python3 8 site packages anyio to thread py line 31 in run sync return await get asynclib run sync in worker thread File venv prefect2 lib python3 8 site packages anyio backends asyncio py line 937 in run sync in worker thread return await future File venv prefect2 lib python3 8 site packages anyio backends asyncio py line 867 in run result = context run func args File flows test py line 26 in exec tasks i wait for tasks i 1 AttributeError coroutine object has no attribute wait for ``` Please let me know what is missing here

Zanie · Answer

Hi With ```import prefect from prefect import flow task get run logger task def hi msg get run logger info msg return 1 flow name= exec def exec list1= run1 sql run2 sql run3 sql for val in list1 hi val exec ```

Zanie · Answer

Your tasks will run sequentially

Zanie · Answer

Here s how you d use `wait for`

Zanie · Answer

``` from prefect import flow task get run logger task def hi msg get run logger info msg return 1 flow name= exec def exec list1= run1 sql run2 sql run3 sql last future = None for val in list1 last future = hi submit val wait for= last future exec ```

Mahesh · Answer

< Zanie> Thanks for the quick reply I tried this solution with this am not getting my shell task s logs and also with this approach how can i pass task name on task run time

Zanie · Answer

How are you defining your task

Zanie · Answer

`hi val ` would become `hi with options name= my dynamic name val `

Mahesh · Answer

Below is my code ```import prefect from prefect import client flow task get run logger from prefect shell import shell run command task def run sql name return show output shell run command fn command= venv prefect2 bin python run sql py o format name return all=True helper command= cd venv prefect2 deployments flow name= exec def exec list1= run1 sql run2 sql run3 sql last future = None for val in list1 last future = run sql submit val wait for= last future if name == main exec ```

Mahesh · Answer

order should be run1 sql gt run2 sql gt run3 sql with dependency if run2 sql failed then flow should fail and run3 sql never run

Mahesh · Answer

i want to give my task name as `val` so that each task will get it name from `val`

Mahesh · Answer

And i should get logs from each tasks also ```def show output lines logger = get run logger return <http logger info|logger info> lines ``` I tried with show output run sql but it is not showing logs

Khuyen Tran · Answer

Where do you put `show output` in your flow