Add Bonusly Data to BI Tool

CREATE DATABASE IF NOT EXISTS BONUSLY;

CREATE SCHEMA IF NOT EXISTS BONUSLY.ANALYTICS;

USE SCHEMA BONUSLY.ANALYTICS;

-- Raw landing table: one row per recognition event, full JSON in DATA

CREATE TABLE IF NOT EXISTS RECOGNITION_EVENTS_RAW (

ROW_KEY STRING PRIMARY KEY,

RECOMPUTED_AT TIMESTAMP_TZ,

DATA VARIANT

);

-- Cursor state for the incremental sync (single row)

CREATE TABLE IF NOT EXISTS SYNC_STATE (

RECOMPUTED_AT TIMESTAMP_TZ,

ROW_KEY STRING

);

-- Stage + file format for the snapshot file

CREATE FILE FORMAT IF NOT EXISTS NDJSON_FORMAT TYPE = JSON;

CREATE STAGE IF NOT EXISTS BONUSLY_SNAPSHOT_STAGE FILE_FORMAT = NDJSON_FORMAT;

# Request the export

curl -s -X POST https://bonus.ly/api/public/analytics/recognition_events/snapshot \

-H "Authorization: Bearer $BONUSLY_PAT"

# → {"snapshot": {"id": "6a2c...", "status": "queued", ...}}

# Poll until status is "ready" (typically a minute or two)

curl -s https://bonus.ly/api/public/analytics/snapshots/6a2c... \

-H "Authorization: Bearer $BONUSLY_PAT"

# → {"snapshot": {"status": "ready", "file_url": "https://...", ...}}

# Download — the file_url is pre-signed, no auth header needed

curl -s -o recognition_events.ndjson "<file_url>"

PUT file://recognition_events.ndjson @BONUSLY_SNAPSHOT_STAGE;

COPY INTO RECOGNITION_EVENTS_RAW (ROW_KEY, RECOMPUTED_AT, DATA)

FROM (

SELECT $1:row_key::STRING, $1:recomputed_at::TIMESTAMP_TZ, $1

FROM @BONUSLY_SNAPSHOT_STAGE

);

-- Seed the sync cursor from the newest row you just loaded

INSERT INTO SYNC_STATE

SELECT RECOMPUTED_AT, ROW_KEY

FROM RECOGNITION_EVENTS_RAW

ORDER BY RECOMPUTED_AT DESC, ROW_KEY DESC

LIMIT 1;

CREATE OR REPLACE VIEW RECOGNITION_EVENTS AS

SELECT

ROW_KEY,

DATA:bonus_id::STRING AS BONUS_ID,

DATA:recognized_at::TIMESTAMP_TZ AS RECOGNIZED_AT,

DATA:amount::NUMBER AS AMOUNT,

DATA:category::STRING AS CATEGORY,

DATA:celebration_subtype::STRING AS CELEBRATION_SUBTYPE,

DATA:giver_id::STRING AS GIVER_ID,

DATA:receiver_ids AS RECEIVER_IDS, -- array; LATERAL FLATTEN to join per-receiver

DATA:reason::STRING AS REASON,

DATA:hashtags AS HASHTAGS,

DATA:privacy::STRING AS PRIVACY,

DATA:via::STRING AS VIA,

RECOMPUTED_AT

FROM RECOGNITION_EVENTS_RAW

WHERE NOT DATA:tombstone::BOOLEAN; -- hide deleted recognitions

-- Store the PAT as a secret (rotate it here when the token is renewed)

CREATE OR REPLACE SECRET BONUSLY_PAT

TYPE = GENERIC_STRING

SECRET_STRING = '<your-pat-here>';

CREATE OR REPLACE NETWORK RULE BONUSLY_API_RULE

MODE = EGRESS TYPE = HOST_PORT VALUE_LIST = ('bonus.ly');

CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION BONUSLY_API

ALLOWED_NETWORK_RULES = (BONUSLY_API_RULE)

ALLOWED_AUTHENTICATION_SECRETS = (BONUSLY_PAT)

ENABLED = TRUE;

CREATE OR REPLACE PROCEDURE SYNC_RECOGNITION_EVENTS()

RETURNS STRING

LANGUAGE PYTHON

RUNTIME_VERSION = '3.11'

PACKAGES = ('snowflake-snowpark-python', 'requests')

HANDLER = 'main'

EXTERNAL_ACCESS_INTEGRATIONS = (BONUSLY_API)

SECRETS = ('pat' = BONUSLY_PAT)

AS

$$

import json

import requests

import _snowflake

API = "https://bonus.ly/api/public/analytics/recognition_events"

def main(session):

token = _snowflake.get_generic_secret_string("pat")

headers = {"Authorization": f"Bearer {token}"}

