How to connect to zeebe cluster and define a service task in python?

Camunda 8 Topics Discussion & Questions
1

I am new to camunda and using self-managed camunda 8.5. I want to try out service task and do this using a python script. I got to know about the pyzeebe library. How can I connect to the Zeebe cluster and try executing service tasks. If you are able to share some code snippets, it would be of great help.

2

Hello @oops_aman,
did you look into this Github Repository?

3 
import asyncio
from pyzeebe import ZeebeWorker, Job, create_insecure_channel
from typing import Dict

grpc_channel = create_insecure_channel(
    hostname=URL,
    port=443,
)

worker = ZeebeWorker(grpc_channel)

@worker.task(task_type="test")
def example_task() -> Dict:
    return {"output": f"Hello world, test!"}

if __name__ == '__main__':
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(worker.work())

For this code, I am getting the below error
RuntimeError: Task <Task pending name=‘Task-2’ coro=<JobPoller.poll() running at C:\Users\test\Lib\site-packages\pyzeebe\worker\job_poller.py:39> cb=[gather.._done_callback() at C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.12_3.12.1520.0_x64__qbz5n2kfra8p0\Lib\asyncio\tasks.py:767]> got Future <Task pending name=‘Task-4’ coro=<UnaryStreamCall._send_unary_request() running at C:\Users\test\Lib\site-packages\grpc\aio_call.py:643>> attached to a different loop
Task was destroyed but it is pending!
task: <Task pending name=‘Task-4’ coro=<UnaryStreamCall._send_unary_request() running at C:\Users\test\Lib\site-packages\grpc\aio_call.py:643>>
sys:1: RuntimeWarning: coroutine ‘UnaryStreamCall._send_unary_request’ was never awaited

4

I just created a Worker with python.
Can you work with that example?


import asyncio

from pyzeebe import ZeebeWorker, create_insecure_channel

channel = create_insecure_channel()
worker = ZeebeWorker(channel)

async def main():
    channel = create_insecure_channel()
    worker = ZeebeWorker(channel)


    @worker.task(task_type="python_task")
    async def my_task(x: int):
        return {"y": x + 1}

    await worker.work()

asyncio.run(main())

There is a workaround to your problem.
Worker Quickstart — pyzeebe 3.0.4 documentation

5

As you shared, I tried the same. It is giving me the following logs:

<pyzeebe.worker.worker.ZeebeWorker object at 0x000002143DEE0BC0>
Failed to activate jobs from the gateway. Exception: ZeebeGatewayUnavailableError(). Retrying in 5 seconds…
Failed to activate jobs from the gateway. Exception: ZeebeGatewayUnavailableError(). Retrying in 5 seconds…
Failed to activate jobs from the gateway. Exception: ZeebeGatewayUnavailableError(). Retrying in 5 seconds…
Failed to activate jobs from the gateway. Exception: ZeebeGatewayUnavailableError(). Retrying in 5 seconds…
Failed to activate jobs from the gateway. Exception: ZeebeGatewayUnavailableError(). Retrying in 5 seconds…
Failed to activate jobs from the gateway. Exception: ZeebeGatewayUnavailableError(). Retrying in 5 seconds…
Failed to activate jobs from the gateway. Exception: ZeebeGatewayUnavailableError(). Retrying in 5 seconds…
Failed to activate jobs from the gateway. Exception: ZeebeGatewayUnavailableError(). Retrying in 5 seconds…
Failed to activate jobs from the gateway. Exception: ZeebeGatewayUnavailableError(). Retrying in 5 seconds…
Failed to activate jobs from the gateway. Exception: ZeebeGatewayUnavailableError(). Retrying in 5 seconds…

grpc._cython.cygrpc.UsageError: Channel is closed.

PS: I am using a self-managed camunda-8.5, and I have the hostname, client id, client secret, oauth token url.

6

Well, seems like the Async Loop problem is gone.
I tried it with a local engine and no verification.

Do you have another client like Java or Node to check the auth configuration?

7

I tried using Node.js configuration. Getting the below error:
Error: getaddrinfo ENOTFOUND ‘HOSTNAME’