pklauke / pycamunda

Python REST api client for the workflow and decision automation engine Camunda.

PyCamunda

PyCamunda is a Python REST api client for the workflow and decision automation engine Camunda.

PyCamunda takes care of Camunda-specific characteristics and gives you an intuitive Python interface so you can focus on developing your actual business application. So PyCamunda deals with http requests, nested dictionaries, camelCase keys, datetime formatting and more for you.

Simply..

• .. create an instance of the PyCamunda class corresponding to a Camunda endpoint,
• .. configure it using keyword arguments and methods,
• .. send the request and get the response serialized in a Python data class.

import pycamunda.processinst

url = 'http://localhost/engine-rest'

get_instances = pycamunda.processinst.GetList(url, process_definition_key='MyProcess')
instances = get_instances()

for instance in instances:
    print('Process instance id:', instance.id_)

Output:

Process instance id: 174cb832-a8a7-11ea-8129-0242ac110002
Process instance id: 1a70a8e9-a7ed-11ea-8129-0242ac110002
...

Installation

PyCamunda requires Python >= 3.7 and can be installed from the offical Python Package Index.

$ pip install pycamunda

Documentation

The latest documentation can be found here.

pycamunda's Issues

externaltask with error, failed to query /errorDetails, missing auth

pycamunda/externaltask.py

       if self.request_error_details:
            for external_task in external_tasks:
                if external_task.error_details is None:
                    try:
                        response = requests.get(self.url + f'/{external_task.id_}/errorDetails')
                    except requests.exceptions.RequestException:
                        raise pycamunda.PyCamundaException()
                    if not response:
                        pycamunda.base._raise_for_status(response)
                    external_task.error_details = response.text

replace:

response = requests.get(self.url + f'/{external_task.id_}/errorDetails')

with

response = self.session.get(self.url + f'/{external_task.id_}/errorDetails')

How can I implement camunda logs here

i tried this logback.xml file for logs in tomcat.

<configuration>

  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
    <!-- encoders are assigned the type
         ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
    <encoder>
      <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
    </encoder>
  </appender>

  <logger name="org.apache.ibatis" level="info" />

  <logger name="javax.activation" level="info" />

  <logger name="org.springframework" level="info" />

  <logger name="org.camunda" level="debug" />

  <logger name="org.camunda.bpm.engine.test" level="debug" />


  <logger name="org.camunda.bpm.engine.impl.persistence.entity" level="debug" />
  <logger name="org.camunda.bpm.engine.history" level="debug" />

  <logger name="org.camunda.bpm.engine.bpmn.parser" level="debug" />
  <logger name="org.camunda.bpm.engine.bpmn.behavior" level="debug" />

  <logger name="org.camunda.bpm.engine.cmmn.transformer" level="debug" />
  <logger name="org.camunda.bpm.engine.cmmn.behavior" level="debug" />
  <logger name="org.camunda.bpm.engine.cmmn.operation" level="debug" />

  <logger name="org.camunda.bpm.engine.cmd" level="debug" />
  <logger name="org.camunda.bpm.engine.persistence" level="debug" />
  <logger name="org.camunda.bpm.engine.tx" level="debug" />
  <logger name="org.camunda.bpm.engine.cfg" level="debug" />
  <logger name="org.camunda.bpm.engine.jobexecutor" level="debug" />
  <logger name="org.camunda.bpm.engine.context" level="debug" />
  <logger name="org.camunda.bpm.engine.core" level="debug" />
  <logger name="org.camunda.bpm.engine.pvm" level="debug" />
  <logger name="org.camunda.bpm.engine.dmn" level="debug" />

  <logger name="org.camunda.bpm.dmn.feel" level="debug" />
  <logger name="org.camunda.bpm.dmn.feel.scala" level="debug" />
  <logger name="org.camunda.feel.FeelEngine" level="debug" />

  <logger name="org.camunda.bpm.engine.externaltask" level="debug" />
  <logger name="org.camunda.bpm.engine.identitye" level="debug" />
  <logger name="org.camunda.bpm.engine.incident" level="debug" />
  <logger name="org.camunda.bpm.engine.migration" level="debug" />
  <logger name="org.camunda.bpm.engine.plugin.admin" level="debug" />
  
  <logger name="org.camunda.bpm.engine.rest.exception" level="debug" />
  <logger name="org.camunda.bpm.engine.script" level="debug" />
  <logger name="org.camunda.bpm.engine.security" level="debug" />
  <logger name="org.camunda.bpm.engine.telemetry" level="debug" />



  <logger name="org.camunda.bpm.engine.metrics" level="debug" />
  <logger name="org.camunda.bpm.engine.util" level="debug" />

  <logger name="org.camunda.bpm.application" level="debug" />
  <logger name="org.camunda.bpm.container" level="debug" />


  <root level="debug">
    <appender-ref ref="STDOUT" />
  </root>

