Added async workflow client implementation, leveraging new durabletask.aio.client implementation

[passuied]

Description

Added async workflow client implementation, leveraging new durabletask.aio.client implementation

Issue reference

#834

Checklist

Please make sure you've completed the relevant tasks for this PR, out of the following list:

• Code compiles correctly
• Created/updated tests
• Extended the documentation - Pending Full async support for WorkflowRuntime

[codecov]

Codecov Report

❌ Patch coverage is 97.26027% with 4 lines in your changes missing coverage. Please review.
✅ Project coverage is 87.54%. Comparing base (bffb749) to head (13f66a3).
⚠️ Report is 51 commits behind head on main.

Files with missing lines	Patch %	Lines
...flow/dapr/ext/workflow/aio/dapr_workflow_client.py	94.91%	3 Missing ⚠️
...apr-ext-workflow/tests/test_workflow_client_aio.py	98.82%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #861      +/-   ##
==========================================
+ Coverage   86.63%   87.54%   +0.90%     
==========================================
  Files          84       99      +15     
  Lines        4473     6461    +1988     
==========================================
+ Hits         3875     5656    +1781     
- Misses        598      805     +207     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[acroca]

@passuied I think this other PR covers this same thing, and is a bit more complete because it also have async context, but it's WIP as it has lots of lint changes that are now applied to main. I'll ask for a cleanup. Can you have a look at that PR and confirm it'd work for you after the cleaning? focus on the ext/dapr-ext-workflow changes :)

[filintod]

@acroca actually this is a bit different, as this tackles the client initiating actions against a workflow(schedule workflow, signal events), instead of the internals of the workflow that my PR tackles. He did the PR in durabletask that complements this dapr/durabletask-python#17. The PRs his/mine are complementary and not overlapping

[acroca]

Looks good, but I'd follow the same pattern as the DaprClient and not have a suffix Async for the async version.

[acroca]

@passuied Can you add an example to examples/workflow that uses this new client? Maybe something like https://github.com/dapr/python-sdk/blob/main/examples/workflow/simple.py but async.
Sorry for adding new items to the review, but I didn't think about it when I reviewed yesterday 🤦

[acroca]

LGTM, but tests are failing, and copyright headers need to be fixed :)

[acroca]

Can't we have an example using this client? I know you removed it because of the workflow_runtime thing that was discussed in a thread, but if I got it right, that part was not necessary right? I think we could just have an example to cover what has been added here, can't we?

[passuied]

Can't we have an example using this client? I know you removed it because of the workflow_runtime thing that was discussed in a thread, but if I got it right, that part was not necessary right? I think we could just have an example to cover what has been added here, can't we?

@acroca It's just hard to provide a meaningful example without other parts of the workflow to be async... It's kinda moot...
We have implemented the hack on our end by overriding the activity annotation that allows us to run async activities (which is what I have tried to recreate in the repo) but if we don't want to merge that part, an example doesn't make sense until the fully async WorkflowRuntime is ready...

[filintod]

@passuied it does make sense to break it, think of the runtime as the backend and the client and the front-end/api-gateway. you can have an async client that just return workflow instance ID (to check status) and a backend that process the work items (workflow runtime).

You should be able to convert any example to asyncio client with:

For example the fan out/fan in:

async def main():
    # async client
    import dapr.ext.workflow.aio as aio_client

    wf_client = aio_client.DaprWorkflowClient()
    instance_id = await wf_client.schedule_new_workflow(workflow=batch_processing_workflow, input=10)

    # ....
    # other asyncio functions calls...
   
    print(f'Workflow started. Instance ID: {instance_id}')
    return  await wf_client.wait_for_workflow_completion(instance_id)

if __name__ == '__main__':
    wfr.start()  # synchronous backend engine
    time.sleep(10)  # wait for workflow backend runtime to start

    try:
        wf_state = asyncio.run(main())
        print(f'Workflow completed! Status: {wf_state}')

    except Exception as e:
        import traceback
        traceback.print_exc()
        exit(1)
    finally:
        wfr.shutdown()

Inside the activity if you need to run asyncio calls just do something similar for now (until we get the other work in durabletask in)

[passuied]

Thanks @filintod that was helpful.

@acroca I have pushed a change with the simple.py example adapted for using the async client. Works very well!

[filintod]

add a try/finally to always execute wfr.shudown() on any issues, or the process might hang if any issue happens

[passuied]

done!

[acroca]

LGTM!

The linter in the CI doesn't say what's wrong (we should change it to ruff format --check), here's what I got locally:

❯ ruff format --check
Would reformat: examples/workflow/simple_aio_client.py
1 file would be reformatted, 229 files already formatted

Also, the examples don't seem to install the workflow extension from the local directory, can you fix that in this PR please? I think we should install all the extensions locally before running validations.

Thanks!