Data import from connectors

A package often needs data that already lives elsewhere: clients and time in awork, properties in an AMS. Nobody writes an importer for that any more. The connector says what can be read, the package says where it goes, and the platform runs it. That works on every instance, with or without a module service.

For administrators

Under System › Modules › Data import you find every connector an installed package names. The same card is on the connector’s own page under Set up AI › Connectors and takes you straight to its mapping; Connector settings leads back.

• Overview: one block per importer: what goes where, in the order of a run, plus starting an import, syncing now and its automatic sync. A connector can have several importers, for example master data once a day and times every quarter of an hour. Each has its own clock and history; while one runs, the next one waits.
• Mapping: the importers and their mappings at the top; add an importer there, rename it, or move a mapping into another one. Below, the source’s fields with real values on the left, the entity’s fields on the right. Drag from the dot onto a target field, or click the source and then the target. Suggest maps by names and values, Suggest with AI adds to that; the AI sees field names and short value lists such as a status, no record contents. The preview shows for every sample row what would be written, and writes nothing.
• Runs: what each run did. A stopped run can continue, an import can be taken back; the records it created then disappear.
• Open items: rows that could not be taken, and fields changed on both sides. Rows with the same cause stand together. Fix the cause and they go away in the next run.

Every saved mapping is a version. A running import keeps the mapping it started with.

Who owns a field

Setting	Meaning
Source wins	Every run writes the source’s value.
Protected	The source writes as long as nobody changed it here. If both change, you decide under Open items.
Only when empty	Written only into an empty field.
Only on create	After that the field is yours.

What an import does not do

• It removes nothing because an answer is empty. If a suspicious amount disappears at once, you decide.
• It does not bring back a record someone deleted here.
• It creates nothing twice: a second run only changes what changed.

The automatic sync starts after a first import and needs the background job MODULE_CONNECTOR_SYNC, which is switched on per instance.

For module developers

The connector describes in plugin.yml, or in the package’s connector entry, what a run cannot tell from the rows. Every read action without a required path argument counts as a stream anyway.

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

The package maps, in connectors[].imports of its module.json, and says which importer runs what. Without importers everything runs in one importer:

{
  "key": "awork",
  "catalog": true,
  "sync": { "everyMinutes": 15 },
  "imports": {
    "importers": [
      { "key": "stamm", "name": "Master data", "everyMinutes": 1440 },
      { "key": "zeiten", "name": "Times", "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" }
      ]
    }]
  }
}

An expression is a path (company.name, teams[].name) or a formula with fixed functions such as round(duration / 3600, 2), weekStart(date) or coalesce(a, b). derive adds worked-out columns, group rolls rows up (sum, all, first), window reads in stretches of time, seed writes records the source has no row for, and missing: "delete" clears what the source no longer lists. after calls an operation of your service once at the end of each run of that importer, because it gets no events about single records during the run. A reference may point to a mapping of another importer.

Some actions only answer for one record at a time, such as the sampling points of one location. Then each reads the stream once per row of another:

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"

Every row read carries its parent row under as, so a mapping reads standort._id, for example. It goes one level deep. A run reads the parent stream whole when it starts (at most 10,000 rows), such a stream has no incremental, and missing: "delete" only clears up once every parent row was read. The editor’s sample reads one parent row and says so. When an action pages differently from the rest of its connector, it gets its own pagination; { type: none } switches paging off for it. For text and times there are also split(text, separator), now() and fromEpoch(ms) for timestamps in milliseconds.

The rules are in the package contract, section 2.14. Check a mapping with nxm validate; mistakes in expressions show at upload, not during a run.