[flytekit] Add support for async workflows

[wild-endeavor]

Background

The new version of the eager task attempts to be more pythonic in its handling of async function definitions. This means however that eager tasks, which are async functions, cannot be called within a normal workflow (@workflow). This is an okay limitation for the first draft of the new eager functionality, but we should fix it.

Goal

Not sure what the goal should look like, except that eager tasks should be callable within workflows. More thought should be given on how to do this.

Maybe we have async workflow functions

@workflow
async def wf() -> T:
    c1 = eager_1()
    n1 = await c1
    so1 = simple_task_1(a=n1)

    c2 = eager_2(a=so1)
    n2 = await asyncio.create_task(c2)
    simple_task_2(a=n2)

    # this needs to error
    # if n2 > 15:
    #     r4 = await eager_4()

    c3 = eager_3(a=so1, b=n2)
    n3, = asyncio.gather(c3)

    return n3

Or maybe we keep workflows static, something like

@workflow
def wf() -> T:
    n1 = asyncio.run(eager_1())
    so1 = simple_task_1(a=n1)

    return so1

In the latter case, how does using asyncio.run (which users will inevitably use) interfere with the loop manager in asyn.py?

cc @cosmicBboy @thomasjpfan @eapolinario

Are you sure this issue hasn't been raised already?

• Yes

Have you read the Code of Conduct?

• Yes

[mattiadevivo]

following this, I need to call an async function in my worflow and I'd be interested in understanding how to do it

[wild-endeavor]

@mattiadevivo until this issue is fixed, you won't be able to. Will wait to see what form the solution here takes but it will need to be awaited.