BI / Power BI
Ziel: die Zeiterfassungsdaten in dein Data Warehouse oder direkt nach Power BI / Looker / Metabase bringen — per inkrementeller Synchronisierung.
Empfohlene Scopes (nur lesend): org:fichajes:read, org:ausencias:read, org:saldos:read, org:estructura:read.
Verwende für BI einen dedizierten Read-only-Key. So kannst du ihn widerrufen oder rotieren, ohne Schreibintegrationen zu beeinträchtigen.
Pattern: inkrementelles Polling mit updated_since
Alle Listen akzeptieren updated_since=<ISO8601> und liefern nur, was sich seit diesem Zeitpunkt geändert hat. Das Pattern:
Erstladung (Backfill)
Gehe jede Ressource durch und paginiere per cursor, bis has_more=false. Speichere den Startzeitpunkt als Wasserzeichen (watermark).
Inkrementelle Läufe
Frage bei jedem geplanten Lauf ?updated_since=<watermark> an und setze dein Wasserzeichen auf den Zeitpunkt vor Beginn des Aufrufs.
Dedupliziere über id
Da updated_since auf updated_at basiert, kann derselbe Datensatz erneut auftauchen, wenn er sich geändert hat. Mache in deinem Speicher ein Upsert über die id (UUID).
Beispiel in Python
import os, requests
from datetime import datetime, timezone
BASE = os.environ["KINMU_BASE_URL"]
KEY = os.environ["KINMU_API_KEY"]
session = requests.Session()
session.headers.update({"Authorization": f"Bearer {KEY}"})
def sync(resource, since=None):
rows, cursor = [], None
params = {"limit": 100}
if since:
params["updated_since"] = since
while True:
if cursor:
params["cursor"] = cursor
r = session.get(f"{BASE}/{resource}", params=params, timeout=30)
r.raise_for_status()
body = r.json()
rows.extend(body["data"])
if not body["meta"]["has_more"]:
break
cursor = body["meta"]["next_cursor"]
return rows
# Marca de agua ANTES de llamar, para no perder registros escritos durante la sync
watermark = datetime.now(timezone.utc).isoformat()
employees = sync("employees", since=load_last_watermark()) # incremental
checkins = sync("check-ins", since=load_last_watermark())
save_watermark(watermark)Nimm das Wasserzeichen vor dem Start der Synchronisierung, nicht danach. So landen Datensätze, die während des Laufs geschrieben wurden, im nächsten Durchgang.
Nützliche Ressourcen für BI
| Ressource | Liefert |
|---|---|
work-summaries | Arbeitszeitmetriken, fertig für die Aggregation (Stunden, Überstunden, Nachtarbeit). |
check-ins | Ereignis-Granularität für Anwesenheits- und Pünktlichkeitsanalysen. |
absences | Abwesenheitsquote nach Typ und Zeitraum. |
vacation-balances | Urlaubssalden und Rückstellungen. |
locations / units | Dimensionen zum Segmentieren (Standort, Abteilung). |
Anbindung aus Power BI
Power BI kann die API direkt mit Web.Contents und Authorization-Header konsumieren. Vereinfachtes Beispiel in Power Query (M):
let
BaseUrl = "https://api.kinmu.app/v1",
ApiKey = "kinmu_sk_live_…", // usa Parámetros / almacén de credenciales, no lo escribas en claro
Source = Json.Document(
Web.Contents(BaseUrl, [
RelativePath = "work-summaries",
Query = [ period = "month", from = "2026-01-01", #"to" = "2026-12-31", limit = "100" ],
Headers = [ Authorization = "Bearer " & ApiKey, Accept = "application/json" ]
])
),
Data = Source[data],
Table = Table.FromRecords(Data)
in
TableZum Paginieren in Power Query kapselst du den Aufruf in eine Funktion, die meta.next_cursor mit List.Generate folgt, bis has_more gleich false ist.
Respektiere die Rate Limits: Plane die Aktualisierung bei großen Volumina außerhalb der Stoßzeiten und behalte X-Kinmu-Quota-Remaining im Blick.
Event-getriebene Alternative
Wenn du nicht pollen willst, abonniere Webhooks (checkin.created, absence.approved, vacation_balance.updated, …) und aktualisiere deinen Speicher bei jedem Event. Beides kombiniert sich gut: Webhooks für Echtzeit + ein nächtliches Polling mit updated_since als Sicherheitsnetz.