Skip to content

Generate inbound template

We can create a template for SleepConnector using

$ generate_template example.SleepConnector example/sleep.json
Generated template for example.SleepConnector

Tip

For more info on template generation see Generate outbound template.

Template conversion

SleepConnector class have been converted to a template:

  • name of the template was picked up from ConnectorConfig.name
  • zeebe:taskDefinition:type was set to ConnectorConfig.type (not visible in Modeler).
  • duration field of the SleepConnector was converted into a zeebe:input property.
  • A zeebe:taskHeader with key 'resultVariable' was added to allow users define the name of the variable where the output of the LogConnector.run method will be stored.
  • A 'Message name' and 'Correlation key' inputs were added.

Template

example/sleep.json
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
{
  "$schema": "https://unpkg.com/@camunda/zeebe-element-templates-json-schema@0.9.0/resources/schema.json",
  "name": "Sleep",
  "id": "a631e9a0-939e-4d73-b78c-34939e44eab0",
  "appliesTo": [
    "bpmn:ServiceTask"
  ],
  "properties": [
    {
      "binding": {
        "type": "zeebe:input",
        "name": "duration"
      },
      "label": "Duration of sleep in seconds",
      "type": "String",
      "group": "input",
      "feel": "optional"
    },
    {
      "binding": {
        "type": "zeebe:taskHeader",
        "key": "resultVariable"
      },
      "label": "Result variable",
      "type": "String",
      "group": "output"
    },
    {
      "binding": {
        "type": "zeebe:taskDefinition:type"
      },
      "type": "Hidden",
      "value": "sleep"
    },
    {
      "binding": {
        "type": "zeebe:input",
        "name": "correlation_key"
      },
      "label": "Correlation key",
      "type": "String",
      "group": "config",
      "feel": "optional"
    },
    {
      "binding": {
        "type": "zeebe:input",
        "name": "message_name"
      },
      "label": "Message name",
      "type": "String",
      "group": "config",
      "feel": "optional"
    }
  ],
  "groups": [
    {
      "id": "input",
      "label": "Input"
    },
    {
      "id": "output",
      "label": "Output"
    },
    {
      "id": "config",
      "label": "Configuration"
    }
  ]
}
example/sleep.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
import asyncio

from pydantic import Field

from python_camunda_sdk import InboundConnector


class SleepConnector(InboundConnector):
    duration: int = Field(description="Duration of sleep in seconds")

    async def run(self) -> bool:
        await asyncio.sleep(self.duration)
        return True

    class ConnectorConfig:
        name = "Sleep"
        type = 'sleep'