V1.4.20

api/specs/sling_platform.yaml

@@ -2,7 +2,7 @@ name: sling-platform
 
 defaults:
   state:
-    base_url: https://api.slingdata.io
+    base_url: '{ coalesce(secrets.base_url, "https://api.slingdata.io") }'
 
   request:
     headers:
@@ -22,9 +22,9 @@ endpoints:
         primary_key: [name]
 
       processors:
-        - expression: date_format(now(), "%Y")
+        - expression: int_parse(date_format(now(), "%Y"))
           output: record.loaded_year
-        - expression: log(record)
+        # - expression: log(record)
 
   executions:
 
@@ -42,7 +42,7 @@ endpoints:
   executions_many:
 
     state:
-      start_date: '{ coalesce(env.START_DATE, state.max_start_time, "2025-01-01") }'
+      start_date: '{ coalesce(env.START_DATE, sync.max_start_time, "2025-01-01T00:00:00.000000Z") }'
       end_date: '{ coalesce(env.END_DATE, date_format(now(), "%Y-%m-%dT%H:%M:%S.%fZ")) }'
 
     # external sync on start/end of replication. null on first run.
@@ -52,7 +52,8 @@ endpoints:
       url: '{state.base_url}/execution/list'
 
       parameters:
-        limit: 500
+        limit: 100
+        order: asc
         filters: >
           {"period":["{state.start_date}","{state.end_date}"]}
 
@@ -67,7 +68,6 @@ endpoints:
             start_time: start_time
           }
         primary_key: [exec_id]
-        # limit: 3
 
       processors:
         - aggregation: maximum
@@ -78,14 +78,11 @@ endpoints:
             )
           output: state.max_start_time
 
-      # pagination:
-      #   # stop if condition is met
-      #   stop_condition: >
-      #     length(response.records) == 0 || 
-      #     state.max_start_time == state.start_date
+    pagination:
+      # stop if condition is met
+      stop_condition: length(response.records) == 0
 
-      #   # set state for next page
-      #   next_state:
-      #     start_date: state.max_start_time
-      #     end_date: date_format(now(), "%Y-%m-%dT%H:%M:%S.%fZ")
+      # set state for next page
+      next_state:
+        start_date: '{state.max_start_time}'
 

api/specs/stripe.yaml

@@ -509,7 +509,7 @@ endpoints:
     response:
       records:
         jmespath: "data[]"
-        primary_key: ["id"]
+        primary_key: [ subscription_id, id ]
 
       processors:
         - expression: "state.subscription_id"
@@ -521,7 +521,7 @@ endpoints:
     docs: https://docs.stripe.com/api/events
 
     state:
-      starting_after: '{ sync.last_id }'  # will be null on first run
+      ending_before: '{ sync.last_id }'  # will be null on first run
       created_gte: '{coalesce(env.CREATED_GTE, date_format(date_add(now(), -30, "day"), "%s"))}'
 
     sync: [ last_id ]
@@ -530,13 +530,20 @@ endpoints:
       url: '{state.base_url}/events'
       parameters:
         limit: '{state.limit}'
+        ending_before: '{state.ending_before}'
         starting_after: '{state.starting_after}'
         # created[gte]: '{state.created_gte}'
         # types: customer.subscription.created,customer.subscription.updated,customer.subscription.deleted
 
+    # Strange pagination logic:
+    # on first run (no `sync.last_id` value), the most recent event is first,
+    # so we have to use `starting_after` to paginate through all the records.
+    # on incremental run (with a `sync.last_id` value), in order to get the
+    # records created after the last known event ID, we have to use `ending_before`
     pagination:
       next_state:
-        starting_after: '{response.records[-1].id}'
+        starting_after: '{ is_null(sync.last_id) ? response.records[length(response.records)-1].id : nil }'
+        ending_before: '{ !is_null(sync.last_id) ? response.records[0].id : nil }'
       stop_condition: "response.json.has_more == false || length(response.records) < 1"
 
     response:
@@ -547,4 +554,4 @@ endpoints:
       processors:
         - expression: "record.id"
           output: "state.last_id"
-          aggregation: "last"
+          aggregation: "first"