Skip to main content

Examples

The examples use Python 3 with requests and read KOMMUNS_API, CLIENT_ID, CLIENT_SECRET from the environment.

kommuns.py — a small client used below
import os
import time
import uuid

import requests

API = os.environ["KOMMUNS_API"]
_token = {"value": None, "until": 0}


def token():
if time.time() > _token["until"] - 60:
r = requests.post(f"{API}/oauth/token", data={"grant_type": "client_credentials"},
auth=(os.environ["CLIENT_ID"], os.environ["CLIENT_SECRET"]), timeout=10)
r.raise_for_status()
_token.update(value=r.json()["access_token"], until=time.time() + r.json()["expires_in"])
return _token["value"]


def call(method, path, body=None, key=None, **params):
headers = {"Authorization": f"Bearer {token()}"}
if key:
headers["Idempotency-Key"] = key
while True:
r = requests.request(method, f"{API}{path}", json=body, params=params, headers=headers, timeout=30)
if r.status_code == 429: # over the per-minute limit
time.sleep(int(r.headers.get("Retry-After", "5")))
continue
r.raise_for_status()
return r.json()


def every(path, **params):
cursor = None
while True:
page = call("GET", path, limit=200, **params, **({"cursor": cursor} if cursor else {}))
yield from page["data"]
cursor = page["next_cursor"]
if not cursor:
return

A sensor alarm opens feedback​

from kommuns import call

def on_alarm(alarm):
"""alarm: {"id": "8841", "equipment": "SAP-EQ-10023", "text": "Chiller 2: high temperature"}"""
node = call("GET", "/inventory", external_id=alarm["equipment"])["data"][0]
return call("POST", "/issues", {"inventory_id": node["id"], "subject": alarm["text"],
"external_id": f"BMS-{alarm['id']}"},
key=f"bms-alarm-{alarm['id']}") # a repeated alarm doesn't open a second one

The equipment carries the automation system's number as its External system number (General tab).

An ERP keeps a copy of the work orders​

import datetime
from kommuns import every

def sync(since=None):
"""since: the largest changed_at of the last run (ISO), None the first time"""
newest = since
params = {}
if since:
moment = datetime.datetime.fromisoformat(since.replace("Z", "+00:00")) - datetime.timedelta(minutes=1)
params["changed_since"] = moment.isoformat().replace("+00:00", "Z")
for work_order in every("/work-orders", **params):
upsert_into_erp(work_order) # by work_order["id"] (or your external_id)
newest = max(newest or "", work_order["changed_at"])
return newest

A webhook receiver​

import json
import os
from http.server import BaseHTTPRequestHandler, HTTPServer

from verify import verified # the function on the Webhooks page

SECRET = os.environ["KOMMUNS_WEBHOOK_SECRET"]
seen = set()


class Receiver(BaseHTTPRequestHandler):
def do_POST(self):
body = self.rfile.read(int(self.headers["Content-Length"]))
if not verified(SECRET, self.headers.get("Kommuns-Signature", ""), body):
self.send_response(400)
self.end_headers()
return
self.send_response(200) # answer first, work afterwards
self.end_headers()
event = json.loads(body)
if event["id"] in seen: # delivered at least once
return
seen.add(event["id"])
print(event["type"], event["data"].get("work_order", {}).get("title"))


HTTPServer(("0.0.0.0", 8080), Receiver).serve_forever()

Power BI / Excel​

Power Query can read a list directly. Create the integration with work_orders:read, get a token (it lasts an hour; for a scheduled refresh, put the token step into the query as below), and use a blank query:

let
Api = "https://app.kommuns.com/api/public/v1",
Auth = Json.Document(Web.Contents(Api & "/oauth/token", [
Headers = [#"Content-Type" = "application/x-www-form-urlencoded",
Authorization = "Basic " & Binary.ToText(Text.ToBinary(ClientId & ":" & ClientSecret), BinaryEncoding.Base64)],
Content = Text.ToBinary("grant_type=client_credentials")])),
Page = (cursor) => Json.Document(Web.Contents(Api & "/work-orders", [
Headers = [Authorization = "Bearer " & Auth[access_token], #"Accept-Language" = "en"],
Query = [limit = "200"] & (if cursor = null then [] else [cursor = cursor])])),
Pages = List.Generate(() => Page(null), each _ <> null,
each if _[next_cursor] = null then null else Page(_[next_cursor])),
Rows = List.Combine(List.Transform(Pages, each _[data])),
Table = Table.FromRecords(Rows)
in
Table

ClientId and ClientSecret are query parameters; keep them out of shared files.