# Data import from connectors

> Data from a connected system comes into a package's entities without a service of its own, with mapping, preview, field ownership and sync.

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.

```yaml
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:

```json
{
  "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:

```yaml
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.