avicenna/service/data-hub/workflows/synapse/bak.syn-extraction.yml

id: syn-extraction
namespace: dev
description: Run Avicenna extraction with synapse

labels:
  env: dev
  project: avicenna

# triggers:
#   - id: schedule
#     type: io.kestra.plugin.core.trigger.Schedule
#     cron: "*/5 * * * *"

tasks:
  - id: synapseConfig
    type: io.kestra.plugin.core.storage.Write
    content: |
      pipelines:
        - name: "patient"
          source:
            type: "rest_api"
            config:
              base_url: "https://avicenna.dispolive.de"
              path: "/data/patientenStammdaten"
              method: POST
              records_path: "data"
              auth:
                strategy: "dispolive"
                dispolive:
                  username: Marcel
                  password: arndt1.!
              rate_limit:
                requests_per_second: 5
                burst: 1
          destination:
            type: "csv"
            config: 
              path: "./patient.csv"
          transform:
            - strategy: keep_fields
              config:
                fields: 
                - "_id"
                - "name"
                - "surname"
                - "street"
                - "zip"
                - "city"
                - "birthday"
                - "kkId"
                - "krankenkasse"
            - strategy: rename_fields
              config:
                mapping:
                  _id: id
            - strategy: add_fields
              config:
                fields: |
                  {
                    "jobId": "{{ execution.id }}"
                  }
        - name: "attendanceRegistration"
          sync:
            strategy: "time_range_iteration"
            time_range_iteration:
              start_date: "2025-02-01"
              end_date: "2025-02-28"
              date_format: "2006-01-02"
              interval: 1d
          source:
            type: "rest_api"
            config:
              base_url: "https://app.planning.nu"
              {% raw %}path: "/avicennaambulance/api2/attendance-registration/{{.SliceStart}}"{% endraw %}
              records_path: "attendanceRegistration"
              auth:
                strategy: "header"
                header:
                  header: "authorization"
                  token: "Token ${SYN_DYFLEXIS_V2_TOKEN}"
              rate_limit:
                requests_per_second: 5
                burst: 1
          destination:
            type: "csv"
            config: 
              path: "./_output/02-2025-attendance-registrations.csv"      
  - id: run
    type: io.kestra.plugin.docker.Run
    containerImage: thedevilisdero/synapse:latest
    pullPolicy: ALWAYS
    inputFiles:
      synapse.yaml: "{{ outputs.synapseConfig.uri }}"
    commands:
      - --config
      - synapse.yaml
      - run
      - patient
    outputFiles:
      - patient.csv
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: "{{ read(outputs.run.outputFiles['patient.csv']) }}"
  - id: deleteSource
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://avicenna_db:5432/avicenna
    username: johndoe
    password: randompassword
    sql: DELETE FROM "PatientSource";
  - id: copyin
    type: io.kestra.plugin.jdbc.postgresql.CopyIn
    url: jdbc:postgresql://avicenna_db:5432/avicenna
    username: johndoe
    password: randompassword
    format: CSV
    from: "{{ outputs.run.outputFiles['patient.csv'] }}"
    table: "\"PatientSource\""
    header: true
    columns:
      - birthday
      - city
      - id
      - "\"jobId\""
      - "\"kkId\""
      - krankenkasse
      - name
      - street
      - surname
      - zip
  # - id: loadStatic
  #   type: io.kestra.plugin.core.flow.EachSequential
  #   value: '{{ outputs.getPartitions.vars.partitions }}'
  # - 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: loadTours
  #       type: "io.kestra.plugin.docker.Run"
  #       containerImage: thedevilisdero/synapse:latest
  #     - 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"