Camunda External Task Client for Python 3
Project description
camunda-external-task-client-python3
This repository contains Camunda External Task Client written in Python3.
Implement your BPMN Service Task in Python3.
Python >= 3.7 is required
Installing
Add following line to requirements.txt
of your Python project.
git+https://github.com/trustfactors/camunda-external-task-client-python3.git/#egg=camunda-external-task-client-python3
Or use pip to install as shown below:
pip install camunda-external-task-client-python3
Usage
- Make sure to have Camunda running.
- Create a simple process model with an External Service Task and define the topic as 'topicName'.
- Deploy the process to the Camunda BPM engine.
- In your Python code:
import time
from camunda.external_task.external_task import ExternalTask, TaskResult
from camunda.external_task.external_task_worker import ExternalTaskWorker
# configuration for the Client
default_config = {
"maxTasks": 1,
"lockDuration": 10000,
"asyncResponseTimeout": 5000,
"retries": 3,
"retryTimeout": 5000,
"sleepSeconds": 30
}
def handle_task(task: ExternalTask) -> TaskResult:
"""
This task handler you need to implement with your business logic.
After completion of business logic call either task.complete() or task.failure() or task.bpmn_error()
to report status of task to Camunda
"""
# add your business logic here
# ...
# mark task either complete/failure/bpmnError based on outcome of your business logic
failure, bpmn_error = random_true(), random_true() # this code simulate random failure
if failure:
# this marks task as failed in Camunda
return task.failure(error_message="task failed", error_details="failed task details",
max_retries=3, retry_timeout=5000)
elif bpmn_error:
return task.bpmn_error(error_cide="BPMN_ERROR_CODE")
# pass any output variables you may want to send to Camunda as dictionary to complete()
return task.complete({"var1": 1, "var2": "value"})
def random_true():
current_milli_time = int(round(time.time() * 1000))
return current_milli_time % 2 == 0
if __name__ == '__main__':
ExternalTaskWorker(config=default_config).subscribe("topicName", handle_task)
About External Tasks
External Tasks are service tasks whose execution differs particularly from the execution of other service tasks (e.g. Human Tasks). The execution works in a way that units of work are polled from the engine before being completed.
camunda-external-task-client-python allows you to create easily such client in Python3.
Features
Fetch and Lock
ExternalTaskWorker().subscribe("topicName", handle_task)
starts long polling of the Camunda engine for external tasks.
- Polling tasks from the engine works by performing a fetch & lock operation of tasks that have subscriptions. It then calls the handler function passed to
subscribe()
function. i.e.handle_task
in above example. - Long Polling is done periodically based on the
asyncResponseTimeout
configuration. Read more about Long Polling.
Complete
from camunda.external_task.external_task import ExternalTask, TaskResult
from camunda.external_task.external_task_worker import ExternalTaskWorker
def handle_task(task: ExternalTask) -> TaskResult:
# add your business logic here
# Complete the task
# pass any output variables you may want to send to Camunda as dictionary to complete()
return task.complete({"var1": 1, "var2": "value"})
ExternalTaskWorker().subscribe("topicName", handle_task)
Handle Failure
from camunda.external_task.external_task import ExternalTask, TaskResult
from camunda.external_task.external_task_worker import ExternalTaskWorker
def handle_task(task: ExternalTask) -> TaskResult:
# add your business logic here
# Handle task Failure
return task.failure(error_message="task failed", error_details="failed task details",
max_retries=3, retry_timeout=5000)
# This client/worker uses max_retries if no retries are previously set in the task
# if retries are previously set then it just decrements that count by one before reporting failure to Camunda
# when retries are zero, Camunda creates an incident which then manually needs to be looked into on Camunda Cockpit
ExternalTaskWorker().subscribe("topicName", handle_task)
Handle BPMN Error
from camunda.external_task.external_task import ExternalTask, TaskResult
from camunda.external_task.external_task_worker import ExternalTaskWorker
def handle_task(task: ExternalTask) -> TaskResult:
# add your business logic here
# Handle a BPMN Failure
return task.bpmn_error(error_code="BPMN_ERROR")
ExternalTaskWorker().subscribe("topicName", handle_task)
Access Process Variables
from camunda.external_task.external_task import ExternalTask, TaskResult
from camunda.external_task.external_task_worker import ExternalTaskWorker
def handle_task(task: ExternalTask) -> TaskResult:
# add your business logic here
# get the process variable 'score'
score = task.get_variable("score")
if int(score) >= 100:
return task.complete(...)
else:
return task.failure(...)
ExternalTaskWorker().subscribe("topicName", handle_task)
License
The source files in this repository are made available under the Apache License Version 2.0.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for camunda-external-task-client-python3-2.0.0.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | dff4c5902a05eb3767d1d993c027c560c65447ccde8d240773111204befc7081 |
|
MD5 | f75ce12161ea94c8a473ee04eb4cc09f |
|
BLAKE2b-256 | ad896e7e7901bb855b4fbf53bd252bc8190d3662fe233968fb0717452ccbbcf7 |
Hashes for camunda_external_task_client_python3-2.0.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | f9580f02fff20f3a4ce08e6537dabd46932bc5c46ffe21cfea94592533d21e8a |
|
MD5 | 1bece35d27d68a72b444207671942f66 |
|
BLAKE2b-256 | 34c6619ed48856e322463053b369e940939a49caa3239b468239ec57a1abc12f |