Datenimport aus Connectoren

Ein Paket braucht oft Daten, die schon woanders stehen: Kunden und Zeiten in awork, Liegenschaften im AMS. Dafür schreibt niemand mehr einen eigenen Importer. Der Connector sagt, was sich lesen lässt, das Paket sagt, wohin es gehört, und die Plattform führt es aus. Das läuft auf jeder Instanz, auch ohne Moduldienst.

Für Administratoren

Unter System › Module › Datenimport steht jeder Connector, den ein installiertes Paket nennt. Dieselbe Karte steht auch auf der Seite des Connectors unter KI einrichten › Connectoren und führt von dort direkt zur Zuordnung; zurück geht es über Connector-Einstellungen.

• Übersicht: ein Block je Importer: was wohin geht, in der Reihenfolge eines Laufs, dazu Import starten, jetzt abgleichen und sein automatischer Abgleich. Ein Connector kann mehrere Importer haben, etwa Stammdaten einmal am Tag und Zeiten jede Viertelstunde. Jeder hat seinen eigenen Takt und seine eigene Historie; läuft einer, wartet der nächste.
• Zuordnung: oben die Importer mit ihren Zuordnungen; dort legen Sie einen Importer an, benennen ihn um oder verschieben eine Zuordnung in einen anderen. Darunter links die Felder der Quelle mit echten Werten, rechts die Felder des Datenobjekts. Ziehen Sie vom Punkt auf ein Zielfeld, oder klicken Sie erst die Quelle und dann das Ziel. Vorschlagen ordnet nach Namen und Werten zu, Mit KI vorschlagen ergänzt das; die KI sieht dafür Feldnamen und kurze Wertelisten wie einen Status, keine Inhalte von Datensätzen. Die Vorschau zeigt für jede Beispielzeile, was geschrieben würde, geschrieben wird dabei nichts.
• Läufe: was jeder Lauf getan hat. Ein abgebrochener Lauf lässt sich fortsetzen, ein Import zurücknehmen; dann verschwinden die Datensätze, die er angelegt hat.
• Offene Punkte: Zeilen, die nicht übernommen werden konnten, und Felder, die auf beiden Seiten geändert wurden. Zeilen mit derselben Ursache stehen zusammen. Lösen Sie die Ursache, gehen sie beim nächsten Lauf von selbst weg.

Jede gespeicherte Zuordnung ist eine Version. Ein laufender Import behält die Zuordnung, mit der er begonnen hat.

Wem ein Feld gehört

Einstellung	Bedeutung
Quelle gewinnt	Jeder Lauf schreibt den Wert der Quelle.
Geschützt	Die Quelle schreibt, solange hier niemand geändert hat. Ändern beide, entscheiden Sie unter Offene Punkte.
Nur wenn leer	Geschrieben wird nur in ein leeres Feld.
Nur beim Anlegen	Danach gehört das Feld Ihnen.

Was ein Import nicht tut

• Er löscht nichts, weil eine Antwort leer ist. Verschwindet auffällig viel auf einmal, entscheiden Sie.
• Er holt keinen Datensatz zurück, den jemand hier gelöscht hat.
• Er legt nichts doppelt an: Ein zweiter Lauf ändert nur, was sich geändert hat.

Der automatische Abgleich läuft erst nach einem ersten Import und braucht den Hintergrunddienst MODULE_CONNECTOR_SYNC, der pro Instanz eingeschaltet wird.

Für Modulentwickler

Der Connector beschreibt in plugin.yml oder im Connector-Eintrag des Pakets, was ein Lauf nicht aus den Zeilen ablesen kann. Jede lesende Aktion ohne Pflicht-Pfad zählt auch ohne Beschreibung als Datenstrom.

streams:
  - key: projects
    action: projects
    incremental: { arg: filterby, template: "updatedOn ge datetime'{since}'" }
    enrich:
      - { as: members, action: project_members, args: { projectId: id } }

Das Paket ordnet zu, in connectors[].imports seiner module.json, und sagt, welcher Importer was ausführt. Ohne importers läuft alles in einem Importer:

{
  "key": "awork",
  "catalog": true,
  "sync": { "everyMinutes": 15 },
  "imports": {
    "importers": [
      { "key": "stamm", "name": "Stammdaten", "everyMinutes": 1440 },
      { "key": "zeiten", "name": "Zeiten", "everyMinutes": 15, "after": { "service": "lagedienst", "operation": "awork" } }
    ],
    "mappings": [{
      "key": "projekte",
      "importer": "stamm",
      "stream": "projects",
      "entity": "projekt",
      "match": [{ "field": "awork_id", "expr": "id" }],
      "fields": [
        { "target": "awork_id", "expr": "id" },
        { "target": "titel", "expr": "coalesce(trim(name), id)" },
        { "target": "kunde", "expr": "companyId", "ref": "kunden", "refElse": "ohne-kunde" },
        { "target": "status", "expr": "projectStatus.type", "map": [{ "from": "closed", "to": "abgeschlossen" }], "default": "laufend" },
        { "target": "art", "expr": "count(retainers) > 0 ? 'retainer' : 'aufwand'", "policy": "create" }
      ]
    }]
  }
}

Ein Ausdruck ist ein Pfad (company.name, teams[].name) oder eine Formel mit festen Funktionen wie round(duration / 3600, 2), weekStart(date) oder coalesce(a, b). derive legt berechnete Spalten an, group fasst Zeilen zusammen (sum, all, first), window liest in Zeiträumen, seed schreibt Datensätze, für die die Quelle keine Zeile hat, und missing: "delete" räumt auf, was die Quelle nicht mehr nennt. after ruft einmal am Ende jedes Laufs dieses Importers eine Operation Ihres Dienstes auf, denn während des Laufs bekommt er keine Ereignisse zu einzelnen Datensätzen. Ein Verweis darf auf eine Zuordnung eines anderen Importers zeigen.

Manche Aktionen antworten nur für einen Datensatz auf einmal, etwa die Entnahmestellen einer Liegenschaft. Dann liest each den Datenstrom einmal je Zeile eines anderen:

streams:
  - key: standorte
    action: location_search
  - key: entnahmestellen
    action: sampling_point_search
    each:
      stream: standorte
      args: { locationId: "number(_id)" }
      as: standort
      filter: "_source.samplingPointsCount > 0"

Jede gelesene Zeile trägt ihre Ausgangszeile unter as, eine Zuordnung liest also zum Beispiel standort._id. Das geht nur eine Ebene tief. Ein Lauf liest den Ausgangsstrom am Anfang ganz (höchstens 10.000 Zeilen), ein solcher Datenstrom hat kein incremental, und missing: "delete" räumt erst auf, wenn alle Ausgangszeilen gelesen sind. Die Vorschau im Editor liest nur eine Ausgangszeile und sagt das dazu. Blättert eine Aktion anders als der Rest ihres Connectors, bekommt sie ihr eigenes pagination; { type: none } schaltet das Blättern für sie ab. Für Texte und Zeitpunkte gibt es außerdem split(text, trenner), now() und fromEpoch(ms) für Zeitstempel in Millisekunden.

Die Regeln stehen im Paketvertrag, Abschnitt 2.14. Prüfen Sie eine Zuordnung mit nxm validate, Fehler in Ausdrücken fallen beim Hochladen auf, nicht erst im Lauf.