Skip to main content

DMI Connector (Technical Reference)

Written by Philippe Trussart

Overview

This connector ingests DMI objects into your data lake. Schemas are discovered dynamically at runtime from DMI’s metadata, so you receive whatever columns currently exist for each table, with types mapped automatically.

For each table, the connector outputs:

  • All source columns (discovered dynamically; column names are lower-cased in the output).

  • Any unique keys used for record identity.

  • Any indexed fields used to slice/iterate data efficiently.

  • A standard metadata column dmi_last_updated_at indicating when the row was written.

Authentication & Setup

Provide the following settings:

  • API key (for DMI Data Sync API)

  • Environment URL

    • Default: di.data-management.com

    • Optional: “custom” + Custom URL

  • Object storage credentials (used to fetch bulk export files)

    • client_id and client_secret for the configured object storage (DigitalOcean Spaces)

    • Region: nyc3

    • Endpoint: https://nyc3.digitaloceanspaces.com

    • Bucket (read/delete during sync): prod-bytespree-dmi-data-loading

Sync Strategy:

Dynamic schema discovery

Before syncing a table, the connector calls DMI to enumerate columns and data types and builds the destination schema. Numeric types are mapped to integer/float; everything else to string.

Incremental windowing

  • Each stream uses a rolling greater-than filter on a last-modified timestamp.

  • The starting point is the stream’s bookmark last_started.

    • On first run (or if an invalid bookmark is found) the start is set to 1971-01-01 01:00:00Z and marked as a full replace run.

  • A 5-minute safety offset is applied (start time is moved back by 5 minutes).

  • All timestamps are evaluated in America/New_York.

Bulk Exports

High-volume tables are synced via DMI’s bulk export workflow:

  1. Create export

    • POST /api/v5/export with:
      ​table, api_key, greater_than = ISO timestamp from the incremental window.

    • Capture the returned export_id.

  2. Poll status & stream intermediate files

    • GET /api/v5/export/{export_id}?api_key=... every ~30s until status is no longer pending/in_progress.

    • While the export is running, the response includes s3_object_keys.

    • The connector processes all but the last key during in-progress polling (the last file may still be growing).

    • Already processed keys are tracked so each object is read exactly once.

  3. Download & process files

    • Files are fetched from DigitalOcean Spaces (bucket above).

    • Each object is a CSV:

      • Header row is parsed; each subsequent line is mapped to a record by header name.

      • Empty lines are ignored.

      • On incremental runs, a hard delete is emitted before writing the new record (key selection described below).

    • After successful processing, the object is deleted from the bucket.

  4. Finalize & close export

    • Once status indicates all files are ready, any remaining file is processed.

    • POST /api/v5/export/{export_id}/close is called to close the export.

  5. State

    • At the end of each stream, the connector writes the bookmark {stream: {last_started: <run start ISO8601>}}.

Deletes

  • During incremental runs, incoming records are preceded by a hard delete for the same key (idempotent backfill safety):

    • Default key: identity

    • For the special INTEGRATIONS_DELETE_LOG stream, delete key is dmi_id and deletes are propagated to the referenced downstream table using identity.

  • Some reference tables are always full-replace (see list below).

Retries

Network/API requests automatically retry for ~20 minutes with ~65s back-off between attempts.

Maintenance window protection

If the connector starts while DMI maintenance is underway (Saturdays between roughly 00:00–02:00 Eastern), it sleeps and rechecks every 10 minutes until the window passes.

Notes & Limitations

  • Dynamic columns: the connector mirrors columns exposed by DMI at sync time. If DMI adds/removes fields, the connector reflects those changes on the next run.

  • Full-replace tables: labeled above; these are always re-fetched completely.

  • Temporary partial files: during a bulk export, the most recent object key is skipped until the export finishes (to avoid reading files still being written).

  • Debug logging: when enabled, a temporary table name is logged and used internally to stage results for troubleshooting.

Did this answer your question?