AshCsvInterchange.Import.Orchestrator (AshCsvInterchange v0.1.0)

Copy Markdown View Source

Runs the CSV import pipeline. import_csv/4 returns a bounded %RunReport{}. stream_import/4 returns lazy per-row outcomes.

Both resolve the named csv_import entity, then parse with Parser.parse_stream/2. The header is read eagerly, so encoding, malformed-CSV, empty-file, missing-header and duplicate-header errors are all fatal. Unknown columns become warnings.

Each non-blank row goes to the configured upsert action. A failing row becomes an :invalid, :errored or :crashed outcome and never aborts the run.

In :commit mode with batch_size > 1, rows are chunked and dispatched via Ash.bulk_create/4 so a batch costs one write round-trip instead of one per row. Any row a batch didn't clearly succeed on — invalid input, a data-layer error, a crash, or an aborted batch — is re-committed through the single-row path, so outcomes stay identical to batch_size: 1 regardless of why the batch didn't take it all the way.

Summary

Functions

import_csv(resource, id, source, opts \\ [])

Parses a CSV source and sends each non-blank row to the type's upsert action. Returns a bounded %RunReport{}: counts is exact, outcomes is capped at :max_outcomes, and outcomes_truncated? marks when rows exceeded the cap.

stream_import(resource, id, source, opts \\ [])

Streams a CSV import as a lazy sequence of per-row outcomes.

Functions

import_csv(resource, id, source, opts \\ [])

@spec import_csv(
  module(),
  atom(),
  AshCsvInterchange.Import.Parser.source(),
  keyword()
) ::
  {:ok, AshCsvInterchange.Import.RunReport.t()}
  | {:error, AshCsvInterchange.Error.t()}

Parses a CSV source and sends each non-blank row to the type's upsert action. Returns a bounded %RunReport{}: counts is exact, outcomes is capped at :max_outcomes, and outcomes_truncated? marks when rows exceeded the cap.

The source is a binary, a {:path, path} tuple, or a re-enumerable Enumerable of binary chunks.

Options:

  • :mode — :dry_run (default) or :commit
  • :max_outcomes — retained outcome preview size (default 100)
  • :retain_records? — keep the full Ash record on committed outcomes (default false)
  • :batch_size — commit-mode rows per Ash.bulk_create/4 dispatch (default 100). batch_size: 1 commits one row per write round-trip, identical to pre-batching behaviour. Ignored in :dry_run mode.
  • :actor, :tenant, :authorize?, :scope — forwarded to Ash.Changeset.for_create/4 for every row.

:commit mode is not atomic. Rows are written as the input is consumed, so a parse or encoding error partway through leaves earlier rows committed and still returns {:error, %Error{}}. Imports are idempotent upserts, so re-running the corrected file converges. Prefer stream_import/4 for large or untrusted input.

stream_import(resource, id, source, opts \\ [])

@spec stream_import(
  module(),
  atom(),
  AshCsvInterchange.Import.Parser.source(),
  keyword()
) ::
  {:ok, AshCsvInterchange.Import.StreamReport.t()}
  | {:error, AshCsvInterchange.Error.t()}

Streams a CSV import as a lazy sequence of per-row outcomes.

Reads the header eagerly, so fatal header and encoding errors return {:error, %Error{}} at once. Returns a %StreamReport{} whose outcomes field is a lazy stream of one %RowOutcome{} per non-blank data row. In :commit mode, writes happen as the stream is consumed.

Takes the same source shapes and options as import_csv/4. The returned stream yields outcomes one batch at a time — it never buffers more than one batch's worth of rows.