BI / Power BI

Obiettivo: portare i dati di rilevazione presenze nel tuo data warehouse o direttamente in Power BI / Looker / Metabase con la sincronizzazione incrementale.

Scopes consigliati (sola lettura): org:fichajes:read, org:ausencias:read, org:saldos:read, org:estructura:read.

Per la BI usa una chiave dedicata di sola lettura. Così puoi revocarla o ruotarla senza toccare le integrazioni di scrittura.

Pattern: polling incrementale con `updated_since`

Tutti gli elenchi accettano updated_since=<ISO8601> e restituiscono solo ciò che è stato modificato da quell’istante. Il pattern:

Caricamento iniziale (backfill)

Percorri ogni risorsa paginando per cursor fino a has_more=false. Salva l’istante di inizio come marca temporale (watermark).

Caricamenti incrementali

A ogni esecuzione pianificata, richiedi ?updated_since=<watermark> e aggiorna la marca temporale all’istante precedente all’inizio della chiamata.

Deduplica per `id`

Poiché updated_since si basa su updated_at, uno stesso record può ricomparire se è cambiato. Fai upsert per id (UUID) nel tuo storage.

Esempio 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)

Prendi la marca temporale prima di avviare la sincronizzazione, non dopo. Così i record scritti mentre il processo era in esecuzione entrano nella passata successiva.

Risorse utili per la BI

Risorsa	Cosa offre
`work-summaries`	Metriche di giornata pronte per l’aggregazione (ore, straordinari, notturne).
`check-ins`	Grana a livello di evento per analisi di presenza e puntualità.
`absences`	Assenteismo per tipo e periodo.
`vacation-balances`	Saldi e accantonamenti ferie.
`locations` / `units`	Dimensioni per segmentare (sede, reparto).

Connessione da Power BI

Power BI può consumare l’API direttamente con Web.Contents e header di autorizzazione. Esempio semplificato 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
    Table

Per paginare in Power Query, avvolgi la chiamata in una funzione che segua meta.next_cursor con List.Generate finché has_more è false.

Rispetta i rate limit: per grandi volumi, pianifica l’aggiornamento fuori dalle ore di punta e tieni d’occhio X-Kinmu-Quota-Remaining.

Alternativa event-driven

Se preferisci non fare polling, sottoscrivi i webhooks (checkin.created, absence.approved, vacation_balance.updated, …) e aggiorna il tuo storage a ogni evento ricevuto. Le due strade si combinano bene: webhooks per il tempo reale + un polling notturno con updated_since come rete di sicurezza.