Python Integration with Camunda

disha_umarwani · September 9, 2018, 12:53am

I’m just starting out with using Camunda and I’m a Python developer. Any resources for integrating Camunda with external Python scripts?

Thanks!

thorben · September 10, 2018, 4:33pm

Hi @disha_umarwani,

In general there are two approaches to use Python code in a process:

External task workers that are implemented in Python and that run independently of the Camunda process. I’m not aware of a Python example, but this article describes the general principle very well and with a proper REST client you should get started in no time.
Script tasks embedded into the process model that are run via Jython from within the Camunda process. You can edit those in Camunda modeler, just set python as the scription language. Furthermore, make sure the Jython scripting engine is in Camunda’s classpath.

I generally recommend approach 1, as it should be easier to use and gives you full flexibility on how you write and build your Python code. Approach 2 could be suitable for rather simple scripts and logic.

Cheers,
Thorben

Nele · September 11, 2018, 8:42am

Hey Disha,
I recently worked on an external python worker. Here are my Code snippets. It might not be perfect but it is working and might help you.

import requests
import json
import time

#Define API Endpoint
url = 'http://localhost:8080/engine-rest/external-task/fetchAndLock'

# Define Json
task= {"workerId":"XXX",
  "maxTasks":2,
  "usePriority":"true",
  "topics":
      [{"topicName": "coolStuff",
      "lockDuration": 10000,
      "variables": ["orderId"]
      }]
}

# Connect to the API (Fetch and Lock)
try:
    res = requests.post(url, json=task)
    print(res.status_code)

#Get Body Text
    
    body = res.text    
    print(body)    
            
except: 
    print('Engine is down')


while body == '[]':
    res = requests.post(url, json=task)
    time.sleep(5)
    
    if body !='[]':
        break
        
#here goes your business logic!

# Connect to the API (Complete)
response = json.loads(body)
x = response[0]['id']


#convert into a string
x = str(x)

complete_url = 'http://localhost:8080/engine-rest/external-task/'+x+'/complete'



response= {
  "workerId": "XXX",
  "variables":
      {"aVariable": {"value": "aStringValue"},
      "anotherVariable": {"value": 42},
      "aThirdVariable": {"value": "true"}}
}

complete = requests.post(complete_url, json=response)

Harm · October 11, 2018, 3:15pm

Hi Nele, thank you for your code. It works fine, but I have one issue. What I would expect that is the following:

the program starts polling for something to happen (it does)
it comes into action at the moment I start a process in Camunda which has the token of this worker.

But it’s the other way around: the worker only picks up the Camunda proces when I start the worker after I start the process win Camunda. This is the other way around than what I would expect.

Do you know what I could be doing wrong?

Here’s my code (actually your code):

import requests
import json
import time

# Define API Endpoint
url = 'http://localhost:8080/engine-rest/external-task/fetchAndLock'

# Define Json
task = {"workerId": "XXX",
        "maxTasks": 10,
        "usePriority": "true",
        "topics":
            [{"topicName": "loadData",
              "lockDuration": 10000,
              "variables": ["orderId"]
              }]
        }
body = []
# Connect to the API (Fetch and Lock)
try:
    res = requests.post(url, json=task)
    print('res.status_code', res.status_code)
    # Get Body Text
    body = res.text
    print('res.text', body)
except:
    print('Engine is down')

while body == '[]':
    print('wait')
    time.sleep(5)
    res = requests.post(url, json=task)
    print('result', res)
    if body != '[]':
        break

# here goes your business logic!
print("Connection succeeded")

# Connect to the API (Complete)
response = json.loads(body)
x = response[0]['id']

print('id', x)

# convert into a string
x = str(x)

complete_url = 'http://localhost:8080/engine-rest/external-task/' + x + '/complete'

response = {
    "workerId": "XXX",
    "variables":
        {"aVariable": {"value": "aStringValue"},
         "anotherVariable": {"value": 42},
         "aThirdVariable": {"value": "true"}}
}

print('response', response)

complete = requests.post(complete_url, json=response)

print('complete', complete)

Nele · October 21, 2018, 7:13pm

Hi @Harm
There was a mistake in my snippet. I am sorry for that.

The variable body was not assigned again in the while loop. Therefore the body stays the way it is from the first rest call.

You need to change the while loop to the following:

