AshCsvInterchange (AshCsvInterchange v0.1.0)

An Ash extension for moving CSV data in and out of Ash resources. Resources that use it expose:

csv_imports do ... end — named import types, each dispatched to a per-row Ash :create action and re-run idempotently through upserts
csv_exports do ... end — named export types, each streaming a read action's records out to CSV

Setup

Name the OTP app that owns your CSV resources, then register the domains holding them under that app:

config :ash_csv_interchange, otp_app: :my_app
config :my_app, AshCsvInterchange, domains: [MyApp.Domain]

otp_app is required. It names the app whose config holds the domain registry. Without it, type discovery raises rather than silently finding nothing.

Add the extension to each resource:

defmodule MyApp.Contact do
  use Ash.Resource,
    extensions: [AshCsvInterchange]

  csv_imports do
    csv_import :contacts do
      label "Contacts Export"
      headers required: ["external_id", "first_name"], optional: []
      upsert_action :import_from_csv
    end
  end

  # ... actions, attributes, identities ...
end

And import a CSV:

AshCsvInterchange.import_csv(:contacts, csv_binary, mode: :commit)

Summary

Types

source()

A CSV import source: an in-memory binary, a {:path, path} tuple, or a re-enumerable Enumerable of binary chunks.

Functions

csv_exports(body)

csv_imports(body)

export_csv(id, opts \\ [])

Runs an export and returns the full CSV as a single binary. For large exports prefer stream_export/2 so the whole result never lives in memory at once.

fetch_export_type(id)

Resolves a registered CSV export id to {:ok, %{resource:, type:}}, where :type is the full %AshCsvInterchange.Export.Type{} struct.

fetch_import_type(id)

Looks up a registered CSV import type by id, returning the owning resource module and the resolved %AshCsvInterchange.Import.Type{} struct.

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

Imports a CSV source against a registered type id. Resolves the owning resource via fetch_import_type/1, then delegates to AshCsvInterchange.Import.Orchestrator.import_csv/4. See its docs for the source shapes and options.

list_export_types(opts \\ [])

Lists every CSV export type registered across the configured domains.

list_import_types(opts \\ [])

Lists every CSV import type registered across the configured domains.

stream_export(id, opts \\ [])

Builds a lazy stream of CSV chunks for the named export type. The first chunk is the header row. Each later chunk is a serialised batch of records from the declared read action.

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

Streams a CSV import against a registered type id as a lazy sequence of per-row outcomes. Resolves the owning resource via fetch_import_type/1, then delegates to AshCsvInterchange.Import.Orchestrator.stream_import/4.

Types

source()

@type source() :: binary() | {:path, Path.t()} | Enumerable.t()

A CSV import source: an in-memory binary, a {:path, path} tuple, or a re-enumerable Enumerable of binary chunks.

Functions

csv_exports(body)

(macro)

csv_imports(body)

(macro)

export_csv(id, opts \\ [])

@spec export_csv(atom(), keyword()) ::
  {:ok, binary()} | {:error, AshCsvInterchange.Error.t()}

Runs an export and returns the full CSV as a single binary. For large exports prefer stream_export/2 so the whole result never lives in memory at once.

Accepts the same options as stream_export/2.

fetch_export_type(id)

@spec fetch_export_type(atom()) ::
  {:ok, %{resource: module(), type: AshCsvInterchange.Export.Type.t()}}
  | {:error, AshCsvInterchange.Error.t()}

Resolves a registered CSV export id to {:ok, %{resource:, type:}}, where :type is the full %AshCsvInterchange.Export.Type{} struct.

Returns {:error, %Error{kind: :type_not_found}} if the id isn't registered.

fetch_import_type(id)

@spec fetch_import_type(atom()) ::
  {:ok, %{resource: module(), type: AshCsvInterchange.Import.Type.t()}}
  | {:error, AshCsvInterchange.Error.t()}

Looks up a registered CSV import type by id, returning the owning resource module and the resolved %AshCsvInterchange.Import.Type{} struct.

Returns {:error, %AshCsvInterchange.Error{kind: :type_not_found}} if no such id is registered.

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

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

list_export_types(opts \\ [])

@spec list_export_types(keyword()) :: [
  %{id: atom(), label: String.t(), resource: module()}
]

Lists every CSV export type registered across the configured domains.

Pass :actor to keep only the types that actor may read. Without one, every registered type comes back.

Raises if two export entities share an :id. Import and export ids are separate namespaces, so one resource may use the same :id in both sections to mark a round-trip pair.

list_import_types(opts \\ [])

@spec list_import_types(keyword()) :: [
  %{id: atom(), label: String.t(), resource: module()}
]

Lists every CSV import type registered across the configured domains.

Walks each domain in config :my_app, AshCsvInterchange, domains: [...] with Ash.Domain.Info.resources/1 and emits one entry per csv_import entity. :my_app is the app named by config :ash_csv_interchange, otp_app: ..., which is required.

Raises if two resources declare the same :id. Type ids must be unique across the registry.

Options:

:actor — drops types whose upsert action the actor cannot perform, via Ash.can?/2. Without an actor, every registered type is returned, so pass one when discovery must be role-gated.

stream_export(id, opts \\ [])

@spec stream_export(atom(), keyword()) ::
  {:ok, Enumerable.t()} | {:error, AshCsvInterchange.Error.t()}

Builds a lazy stream of CSV chunks for the named export type. The first chunk is the header row. Each later chunk is a serialised batch of records from the declared read action.

Options:

:actor — runs the stream as this actor. The read action's policies apply.
:input — arguments for the read action, as a map. Defaults to %{}. A read action with required arguments raises its own missing-argument error once the stream is consumed.
:batch_size — page size for the underlying read. Defaults to 500.

Returns {:error, %Error{}} before any database work when the id is not registered, or when the actor cannot perform the read action. Errors raised while the stream is consumed — a calculation crash, a formatter exception — reach the consumer rather than being swallowed.

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

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

Returns {:ok, %AshCsvInterchange.Import.StreamReport{}} whose outcomes field is a lazy stream. In :commit mode, writes happen as the stream is consumed — prefer this over import_csv/3 for large files. See AshCsvInterchange.Import.Orchestrator.stream_import/4 for options.