</configuration>
```

handle requests timeout

add parameter timeout to http requests

Any plan to add `/history/*` ?

These calls can be used to retrieve history of committed tasks

https://docs.camunda.org/manual/7.8/reference/rest/history/

Migrate Extension to the Camunda Community Hub?

Hello!

I hope today finds you well! My name is Rin Oliver, and I'm a Technical Community Builder at Camunda. I'm opening this issue today to ask if you might be interested in migrating this extension over to the Camunda Community Hub?

We'd love to hear your feedback on how we might improve the overall community extension development process for maintainers and contributors moving forward, and learn more about what you'd like to see from this initiative in the future.

If migrating this extension something you'd be interested in, I can send you the instructions for doing so via email, or whatever communication method works best for you. I'm also more than happy to schedule a call and talk over this in more detail if that would be preferable to you, for whenever best suits your schedule over the next few weeks.

Thank you so much for your time. I look forward to hearing from you.

Best,
Rin

Current Camunda version (7.16) no longer has field 'errorDetails' in http://localhost:8080/engine-rest/external-task

which makes line 55 of externaltask.py fail:

File "C:\...\pycamunda\externaltask.py", line 55, in load
    error_details=data['errorDetails'],
KeyError: 'errorDetails'

can a request.Session object be passed in and used for all subsequent CamundaRequest.__call__ s?

pycamunda/pycamunda/base.py

Line 41 in 44dba50

def __call__(self, method: RequestMethod, *args, **kwargs) -> requests.Response:

Is there a nicer way to pass in request.Session object and headers?
In order to access our camunda setup I need to pass BasicAuth as well as go through a login user/pass.

So I set up a request.Session object outside of pycamunda... and I hacked base.py to accept it. It would be useful if this Session object could be set once and used by all subsequent calls.

I did this to base.py on my local venv (apologies!)

    def set_console_camunda_session(self, session: object):     # kludge in my session obj 
        self._console_camunda_session = session

    def __call__(self, method: RequestMethod, *args, **kwargs) -> requests.Response:

        try:
            if self._console_camunda_session:           # check for my session object
                s = self._console_camunda_session
                response = s.session.request(
                    method=method.value,
                    url=self.url,
                    headers=s.headers,                   # add headers
                    params=self.query_parameters(),
                    json=self.body_parameters(),
                    auth=self.auth,
                    files=self.files
                )
            else:
                response = requests.request(
                    method=method.value,
                    url=self.url,
                    params=self.query_parameters(),
                    json=self.body_parameters(),
                    auth=self.auth,
                    files=self.files
                )```

externaltask request_error_details does not use same requests session as other calls

Hello,

I'm using a custom requests.Session that has authentication pre-applied. When using external task Get or GetList, the default is for request_error_details to be True, which uses requests.get and not the session. For me, this caused a 401 error, since it did not send the authentication from the session.

https://github.com/pklauke/pycamunda/blob/master/pycamunda/externaltask.py#L120
https://github.com/pklauke/pycamunda/blob/master/pycamunda/externaltask.py#L269

Thanks, love the lib!

QueryParameter needs to append "In"

pycamunda/pycamunda/variable.py

Line 80 in 44dba50

process_instance_id_in = QueryParameter('processInstanceId')

I was querying variables on my server and got OutOfMemory even though I was passing in a processInstanceId. Turns out there is a small bug in the naming. Should be a quick fix.

task.GetList sort_by wrong typing

pycamunda/pycamunda/task.py

Line 236 in 7689040

'created': 'description',

this is the wrong typing?
should be this
'created': 'created',

please add pycamunda.task.Delete

Hi,

can you add pycamunda.task.Delete pls. per https://docs.camunda.org/manual/7.18/reference/rest/task/delete/

class Delete(pycamunda.base.CamundaRequest):

    task_id = PathParameter('id')

    def __init__(self, url: str, task_id: str):
        """Delete a task.

        :param url: Camunda Rest engine URL.
        :param task_id: Id of the task.
        """
        super().__init__(url=url + URL_SUFFIX + '/{id}')
        self.task_id = task_id

    def __call__(self, *args, **kwargs) -> None:
        """Send the request."""
        super().__call__(pycamunda.base.RequestMethod.DELETE, *args, **kwargs)

please add HistoryProcessInstance

https://docs.camunda.org/rest/camunda-bpm-platform/7.19/#tag/Historic-Process-Instance/operation/getHistoricProcessInstances

Message Correlation with results Failure

When I try to correlate a message with results and variables in results, result loading fails with KeyError https://github.com/fasfoxcom/pycamunda/blob/master/pycamunda/message.py#L52. Here is the link of the doc https://docs.camunda.org/manual/7.17/reference/rest/message/post-message/#response-2. According to the doc, there is an error in the loading logic.

Process instance variables subresource

Hello!

First of all I would like to thank you @pklauke for developing pycamunda. It seems to be the most complete Camunda REST API client out there.

I'm looking into ways to update process variables of a running process instance. The Camunda API documentation explains that there is a "Variables" subresource under "Process Instance":
https://docs.camunda.org/manual/7.15/reference/rest/process-instance/
https://docs.camunda.org/manual/7.15/reference/rest/process-instance/variables/

However, I cannot find any info how to access that "Variables" subresource in the pycamunda documentation:
https://pycamunda.readthedocs.io/en/latest/src/api/processinst.html

Am I looking in the wrong place or is this not implemented yet?

Source is missing for new deploymen

Hi! First of all thank you for this library.

I was playing with new deployments and have noticed that new deployments are created with source field set to null.

Code is like this:

import logging
from pycamunda import deployment

logging.basicConfig(level=logging.DEBUG)

url = "http://localhost:8080/engine-rest"

create = deployment.Create(url, name="deployment", tenant_id="123", source="dep.bpmn")
create.add_resource(open("diagram_1.bpmn", "rb"))
response = create()
print(response)

Deployment is created, file is uploaded and diagram is visible in UI, but the source and tenant fields are set to null:

Body parameters are (excluding file):

{'deployment-name': 'deployment', 'enable-duplicate-filtering': False, 'deploy-changed-only': False, 'deployment-source': 'dep.bpmn', 'tenant-id': '123'}

Also, the /deployment/create endpoint returns DeploymentWithDefinitions object but the library does not return anything, is this intentional? Reference.

And last question — is this project production ready?

Thanks!

any plan to implement the migration API?

https://docs.camunda.org/manual/7.15/reference/rest/migration/
I have to migrate many process instances and this would be useful.