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.