Python Integration with Camunda


#1

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

Thanks!


#2

Hi @disha_umarwani,

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

  1. 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.
  2. 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


#3

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)