pyzeebe_worker.py

import asyncio
import os

from dotenv import load_dotenv
from pyzeebe import Job, JobController, ZeebeWorker, create_camunda_cloud_channel

load_dotenv()

client_id = os.environ["CAMUNDA_CLIENT_ID"]
client_secret = os.environ["CAMUNDA_CLIENT_SECRET"]
cluster_id = os.environ["CAMUNDA_CLUSTER_ID"]

channel = create_camunda_cloud_channel(client_id, client_secret, cluster_id)
worker = ZeebeWorker(channel)


async def on_error(exception: Exception, job: Job, job_controller: JobController):
    """on_error will be called when the task fails"""
    print(exception)
    await job_controller.set_error_status(
        f"Failed to handle job {job}. Error: {str(exception)}"
    )


@worker.task(task_type="credit-deduction", exception_handler=on_error)
async def credit_deduction_task() -> dict:
    return {"output": {"task_type": "credit-deduction"}}


@worker.task(task_type="credit-card-charging", exception_handler=on_error)
async def credit_card_charging_task() -> dict:
    return {"output": {"task_type": "credit-card-charging"}}


loop = asyncio.get_event_loop()
loop.run_until_complete(worker.work())