Asyncio ZeebeWorker support

[JonatanMartens]

Is your feature request related to a problem? Please describe.

pyzeebe currently has no support for writing async tasks. I believe that asyncio could greatly increase pyzeebe's usability and developer friendliness.

Describe the solution you'd like

In addition to expressing regular tasks like this:

@worker.task("task_type")
def task():
    ...

We should also support doing this:

@worker.task("task_type")
async def task():
    await something()

I would also suggest splitting the current ZeebeWorker class into two classes:

1. A JobPoller class who is responsible for polling the gateway for jobs.
2. The normal ZeebeWorker class (minus the job polling responsibility).

Each registered task would run two threads:

1. A thread that polls for jobs using the JobPoller
2. A thread that executes jobs using the task handler.
   • Async task handlers would be run using the await keyword. This is something I'm not entirely certain about (see below)
   • Regular task handlers would be run using asyncio.loop.run_in_executor

This would mean that something like this:

import time


@worker.task("task_type")
async def my_async_task():
    time.sleep(10)  # This statement will block the event loop

would block the event loop. We might therefore still consider spawning a thread for each job.

Zeebe-grpc

Currently we are using zeebe-grpc under the hood to communicate to the Zeebe gateway. This library currently has no support for asyncio, so we may want to implement it ourselves.

Describe alternatives you've considered

Introduce AsyncZeebeWorker, which would support async tasks. I'm not sure this is ideal, but it would make sure there are no breaking changes (which will happen anyway with pyzeebe 3.0.0).

[kbakk]

I think adding async support as a separate module would be the way to go. It would allow us to introduce this without changing the current implementation, and get some real-world usage experiences around async. We also wouldn't have to worry about breaking changes when first implementing async, as there would be nothing to break. We could then later replace the thread-based worker if it makes sense.

Wonder if it then also makes sense to specify async as extras (to get the additional packages needed), so one would install Pyzeebe with pip install pyzeebe[async].

Before using Pyzeebe at NEP, we implemented a PoC around Zeebe workers with Trio. It offers some very useful functionality with regards to task management (https://trio.readthedocs.io/en/stable/tutorial.html#okay-let-s-see-something-cool-already).

That's also one consideration that should be made with regards to which async framework to use. I know Trio has some compatibility libraries that allows you to use Trio with AsyncIO.

[Chadys]

I really like this plan to add async support. Polling for jobs is by definition an IO process, very well suited for asynchronous programming. It was clearly something I checked when I first started looking into this project and it might encourage a lot more people to use it.

@JonatanMartens I don't think any new thread should be spawned for an async task handler. It is the responsibility of the programmer not to use any blocking operation in a async function (in your example that would mean replacing time.sleep with await asyncio.sleep) and I think it is a sufficiently well known fact as soon as you get into the asynchronous pattern. Making developers able to use blocking operation in an async function without consequence is an anti-pattern imho. Beside, creating a thread for a task that is very short-lived has a performance cost and also some task may simply just not be thread-safe. If you let the possibility to have regular task handlers be run in a thread pool with asyncio.loop.run_in_executor that would be perfect and the developer will be able to make a conscious choice depending on their knowledge of whether the concerned task is long running or not.

@kbakk Concerning the async framework, why not simply use native asyncio? Most async framework I know were created to compensate on missing features in Python's early days with async. Unless I'm missing something or unless this library is planning on supporting old versions of Python (support of 3.6 might need to be dropped to be able to use asyncio, some major change were introduced with Python 3.7), asyncio is plenty sufficient and remove the need of adding an external dependency.
I don't have an opinion on the separate module for async debate, as long as it doesn't overly complicate async use. It would be absolutely great however if it all came together to be available (even in a separate module) with Zeebe 1.0.0 and pyzeebe 3.0.0.

I've just started using this project so I'm sorry if my early assumption made me misjudge something :)

[JonatanMartens]

Released in v3.0.0