while body =='[ ]':
   res = requests.post(url, json =task)
   body = res.text
   time.sleep(5)

    if body !='[ ]':
       break

That should do the trick.

Anyhow the worker will terminate as soon as it has fetched and locked and completed one Task.
So you might need to change it if you want the worker to start again after getting one task

Cheers Nele

Harm · October 22, 2018, 2:42pm

Hi Nele,

Thanks, I got it working already.

Regards,
Harm

Harm · October 22, 2018, 2:46pm

Hi Disha,

This an example of a script that executes a python script and gives back the returnvalue to be evaluated in the proces. This way you can execute Python from a Camunda process step.

def pmdCommand = '/usr/local/bin/python3.7 /users/user/Documents/FTKDemo/DEXRS/DEXDNBNew/software/loadSourceDataWorkflow.py -t '+ Regulation + ' -f '+Frequency+ ' -d '+ ReportingDate.format('yyyy-MM-dd')

def sout = new StringBuffer()
def serr = new StringBuffer()

def process = pmdCommand.execute()
process.consumeProcessOutput(sout, serr)
process.waitForProcessOutput()
execution.setVariable("loadReturnValue", process.exitValue())

olekslev · November 13, 2018, 8:53am

THanks for sharing some python examples. Its exactly what I am looking for right now!

Harm · November 13, 2018, 9:10am

My pleasure!

Nele · November 28, 2019, 2:12pm

Maybe just an update as I got a pn regarding the python snippet recently.

I put the snippets together in a python module which can be downloaded and the imported into your python code, so you can use the main functions to communicate with the Camunda Engine.

philippfromme · November 28, 2019, 2:58pm

I like the fact that this topic is in the category Modeler.

yogeshrnaik · October 24, 2020, 12:59pm

Hello,
You can use this

Regards,
Yogesh

Noordsestern · January 22, 2021, 7:37am

Hi,

I used @yogeshrnaik great external task client, too.

If you need to integrate not only external task patterns but also for instance deployments, process definitions and so on, then the generic camunda client might be something for you. It is auto generated from openapi specification of Camunda, thus it looks a bit bulky, but it is reliable.

pip install generic-camunda-client

Documentation (also generated) is here. Code snippets from the documentation are very reliable.

Greetings,
Markus

vitalik · January 28, 2021, 10:00am

Hi Yogesh! I decided to use your lib, but I ran into the lack of the ability to parallel listening to several topics, as in Java. Can you give any advice other than threading?

yogeshrnaik · January 28, 2021, 1:29pm

@vitalik
Each ExternalTaskWorker in my library listens forever (in while True) loop. So, if you want multiple workers listening parallel, you need to use threading.
Any particular reason, why you don’t want to use threads?

ExternalTaskWorker.subscribe() accepts one topic or a list of topic_names. So, you can use a single worker if you want to listen to multiple topics. But this worker’s subscribe method uses while True loop. So you need to run it in a separate thread. If you run it in main thread, it will be a blocking call.

Alternative to ExternalTaskWorker.subscribe() is that you can use ExternalTaskWorker.fetch_and_execute() method when you want to fetch and execute any particular External task. But, this is a one time call and will raise NoExternalTaskFound immediately if there are no matching external tasks. You then have to call this fetch_and_execute() in some loop (which subscribe() is doing already).

Another option to threads is to change my library to use async/await mechanism. But, for that you will have to rewrite all the network calls using async/await. I had done that in early versions of my library but I ran into some issues and hence removed that from library.

Nele · May 11, 2021, 4:51pm

Hello Python Lovers,

I found another Python Camunda module, which looks pretty amazing. I haven’t tried it out yet, but still like to share it:
https://pycamunda.readthedocs.io/en/latest/index.html

Curious to see what you think

ubaidce · June 1, 2021, 3:27pm

Hi Yogesh, thank you for sharing your wonderful libraries!!.

" ExternalTaskWorker.subscribe() accepts one topic or a list of topic_names. So, you can use a single worker if you want to listen to multiple topics. But this worker’s subscribe method uses while True loop. So you need to run it in a separate thread. If you run it in main thread, it will be a blocking call."

Slightly confused about the above - for multiple topics, how do I use the external task worker to map the topics to the distinct actions ?

Sahil_s · June 5, 2022, 10:19pm

I have used Pycamunda to build an app which executes external Tasks. Think it is pretty good.

system · January 30, 2024, 11:22am