Pipeline for agent status data

definitions/assertions/agent_status_data_freshness.sqlx

@@ -0,0 +1,9 @@
+config { 
+    type: "assertion",
+    tags: ["agent_status"] }
+
+select * from(
+select max(timestamp) max_timestamp
+from ${ref("pphe_five9_raw","agent_status_*")}
+where _TABLE_SUFFIX = FORMAT_DATE('%Y%m%d',current_date("Europe/Amsterdam")))
+where TIMESTAMP_DIFF(max_timestamp, TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY), hour) < 0

definitions/sources/agent_status.sqlx

@@ -0,0 +1,6 @@
+config {
+      type: "declaration",
+      schema: "pphe_five9_raw",
+      name: "agent_status_*",
+      description: ""
+    }

definitions/staging/stg_agent_status.sqlx

@@ -0,0 +1,30 @@
+config {
+  type: "incremental",  
+  uniqueKey: ["AGENT_ID", "TIMESTAMP"],
+  schema: "pphe_five9_stg",  
+  assertions: {
+    uniqueKey: ["AGENT_ID", "TIMESTAMP"],
+    nonNull: ["AGENT_ID", "TIMESTAMP"]
+  },
+  dependencies: ["agent_status_data_freshness"],
+  description: "All history data for agent status"
+}
+
+--Numerates by key for deduplication
+with agent_status_deduplicated as (
+select  *,
+        ROW_NUMBER()
+          OVER (PARTITION BY AGENT_ID, TIMESTAMP)
+          row_number
+from ${ref("pphe_five9_raw","agent_status_*")}
+${ when(incremental(), `where _TABLE_SUFFIX = FORMAT_DATE('%Y%m%d',current_date("Europe/Amsterdam"))`) }) --last date table is taken
+select  AGENT_ID,
+        AGENT_NAME,
+        AGENT_START_DATE,
+        AGENT_GROUP,
+        STATE,
+        AGENT_STATE_TIME,
+        TIMESTAMP,
+        DATE
+from agent_status_deduplicated
+where row_number = 1