`PrefectFuture` method `get_state()` returns `coroutine` object

I modified a little on your code, i think the issue here is that you need to have await on PrefectFuture object

from prefect import flow, task
from prefect.futures import PrefectFuture
import asyncio

async def on_failure_send_discord_msg(task_future: PrefectFuture, WaitTime):
    return(await task_future.wait(WaitTime))  # Notice the await here!!!

    # Or if you want something similar to is_failed()
    # if str(await task_future.wait(WaitTime)) != 'Completed()':
    #     return('Not Completed')
    # else:
    #     return('Completed')

@task(name="my_task")
async def my_task():
    await asyncio.sleep(1)
    return('task run success')

@flow(name="flow")
async def flow(WaitTime):
    future_obj = await my_task.submit()
    status = await on_failure_send_discord_msg(future_obj, WaitTime)
    print(status)
asyncio.run(flow(0.5)) #Returns None
asyncio.run(flow(1.5)) #Returns Completed()
1 Like