99 lines
3.8 KiB
YAML
99 lines
3.8 KiB
YAML
id: main_sync_workflow
|
|
namespace: ceo.genius.etl
|
|
|
|
tasks:
|
|
# Schritt 1: Definiere alle Entitäten und berechne die Zeiträume für die Synchronisation
|
|
- id: setup_parameters
|
|
type: io.kestra.plugin.scripts.python.Script
|
|
containerImage: python:3.11-slim
|
|
beforeCommands:
|
|
- "pip install python-dateutil"
|
|
- "pip install kestra"
|
|
script: |
|
|
from kestra import Kestra
|
|
import json
|
|
from datetime import date
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
today = date.today()
|
|
|
|
# 1. Definiere die Zeiträume: aktueller Monat + die letzten 3 vollen Monate
|
|
date_ranges = []
|
|
# Aktueller Monat (vom 1. bis heute)
|
|
date_ranges.append({
|
|
"start": today.replace(day=1).strftime("%Y-%m-%d"),
|
|
"end": today.strftime("%Y-%m-%d")
|
|
})
|
|
# Letzte 3 volle Monate
|
|
for i in range(1, 4):
|
|
month = today - relativedelta(months=i)
|
|
start_of_month = month.replace(day=1)
|
|
end_of_month = (start_of_month + relativedelta(months=1)) - relativedelta(days=1)
|
|
date_ranges.append({
|
|
"start": start_of_month.strftime("%Y-%m-%d"),
|
|
"end": end_of_month.strftime("%Y-%m-%d")
|
|
})
|
|
|
|
# 2. Definiere die Entitäten mit ihrem Schema.
|
|
# HIER IST DER ZENTRALE ORT, UM NEUE ENTITÄTEN HINZUZUFÜGEN ODER ZU ÄNDERN.
|
|
config = {
|
|
"static_entities": [
|
|
{
|
|
"name": "patient",
|
|
"db_table": "PatientSource",
|
|
"columns": ["birthday", "city", "id", "\"jobId\"", "\"kkId\"", "krankenkasse", "name", "street", "surname", "zip"]
|
|
}
|
|
# FÜGE HIER WEITERE STATISCHE ENTITÄTEN HINZU
|
|
# {
|
|
# "name": "neue_entitaet",
|
|
# "db_table": "NeueEntitaetSource",
|
|
# "columns": ["spalte1", "spalte2"]
|
|
# }
|
|
],
|
|
"dynamic_entities": [
|
|
{
|
|
"name": "attendanceRegistration",
|
|
"db_table": "AttendanceRegistrationSource",
|
|
"columns": ["\"attendanceRegistrationId\"", "\"dateTime\"", "\"employeeId\"", "event", "\"jobId\"", "\"personnelNumber\""]
|
|
}
|
|
# FÜGE HIER WEITERE DYNAMISCHE ENTITÄTEN HINZU
|
|
],
|
|
"date_ranges": date_ranges
|
|
}
|
|
Kestra.outputs({'entity_config': config})
|
|
print(json.dumps(config))
|
|
|
|
- id: sync_static_entities
|
|
type: io.kestra.plugin.core.flow.EachParallel
|
|
value: "{{ outputs.setup_parameters.vars.entity_config.static_entities }}"
|
|
tasks:
|
|
- id: run_static_subflow
|
|
type: io.kestra.plugin.core.flow.Subflow
|
|
namespace: ceo.genius.etl
|
|
flowId: subflow_static_entity
|
|
inputs:
|
|
entity_name: "{{ taskrun.value.name }}"
|
|
db_table: "{{ taskrun.value.db_table }}"
|
|
columns: "{{ taskrun.value.columns }}"
|
|
|
|
- id: sync_dynamic_entities
|
|
type: io.kestra.plugin.core.flow.EachParallel
|
|
value: |
|
|
{%- set result = [] -%}
|
|
{%- for entity in outputs.setup_parameters.output.dynamic_entities -%}
|
|
{%- for range in outputs.setup_parameters.output.date_ranges -%}
|
|
{%- do result.append({ "entity": entity, "range": range }) -%}
|
|
{%- endfor -%}
|
|
{%- endfor -%}
|
|
{{ result | json }}
|
|
tasks:
|
|
- id: run_dynamic_subflow
|
|
type: io.kestra.plugin.core.flow.Subflow
|
|
namespace: ceo.genius.etl # Passe dies an, falls die Subflows in einem anderen Namespace liegen
|
|
flowId: subflow_dynamic_entity
|
|
inputs:
|
|
entity_name: "{{ taskrun.value.entity.name }}"
|
|
db_table: "{{ taskrun.value.entity.db_table }}"
|
|
columns: "{{ taskrun.value.entity.columns }}"
|
|
startDate: "{{ taskrun.value.range.start }}"
|
|
endDate: "{{ taskrun.value.range.end }}" |