init data-connector

2025-04-10 17:32:06 +02:00
parent 52c1a14609
commit 46631db5b2
105 changed files with 8081 additions and 338 deletions
@@ -0,0 +1,63 @@
+id: avic-etl
+namespace: dev
+description: Run Avicenna ETL
+
+labels:
+  env: dev
+  project: avicenna
+
+triggers:
+  - id: schedule
+    type: io.kestra.plugin.core.trigger.Schedule
+    cron: "*/5 * * * *"
+
+tasks:
+  - id: getPartitions
+    type: io.kestra.plugin.scripts.python.Script
+    taskRunner:
+      type: io.kestra.plugin.scripts.runner.docker.Docker
+    containerImage: ghcr.io/kestra-io/pydata:latest
+    beforeCommands:
+      - pip install -U arrow
+    script: |
+      from kestra import Kestra
+      import arrow
+      partitions = []
+
+      now = arrow.utcnow() #.shift(months=-1)
+      partitions.append("from={}&to={}".format(now.span('month')[0].format('YYYY-MM-DD'), now.shift(days=14).format('YYYY-MM-DD')))
+      lastMonth = now.shift(months=-1)
+      partitions.append("from={}&to={}".format(lastMonth.span('month')[0].format('YYYY-MM-DD'), lastMonth.span('month')[-1].format('YYYY-MM-DD')))
+      monthBeforeLast = lastMonth.shift(months=-1)
+      partitions.append("from={}&to={}".format(monthBeforeLast.span('month')[0].format('YYYY-MM-DD'), monthBeforeLast.span('month')[-1].format('YYYY-MM-DD')))
+      monthBeforeLast = monthBeforeLast.shift(months=-1)
+      partitions.append("from={}&to={}".format(monthBeforeLast.span('month')[0].format('YYYY-MM-DD'), monthBeforeLast.span('month')[-1].format('YYYY-MM-DD')))
+      Kestra.outputs({'partitions': partitions})
+  - id: loadThreeMonths
+    type: io.kestra.plugin.core.flow.EachSequential
+    value: '{{ outputs.getPartitions.vars.partitions }}'
+    tasks:
+      - id: "request"
+        type: "io.kestra.plugin.core.http.Request"
+        uri: "http://avicenna_data:3000/etl?{{ taskrun.value }}"
+        headers: 
+          user-agent: "kestra-io"
+        options:
+          readTimeout: "PT180S"
+        method: "GET"
+  # - id: checkForAnomalies
+  #   type: "io.kestra.plugin.core.http.Request"
+  #   uri: "http://avicenna_avicenna-data:3000/anomalies?from=2024-07-01&to=2024-07-31"
+  #   headers: 
+  #     user-agent: "kestra-io"
+  #   options:
+  #     readTimeout: "PT90S"
+  #   method: "GET"
+  # - id: updateTickets
+  #   type: "io.kestra.plugin.core.http.Request"
+  #   uri: "http://avicenna_avicenna-data:3000/tickets?from=2024-07-01&to=2024-07-31"
+  #   headers: 
+  #     user-agent: "kestra-io"
+  #   options:
+  #     readTimeout: "PT90S"
+  #   method: "GET"
@@ -0,0 +1,55 @@
+id: avic-tickets
+namespace: dev
+description: Run Avicenna Ticket Creation
+
+labels:
+  env: dev
+  project: avicenna
+
+triggers:
+  - id: schedule
+    type: io.kestra.plugin.core.trigger.Schedule
+    cron: "*/5 * * * *"
+
+tasks:
+  - id: getPartitions
+    type: io.kestra.plugin.scripts.python.Script
+    taskRunner:
+      type: io.kestra.plugin.scripts.runner.docker.Docker
+    containerImage: ghcr.io/kestra-io/pydata:latest
+    beforeCommands:
+      - pip install -U arrow
+    script: |
+      from kestra import Kestra
+      import arrow
+      partitions = []
+
+      now = arrow.utcnow() #.shift(months=-1)
+      partitions.append("from={}&to={}".format(now.span('month')[0].format('YYYY-MM-DD'), now.format('YYYY-MM-DD')))
+      lastMonth = now.shift(months=-1)
+      partitions.append("from={}&to={}".format(lastMonth.span('month')[0].format('YYYY-MM-DD'), lastMonth.span('month')[-1].format('YYYY-MM-DD')))
+      monthBeforeLast = lastMonth.shift(months=-1)
+      partitions.append("from={}&to={}".format(monthBeforeLast.span('month')[0].format('YYYY-MM-DD'), monthBeforeLast.span('month')[-1].format('YYYY-MM-DD')))
+      monthBeforeLast = monthBeforeLast.shift(months=-1)
+      partitions.append("from={}&to={}".format(monthBeforeLast.span('month')[0].format('YYYY-MM-DD'), monthBeforeLast.span('month')[-1].format('YYYY-MM-DD')))
+      Kestra.outputs({'partitions': partitions})
+  - id: loadThreeMonths
+    type: io.kestra.plugin.core.flow.EachSequential
+    value: '{{ outputs.getPartitions.vars.partitions }}'
+    tasks:
+      - id: updateTickets
+        type: "io.kestra.plugin.core.http.Request"
+        uri: "http://avicenna_data:3000/tickets?{{taskrun.value}}"
+        headers: 
+          user-agent: "kestra-io"
+        options:
+          readTimeout: "PT90S"
+        method: "GET"
+      - id: checkForAnomalies
+        type: "io.kestra.plugin.core.http.Request"
+        uri: "http://avicenna_data:3000/anomalies?{{ taskrun.value }}"
+        headers: 
+          user-agent: "kestra-io"
+        options:
+          readTimeout: "PT90S"
+        method: "GET"