Memory-bounded CSV exports from Ecto queries and enumerables.

BeamExport streams records through a lazy pipeline (source -> rows -> CSV chunks) so memory stays proportional to :chunk_size rather than to the size of the export.

• to_file/3 writes a file atomically (temporary file, sync, rename).
• to_iodata/2 collects a small export in memory.
• to_stream/2 returns a lazy stream of CSV iodata chunks.
• BeamExport.Plug serves downloads from Plug/Phoenix.

Sources

• any Enumerable (lists, ranges, streams)
• {Repo, queryable}: an Ecto query or schema. Preloads are not supported (BeamExport.PreloadError); use joins or select instead.

Records

Each record becomes one CSV row:

• with :mapper, the mapper's return value: a list (or tuple) of field values
• a list or tuple is used as-is (so Ecto select: [u.id, u.name] and select: {u.id, u.name} work without a mapper)
• a map or struct is read by :headers. Headers must be its keys: atoms, or strings naming existing atom keys (:id or "id"). A header that is not a field of a struct, or a map row that matches none of the headers, raises BeamExport.EncodeError rather than exporting blank columns. To use human-readable titles such as "Email Address", pass a :mapper.

Values

Lists, maps, tuples and other terms raise BeamExport.EncodeError: convert them explicitly in your :mapper (for example with Enum.join/2 or a JSON encoder).

Options

Unknown options raise ArgumentError, so a typo cannot silently disable a feature.

Common to every function:

• :headers - list of column titles (strings or atoms) written as the first row. Also selects the keys read from map/struct records when no :mapper is given.
• :mapper - fn record -> [field, ...] end.
• :chunk_size - rows per emitted chunk (and per file write). Default 500.
• :separator - field separator. Default ",". Must not contain ", CR or LF.
• :line_terminator - "\r\n" (default, RFC 4180) or "\n".
• :sanitize_formulas - neutralize spreadsheet formula injection. Default true. See "Security".
• :include_record_in_errors - include record contents in BeamExport.MappingError. Default false. Never affects telemetry, which is always redacted.

Ecto sources:

• :max_rows - rows fetched from the database per round trip. Default 500.
• :timeout - how long the export may hold its database connection, in milliseconds, or :infinity (default). This bounds the whole export: when it elapses, the export fails.
• :consumer_timeout - to_stream/2 only, outside a transaction: how long the database worker waits for the consumer to ask for the next batch. Default 30_000. When it elapses the cursor is closed and the consumer gets BeamExport.TimeoutError.
• :batch_timeout - to_stream/2 only, outside a transaction: how long the consumer waits for the database to deliver the next batch. Default 120_000.

to_file/3 only:

• :atomic - write to a temporary file in the target directory, sync it, then rename it over the target. Default true. With atomic: false the target is written in place: a failure after the target was opened leaves no partial file, but the previous contents are lost.
• :compress - true or :gzip to gzip the output; false to never compress. By default, compression is enabled when the path ends in .gz.
• :file_mode - permissions of the file created by the export. Default 0o600 (owner read/write only). An existing file written with atomic: false keeps its permissions.

Return values and errors

to_file/3 and to_iodata/2 return {:ok, ...} or {:error, exception}. Every error is an exception struct:

• BeamExport.MappingError - the mapper raised or returned a non-list.
• BeamExport.EncodeError - a record or value has no faithful CSV form.
• BeamExport.PreloadError - the query has preloads.
• BeamExport.TimeoutError - an Ecto stream stopped making progress.
• BeamExport.ExportError - other failures, such as a transaction that did not commit.
• File.Error / File.RenameError - filesystem errors, including errors only reported when the file is closed (for example a full disk).
• exceptions raised by your source or database driver, unchanged.

Invalid options raise ArgumentError. Exits and throws from the source are not converted into error tuples; to_file/3 still removes its partial file.

to_stream/2 is lazy: the same exceptions are raised while the stream is consumed.

An export that does not complete is never reported as successful.

Security

• Formula injection: by default, text fields starting with =, +, -, @, TAB or CR (optionally after whitespace) are prefixed with ' so spreadsheet applications display rather than evaluate them (OWASP recommendation). This changes such values, for example "-5" (a string) becomes "'-5". Numbers are never changed. Pass sanitize_formulas: false for machine-to-machine exports.
• Errors and telemetry never contain record contents unless you pass include_record_in_errors: true (which still never applies to telemetry). Exceptions raised by your own source or database driver are passed through.
• Files are created with mode 0o600. Never pass untrusted input as the path of to_file/3 or the :tmp_dir of BeamExport.Plug.
• Resources: BeamExport imposes no limit on export size or concurrency. Rate-limit export endpoints, and run large exports in background jobs.

Examples

import Ecto.Query

query = from(u in MyApp.User, where: u.active, order_by: u.id)

{:ok, %{rows: rows, bytes: bytes}} =
  BeamExport.to_file({MyApp.Repo, query}, "/var/exports/users.csv",
    headers: ["ID", "Name", "Email"],
    mapper: fn user -> [user.id, user.name, user.email] end
  )

# Map records with key headers, no mapper needed:
{:ok, iodata, _meta} =
  BeamExport.to_iodata([%{id: 1, name: "Alice"}], headers: [:id, :name])