BI / Power BI

Objetivo: levar os dados de controle de ponto para o seu data warehouse ou diretamente para o Power BI / Looker / Metabase com sincronização incremental.

Scopes recomendados (somente leitura): org:fichajes:read, org:ausencias:read, org:saldos:read, org:estructura:read.

Para BI, use uma chave dedicada somente de leitura. Assim você pode revogá-la ou rotacioná-la sem afetar integrações de escrita.

Padrão: polling incremental com `updated_since`

Todas as listagens aceitam updated_since=<ISO8601> e retornam apenas o que foi modificado desde esse instante. O padrão:

Carga inicial (backfill)

Percorra cada recurso paginando por cursor até has_more=false. Guarde o instante de início como marca-d’água (watermark).

Cargas incrementais

Em cada execução agendada, peça ?updated_since=<watermark> e atualize sua marca-d’água para o instante anterior ao início da chamada.

Deduplique por `id`

Como updated_since se baseia em updated_at, um mesmo registro pode aparecer de novo se mudou. Faça upsert por id (UUID) no seu armazenamento.

Exemplo em 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)

Capture a marca-d’água antes de começar a sincronização, não depois. Assim os registros escritos enquanto o processo rodava entram na próxima passada.

Recursos úteis para BI

Recurso	Traz
`work-summaries`	Métricas de jornada prontas para agregação (horas, extras, noturnas).
`check-ins`	Grão de evento para análise de presença e pontualidade.
`absences`	Absenteísmo por tipo e período.
`vacation-balances`	Saldos e provisões de férias.
`locations` / `units`	Dimensões para segmentar (local, departamento).

Conexão a partir do Power BI

O Power BI pode consumir a API diretamente com Web.Contents e header de autorização. Exemplo simplificado em 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

Para paginar no Power Query, envolva a chamada em uma função que siga meta.next_cursor com List.Generate até has_more ser false.

Respeite os rate limits: para volumes grandes, agende a atualização fora dos horários de pico e acompanhe X-Kinmu-Quota-Remaining.

Alternativa event-driven

Se preferir não fazer polling, inscreva-se em webhooks (checkin.created, absence.approved, vacation_balance.updated, …) e atualize seu armazenamento a cada evento recebido. Os dois combinam bem: webhooks para tempo real + um polling noturno com updated_since como rede de segurança.