[](https://github.com/camunda-community-hub/community)[](https://github.com/Camunda-Community-Hub/community/blob/main/extension-lifecycle.md#stable-)
# camunda-external-task-client-python3
This repository contains Camunda External Task Client written in Python3.
Implement your [BPMN Service Task](https://docs.camunda.org/manual/latest/user-guide/process-engine/external-tasks/) 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
```
## Running Camunda with Docker
To run the examples provided in [examples](./examples) folder, you need to have Camunda running locally or somewhere.
To run Camunda locally with Postgres DB as backend, you can use [docker-compose.yml](./docker-compose.yml) file.
```
$> docker-compose -f docker-compose.yml up
```
### Auth Basic Examples
To run the examples with Auth Basic provided in [examples/examples_auth_basic](./examples/examples_auth_basic) folder, you need to have Camunda with AuthBasic, running locally or somewhere.
To run Camunda with AuthBasic locally with Postgres DB as backend, you can use [docker-compose-auth.yml](./docker-compose-auth.yml) file.
```
$> docker-compose -f docker-compose-auth.yml up
```
## Usage
1. Make sure to have [Camunda](https://camunda.com/download/) running.
2. Create a simple process model with an External Service Task and define the topic as 'topicName'.
3. Deploy the process to the Camunda BPM engine.
4. In your Python code:
```python
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_code="BPMN_ERROR_CODE", error_message="BPMN Error occurred",
variables={"var1": "value1", "success": False})
# 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(worker_id="1", 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
### [Start process](https://docs.camunda.org/manual/latest/reference/rest/process-definition/post-start-process-instance/)
Camunda provides functionality to start a process instance for a given process definition.
To start a process instance, we can use `start_process()` from [engine_client.py](./camunda/client/engine_client.py#L24)
You can find a usage example [here](./examples/start_process.py).
```python
client = EngineClient()
resp_json = client.start_process(process_key="PARALLEL_STEPS_EXAMPLE", variables={"intVar": "1", "strVar": "hello"},
tenant_id="6172cdf0-7b32-4460-9da0-ded5107aa977", business_key=str(uuid.uuid1()))
```
### [Fetch and Lock](https://docs.camunda.org/manual/latest/reference/rest/external-task/fetch/)
`ExternalTaskWorker(worker_id="1").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](https://docs.camunda.org/manual/latest/user-guide/process-engine/external-tasks/#long-polling-to-fetch-and-lock-external-tasks).
### [Complete](https://docs.camunda.org/manual/latest/reference/rest/external-task/post-complete/)
```python
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(worker_id="1").subscribe("topicName", handle_task)
```
### [Handle Failure](https://docs.camunda.org/manual/latest/reference/rest/external-task/post-failure/)
```python
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(worker_id="1").subscribe("topicName", handle_task)
```
### [Handle BPMN Error](https://docs.camunda.org/manual/latest/reference/rest/external-task/post-bpmn-error/)
```python
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", error_message="BPMN error occurred")
ExternalTaskWorker(worker_id="1" ).subscribe("topicName", handle_task)
```
### Access Process Variables
```python
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)
```
### [Correlate message](https://docs.camunda.org/manual/7.13/reference/bpmn20/events/message-events/)
Camunda provides functionality to send a message event to a running process instance.
You can read more about the message events here: https://docs.camunda.org/manual/7.13/reference/bpmn20/events/message-events/
In our to send a message event to a process instance, a new function called `correlate_message()` is added to [engine_client.py](./camunda/client/engine_client.py#L60)
We can correlate the message by:
- process_instance_id
- tenant_id
- business_key
- process_variables
You can find a usage example [here](./examples/correlate_message.py).
```python
client = EngineClient()
resp_json = client.correlate_message("CANCEL_MESSAGE", business_key="b4a6f392-12ab-11eb-80ef-acde48001122")
```
## AuthBasic Usage
To create an EngineClient with AuthBasic simple
```python
client = EngineClient(config={"auth_basic": {"username": "demo", "password": "demo"}})
resp_json = client.start_process(process_key="PARALLEL_STEPS_EXAMPLE", variables={"intVar": "1", "strVar": "hello"},
tenant_id="6172cdf0-7b32-4460-9da0-ded5107aa977", business_key=str(uuid.uuid1()))
```
To create an ExternalTaskWorker with AuthBasic simple
```python
from camunda.external_task.external_task import ExternalTask, TaskResult
from camunda.external_task.external_task_worker import ExternalTaskWorker
config = {"auth_basic": {"username": "demo", "password": "demo"}}
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(worker_id="1", config=config).subscribe("topicName", handle_task)
```
## License
The source files in this repository are made available under the [Apache License Version 2.0](./LICENSE).
Raw data
{
"_id": null,
"home_page": "https://github.com/yogeshrnaik/camunda-external-task-client-python3",
"name": "camunda-external-task-client-python3",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": "",
"keywords": "",
"author": "Deserve Labs",
"author_email": "devteam@deserve.com",
"download_url": "https://files.pythonhosted.org/packages/4b/64/72d6c9cf255d364806363102093ffd3db60f8d61d27fcd05d0b87dd6eabb/camunda_external_task_client_python3-4.4.0.tar.gz",
"platform": null,
"description": "[](https://github.com/camunda-community-hub/community)[](https://github.com/Camunda-Community-Hub/community/blob/main/extension-lifecycle.md#stable-)\n\n# camunda-external-task-client-python3\n\n\nThis repository contains Camunda External Task Client written in Python3.\n\n\nImplement your [BPMN Service Task](https://docs.camunda.org/manual/latest/user-guide/process-engine/external-tasks/) in Python3.\n\n> Python >= 3.7 is required\n\n## Installing\nAdd following line to `requirements.txt` of your Python project.\n```\ngit+https://github.com/trustfactors/camunda-external-task-client-python3.git/#egg=camunda-external-task-client-python3\n```\n\nOr use pip to install as shown below:\n```\npip install camunda-external-task-client-python3\n```\n\n## Running Camunda with Docker\nTo run the examples provided in [examples](./examples) folder, you need to have Camunda running locally or somewhere.\n\nTo run Camunda locally with Postgres DB as backend, you can use [docker-compose.yml](./docker-compose.yml) file.\n\n```\n$> docker-compose -f docker-compose.yml up\n```\n### Auth Basic Examples\n\nTo run the examples with Auth Basic provided in [examples/examples_auth_basic](./examples/examples_auth_basic) folder, you need to have Camunda with AuthBasic, running locally or somewhere.\n\nTo run Camunda with AuthBasic locally with Postgres DB as backend, you can use [docker-compose-auth.yml](./docker-compose-auth.yml) file.\n\n```\n$> docker-compose -f docker-compose-auth.yml up\n```\n\n## Usage\n\n1. Make sure to have [Camunda](https://camunda.com/download/) running.\n2. Create a simple process model with an External Service Task and define the topic as 'topicName'.\n3. Deploy the process to the Camunda BPM engine.\n4. In your Python code:\n\n```python\nimport time\nfrom camunda.external_task.external_task import ExternalTask, TaskResult\nfrom camunda.external_task.external_task_worker import ExternalTaskWorker\n\n# configuration for the Client\ndefault_config = {\n \"maxTasks\": 1,\n \"lockDuration\": 10000,\n \"asyncResponseTimeout\": 5000,\n \"retries\": 3,\n \"retryTimeout\": 5000,\n \"sleepSeconds\": 30\n}\n\ndef handle_task(task: ExternalTask) -> TaskResult:\n \"\"\"\n This task handler you need to implement with your business logic.\n After completion of business logic call either task.complete() or task.failure() or task.bpmn_error() \n to report status of task to Camunda\n \"\"\"\n # add your business logic here\n # ...\n \n # mark task either complete/failure/bpmnError based on outcome of your business logic\n failure, bpmn_error = random_true(), random_true() # this code simulate random failure\n if failure:\n # this marks task as failed in Camunda\n return task.failure(error_message=\"task failed\", error_details=\"failed task details\", \n max_retries=3, retry_timeout=5000)\n elif bpmn_error:\n return task.bpmn_error(error_code=\"BPMN_ERROR_CODE\", error_message=\"BPMN Error occurred\", \n variables={\"var1\": \"value1\", \"success\": False})\n \n # pass any output variables you may want to send to Camunda as dictionary to complete()\n return task.complete({\"var1\": 1, \"var2\": \"value\"}) \n\ndef random_true():\n current_milli_time = int(round(time.time() * 1000))\n return current_milli_time % 2 == 0\n\nif __name__ == '__main__':\n ExternalTaskWorker(worker_id=\"1\", config=default_config).subscribe(\"topicName\", handle_task)\n```\n\n## About External Tasks\n\nExternal Tasks are service tasks whose execution differs particularly from the execution of other service tasks (e.g. Human Tasks).\nThe execution works in a way that units of work are polled from the engine before being completed.\n\n**camunda-external-task-client-python** allows you to create easily such client in Python3.\n\n## Features\n\n### [Start process](https://docs.camunda.org/manual/latest/reference/rest/process-definition/post-start-process-instance/)\nCamunda provides functionality to start a process instance for a given process definition.\n\nTo start a process instance, we can use `start_process()` from [engine_client.py](./camunda/client/engine_client.py#L24)\n\nYou can find a usage example [here](./examples/start_process.py).\n\n```python\nclient = EngineClient()\nresp_json = client.start_process(process_key=\"PARALLEL_STEPS_EXAMPLE\", variables={\"intVar\": \"1\", \"strVar\": \"hello\"},\n tenant_id=\"6172cdf0-7b32-4460-9da0-ded5107aa977\", business_key=str(uuid.uuid1()))\n```\n\n### [Fetch and Lock](https://docs.camunda.org/manual/latest/reference/rest/external-task/fetch/)\n\n`ExternalTaskWorker(worker_id=\"1\").subscribe(\"topicName\", handle_task)` starts long polling of the Camunda engine for external tasks.\n\n* 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.\n* Long Polling is done periodically based on the `asyncResponseTimeout` configuration. Read more about [Long Polling](https://docs.camunda.org/manual/latest/user-guide/process-engine/external-tasks/#long-polling-to-fetch-and-lock-external-tasks).\n\n### [Complete](https://docs.camunda.org/manual/latest/reference/rest/external-task/post-complete/)\n```python\nfrom camunda.external_task.external_task import ExternalTask, TaskResult\nfrom camunda.external_task.external_task_worker import ExternalTaskWorker\ndef handle_task(task: ExternalTask) -> TaskResult:\n # add your business logic here\n \n # Complete the task\n # pass any output variables you may want to send to Camunda as dictionary to complete()\n return task.complete({\"var1\": 1, \"var2\": \"value\"})\n\nExternalTaskWorker(worker_id=\"1\").subscribe(\"topicName\", handle_task)\n```\n\n### [Handle Failure](https://docs.camunda.org/manual/latest/reference/rest/external-task/post-failure/)\n```python\nfrom camunda.external_task.external_task import ExternalTask, TaskResult\nfrom camunda.external_task.external_task_worker import ExternalTaskWorker\ndef handle_task(task: ExternalTask) -> TaskResult:\n # add your business logic here\n \n # Handle task Failure\n return task.failure(error_message=\"task failed\", error_details=\"failed task details\", \n max_retries=3, retry_timeout=5000)\n # This client/worker uses max_retries if no retries are previously set in the task\n # if retries are previously set then it just decrements that count by one before reporting failure to Camunda\n # when retries are zero, Camunda creates an incident which then manually needs to be looked into on Camunda Cockpit \n\nExternalTaskWorker(worker_id=\"1\").subscribe(\"topicName\", handle_task)\n```\n\n### [Handle BPMN Error](https://docs.camunda.org/manual/latest/reference/rest/external-task/post-bpmn-error/)\n```python\nfrom camunda.external_task.external_task import ExternalTask, TaskResult\nfrom camunda.external_task.external_task_worker import ExternalTaskWorker\ndef handle_task(task: ExternalTask) -> TaskResult:\n # add your business logic here\n \n # Handle a BPMN Failure\n return task.bpmn_error(error_code=\"BPMN_ERROR\", error_message=\"BPMN error occurred\")\n\nExternalTaskWorker(worker_id=\"1\" ).subscribe(\"topicName\", handle_task)\n```\n\n### Access Process Variables\n```python\nfrom camunda.external_task.external_task import ExternalTask, TaskResult\nfrom camunda.external_task.external_task_worker import ExternalTaskWorker\ndef handle_task(task: ExternalTask) -> TaskResult:\n # add your business logic here\n # get the process variable 'score'\n score = task.get_variable(\"score\")\n if int(score) >= 100:\n return task.complete(...)\n else:\n return task.failure(...) \n\nExternalTaskWorker().subscribe(\"topicName\", handle_task)\n```\n\n### [Correlate message](https://docs.camunda.org/manual/7.13/reference/bpmn20/events/message-events/)\nCamunda provides functionality to send a message event to a running process instance.\n\nYou can read more about the message events here: https://docs.camunda.org/manual/7.13/reference/bpmn20/events/message-events/\n\nIn our to send a message event to a process instance, a new function called `correlate_message()` is added to [engine_client.py](./camunda/client/engine_client.py#L60)\n\nWe can correlate the message by:\n- process_instance_id\n- tenant_id\n- business_key\n- process_variables\n\nYou can find a usage example [here](./examples/correlate_message.py).\n\n```python\nclient = EngineClient()\nresp_json = client.correlate_message(\"CANCEL_MESSAGE\", business_key=\"b4a6f392-12ab-11eb-80ef-acde48001122\")\n```\n## AuthBasic Usage\n\nTo create an EngineClient with AuthBasic simple\n\n```python\nclient = EngineClient(config={\"auth_basic\": {\"username\": \"demo\", \"password\": \"demo\"}})\nresp_json = client.start_process(process_key=\"PARALLEL_STEPS_EXAMPLE\", variables={\"intVar\": \"1\", \"strVar\": \"hello\"},\n tenant_id=\"6172cdf0-7b32-4460-9da0-ded5107aa977\", business_key=str(uuid.uuid1()))\n```\n\nTo create an ExternalTaskWorker with AuthBasic simple\n\n```python\nfrom camunda.external_task.external_task import ExternalTask, TaskResult\nfrom camunda.external_task.external_task_worker import ExternalTaskWorker\n\nconfig = {\"auth_basic\": {\"username\": \"demo\", \"password\": \"demo\"}}\n\ndef handle_task(task: ExternalTask) -> TaskResult:\n # add your business logic here\n \n # Complete the task\n # pass any output variables you may want to send to Camunda as dictionary to complete()\n return task.complete({\"var1\": 1, \"var2\": \"value\"})\n\nExternalTaskWorker(worker_id=\"1\", config=config).subscribe(\"topicName\", handle_task)\n```\n\n## License\nThe source files in this repository are made available under the [Apache License Version 2.0](./LICENSE).\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "Camunda External Task Client for Python 3",
"version": "4.4.0",
"project_urls": {
"Homepage": "https://github.com/yogeshrnaik/camunda-external-task-client-python3"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "a5487e6ca52ce3705233685ad5a6574791a9d30af7e1e8e1a50213c479cbf6bf",
"md5": "5f92248b53662d2b892d76d3f7212be7",
"sha256": "6abf322499541472f8231b3f1dbb18e97b8c3f732f53dcc37ea3dc3f32e26cc8"
},
"downloads": -1,
"filename": "camunda_external_task_client_python3-4.4.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "5f92248b53662d2b892d76d3f7212be7",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7",
"size": 45764,
"upload_time": "2023-08-01T04:35:49",
"upload_time_iso_8601": "2023-08-01T04:35:49.892846Z",
"url": "https://files.pythonhosted.org/packages/a5/48/7e6ca52ce3705233685ad5a6574791a9d30af7e1e8e1a50213c479cbf6bf/camunda_external_task_client_python3-4.4.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "4b6472d6c9cf255d364806363102093ffd3db60f8d61d27fcd05d0b87dd6eabb",
"md5": "c063082672a8a63bac960ea838f26b93",
"sha256": "ff95b029e06505e14db65b49773662618822e8bd109900e701e1a9068a0c288b"
},
"downloads": -1,
"filename": "camunda_external_task_client_python3-4.4.0.tar.gz",
"has_sig": false,
"md5_digest": "c063082672a8a63bac960ea838f26b93",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 29397,
"upload_time": "2023-08-01T04:35:51",
"upload_time_iso_8601": "2023-08-01T04:35:51.931577Z",
"url": "https://files.pythonhosted.org/packages/4b/64/72d6c9cf255d364806363102093ffd3db60f8d61d27fcd05d0b87dd6eabb/camunda_external_task_client_python3-4.4.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-08-01 04:35:51",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "yogeshrnaik",
"github_project": "camunda-external-task-client-python3",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [],
"lcname": "camunda-external-task-client-python3"
}
JSON
