BI / Power BI

Objectif : amener les données de suivi du temps dans votre data warehouse ou directement dans Power BI / Looker / Metabase via une synchronisation incrémentale.

Scopes recommandés (lecture seule) : org:fichajes:read, org:ausencias:read, org:saldos:read, org:estructura:read.

Pour la BI, utilisez une clé dédiée en lecture seule. Vous pourrez ainsi la révoquer ou la faire tourner sans impacter vos intégrations en écriture.

Pattern : polling incrémental avec `updated_since`

Toutes les listes acceptent updated_since=<ISO8601> et ne renvoient que ce qui a été modifié depuis cet instant. Le pattern :

Chargement initial (backfill)

Parcourez chaque ressource en paginant par cursor jusqu’à has_more=false. Conservez l’instant de départ comme marque d’eau (watermark).

Chargements incrémentaux

À chaque exécution planifiée, appelez ?updated_since=<watermark> et mettez à jour votre marque d’eau avec l’instant précédant le début de l’appel.

Dédupliquez par `id`

Comme updated_since s’appuie sur updated_at, un même enregistrement peut réapparaître s’il a changé. Faites un upsert par id (UUID) dans votre entrepôt.

Exemple en 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)

Prenez la marque d’eau avant de démarrer la synchronisation, pas après. Les enregistrements écrits pendant l’exécution seront ainsi repris à la passe suivante.

Ressources utiles pour la BI

Ressource	Apporte
`work-summaries`	Métriques de temps de travail prêtes à agréger (heures, supplémentaires, de nuit).
`check-ins`	Grain événement pour l’analyse de présence et de ponctualité.
`absences`	Absentéisme par type et par période.
`vacation-balances`	Soldes et provisions de congés.
`locations` / `units`	Dimensions de segmentation (site, département).

Connexion depuis Power BI

Power BI peut consommer l’API directement avec Web.Contents et le header d’autorisation. Exemple simplifié en 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

Pour paginer en Power Query, encapsulez l’appel dans une fonction qui suit meta.next_cursor avec List.Generate jusqu’à ce que has_more vaille false.

Respectez les limites de débit : pour les gros volumes, planifiez l’actualisation hors des heures de pointe et surveillez X-Kinmu-Quota-Remaining.

Alternative event-driven

Si vous préférez éviter le polling, abonnez-vous aux webhooks (checkin.created, absence.approved, vacation_balance.updated, …) et mettez à jour votre entrepôt à la réception de chaque événement. Les deux approches se combinent bien : webhooks pour le temps réel + un polling nocturne avec updated_since en filet de sécurité.