state = session.sql("SELECT RECOMPUTED_AT, ROW_KEY FROM SYNC_STATE LIMIT 1").collect()

params = {"page_size": 1000}

if state:

params["recomputed_at"] = state[0][0].isoformat()

params["row_key"] = state[0][1]

last, total = None, 0

while True:

resp = requests.get(API, headers=headers, params=params)

resp.raise_for_status()

body = resp.json()

rows = body["data"]

if rows:

page = session.create_dataframe(

[(r["row_key"], r["recomputed_at"], json.dumps(r)) for r in rows],

schema=["ROW_KEY", "RECOMPUTED_AT", "RAW"],

)

page.write.save_as_table("RECOGNITION_EVENTS_PAGE",

mode="overwrite", table_type="temporary")

session.sql("""

MERGE INTO RECOGNITION_EVENTS_RAW t

USING (

SELECT ROW_KEY,

RECOMPUTED_AT::TIMESTAMP_TZ AS RECOMPUTED_AT,

PARSE_JSON(RAW) AS DATA

FROM RECOGNITION_EVENTS_PAGE

) s

ON t.ROW_KEY = s.ROW_KEY

WHEN MATCHED THEN UPDATE

SET t.RECOMPUTED_AT = s.RECOMPUTED_AT, t.DATA = s.DATA

WHEN NOT MATCHED THEN INSERT (ROW_KEY, RECOMPUTED_AT, DATA)

VALUES (s.ROW_KEY, s.RECOMPUTED_AT, s.DATA)

""").collect()

last = rows[-1]

total += len(rows)

if not body["meta"]["has_more"]:

break

cursor = body["meta"]["next_cursor"]

params["recomputed_at"] = cursor["recomputed_at"]

params["row_key"] = cursor["row_key"]

if last:

session.sql("DELETE FROM SYNC_STATE").collect()

session.sql(

"INSERT INTO SYNC_STATE VALUES (?::TIMESTAMP_TZ, ?)",

params=[last["recomputed_at"], last["row_key"]],

).collect()

return f"Synced {total} rows"

$$;

CREATE OR REPLACE TASK SYNC_BONUSLY_RECOGNITION_EVENTS

WAREHOUSE = MY_WAREHOUSE

SCHEDULE = 'USING CRON 0 6 * * * UTC'

AS

CALL SYNC_RECOGNITION_EVENTS();

ALTER TASK SYNC_BONUSLY_RECOGNITION_EVENTS RESUME;

let

BaseUrl = "https://bonus.ly/api/public",

Token = "your-pat-here",

GetPage = (cursor as nullable record) as record =>

Json.Document(

Web.Contents(BaseUrl, [

RelativePath = "analytics/recognition_events",

Query = [page_size = "1000"]

& (if cursor = null then []

else [recomputed_at = cursor[recomputed_at],

row_key = cursor[row_key]]),

Headers = [Authorization = "Bearer " & Token]

])

),

AllPages = List.Generate(

() => GetPage(null),

each _ <> null,

each if _[meta][has_more] then GetPage(_[meta][next_cursor]) else null,

each _[data]

),

Rows = List.Combine(AllPages),

AsTable = Table.FromList(Rows, Splitter.SplitByNothing(), {"Record"}),

Expanded = Table.ExpandRecordColumn(AsTable, "Record",

{"row_key", "bonus_id", "recognized_at", "amount", "category",

"celebration_subtype", "giver_id", "receiver_ids", "reason",

"hashtags", "privacy", "via", "tombstone", "recomputed_at"}),

Typed = Table.TransformColumnTypes(Expanded, {

{"recognized_at", type datetimezone},

{"recomputed_at", type datetimezone},

{"amount", Int64.Type},

{"tombstone", type logical}

}),

Live = Table.SelectRows(Typed, each not [tombstone]) // hide deleted recognitions

in

Live

let

Source = Lines.FromBinary(

File.Contents("C:\data\recognition_events.ndjson"), null, null, 65001),

Rows = List.Transform(Source, each Json.Document(_)),

AsTable = Table.FromList(Rows, Splitter.SplitByNothing(), {"Record"})

// expand the same columns as the main query

in

AsTable

AllPages = List.Generate(

() => GetPage([recomputed_at = "2026-06-12T15:17:50.000Z",

row_key = "C6DEF...E69"]),

each _ <> null,

each if _[meta][has_more] then GetPage(_[meta][next_cursor]) else null,

each _[data]

),

Combined = Table.Combine({SnapshotTable, DeltaTable}),

Sorted = Table.Sort(Combined, {{"recomputed_at", Order.Descending}}),

Deduped = Table.Distinct(Sorted, {"row_key"}),

Live = Table.SelectRows(Deduped, each not [tombstone])