Csv unstable

Encodes and decodes CSV streams, with optional schema transformations.

Details

Follows RFC 4180 with a configurable delimiter. Cells are strings, and whitespace and empty cells are kept as they are. Quoted fields can contain delimiters, line breaks and doubled quotes. Decoders accept LF, CRLF and CR line endings and strip a leading BOM. Schema adapters convert cells to and from typed values.

15 exports Since v4.1.0 Source

Configuration

DecodeOptions

interface unstable

Since v4.1.0

CSV decoding options.

Details

delimiter: a single character other than a quote or line break. Defaults to ",".
maxRecordSize: the maximum record length in UTF-16 code units, including quotes and delimiters but not the record separator. Defaults to 1 MiB; Infinity disables the limit.
ignoreEmptyLines: skips records with no characters. Defaults to false, where an empty line is a record with one empty cell.
headers: uses the first record as unique field names and emits the remaining records as objects of the same width. Defaults to false.

Invalid options throw when the channel is created.

Signature

interface DecodeOptions {
  readonly delimiter?: string;
  readonly headers?: boolean;
  readonly ignoreEmptyLines?: boolean;
  readonly maxRecordSize?: number;
}

DecodeSchemaOptions

interface unstable

Since v4.1.0

Schema decoding options.

Details

By default the schema is wrapped in Schema.toCodecStringTree, so columns can use schemas such as Schema.Number, Schema.Boolean or Schema.Date. With header records, an empty cell for a property that is optional in the encoded schema is decoded as a missing key. Set disableCodecs to decode with the schema exactly as given.

Signature

interface DecodeSchemaOptions extends DecodeOptions {
  readonly disableCodecs?: boolean;
}

EncodeOptions

interface unstable

Since v4.1.0

CSV encoding options.

Details

delimiter: a single character other than a quote or line break. Defaults to ",".
newline: the record separator written after every record. Defaults to "\r\n".
columns: encodes object records by these keys, in order. Missing or undefined values become empty cells.
headers: writes columns as the first record, even for an empty stream. Requires columns.

Invalid options throw when the channel is created.

Signature

interface EncodeOptions {
  readonly columns?: readonly Array<string>;
  readonly delimiter?: string;
  readonly headers?: boolean;
  readonly newline?: "\n" | "\r\n";
}

EncodeSchemaOptions

type unstable

Since v4.1.0

Schema encoding options.

Details

By default the schema is wrapped in Schema.toCodecStringTree, so every column encodes to a string; set disableCodecs to encode with the schema exactly as given. columns defaults to the encoded keys of a struct schema and is required for other object schemas. Tuple and array schemas write cells by position and use columns only as header names. Every column must encode to a string, otherwise the channel throws when it is created.

Signature

type EncodeSchemaOptions<S extends Schema.Constraint> = Omit<EncodeOptions, "columns"> & {
  readonly columns?: S["Encoded"] extends ReadonlyArray<unknown> ? ReadonlyArray<string> : ReadonlyArray<S["Encoded"] extends infer E ? E extends unknown ? Extract<keyof E, string> : never : never>;
  readonly disableCodecs?: boolean;
}

Decoding

decode

unstable

Since v4.1.0

Creates a channel that parses UTF-8 byte chunks into CSV records.

See

decodeString

Signature

declare const decode: {
  <IE = never, Done = unknown>(options: DecodeOptions & {
    readonly headers: true;
  }): Channel<readonly [ReadonlyRecord<string, string>, ReadonlyRecord<string, string>], CsvError | IE, Done, readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], IE, Done>;
  <IE = never, Done = unknown>(options?: DecodeOptions & {
    readonly headers?: false;
  }): Channel<readonly [Row, Row], CsvError | IE, Done, readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], IE, Done>;
  <IE = never, Done = unknown>(options?: DecodeOptions): Channel<readonly [ReadonlyRecord<string, string> | Row, ReadonlyRecord<string, string> | Row], CsvError | IE, Done, readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], IE, Done>;
}

decodeSchema

unstable

Since v4.1.0

Creates a channel that parses UTF-8 CSV and decodes each record with a schema.

See

decodeSchemaString

Signature

declare function decodeSchema<S extends Constraint>(schema: S): <IE = never, Done = unknown>(options?: DecodeSchemaOptions, parseOptions?: ParseOptions) => Channel<readonly [S["Type"], S["Type"]], SchemaError | CsvError | IE, Done, readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], IE, Done, S["DecodingServices"]>

decodeSchemaString

unstable

Since v4.1.0

Creates a channel that parses CSV text and decodes each record with a schema.

Details

With headers: true the schema receives objects keyed by header name; otherwise it receives arrays of cells. Columns that do not encode to a string, such as arrays or nested structs, throw when the channel is created. Each chunk is decoded in one pass. Schema failures are SchemaErrors whose issue paths start with the record number.

See

decodeSchema
encodeSchemaString

Signature

declare function decodeSchemaString<S extends Constraint>(schema: S): <IE = never, Done = unknown>(options?: DecodeSchemaOptions, parseOptions?: ParseOptions) => Channel<readonly [S["Type"], S["Type"]], SchemaError | CsvError | IE, Done, readonly [string, string], IE, Done, S["DecodingServices"]>

Example Decoding renamed headers

import { Effect, Schema, Stream } from "effect"
import { Csv } from "effect/encoding"

const Person = Schema.Struct({
  name: Schema.String,
  age: Schema.Number
}).pipe(Schema.encodeKeys({ name: "Full Name", age: "Age (years)" }))

const people = Stream.make("Full Name,Age (years)\nAlice,32").pipe(
  Stream.pipeThroughChannel(Csv.decodeSchemaString(Person)({ headers: true })),
  Stream.runCollect
)
await Effect.runPromise(people) // => [{ name: "Alice", age: 32 }]

Example Transforming positional records into objects

import { Effect, Schema, SchemaGetter, Stream } from "effect"
import { Csv } from "effect/encoding"

const Person = Schema.Tuple([Schema.String, Schema.Number]).pipe(
  Schema.decodeTo(Schema.Struct({ name: Schema.String, age: Schema.Number }), {
    decode: SchemaGetter.transform(([name, age]) => ({ name, age })),
    encode: SchemaGetter.transform(({ name, age }) => [name, age] as const)
  })
)

const people = Stream.make("Alice,32").pipe(
  Stream.pipeThroughChannel(Csv.decodeSchemaString(Person)()),
  Stream.runCollect
)
await Effect.runPromise(people) // => [{ name: "Alice", age: 32 }]

decodeString

unstable

Since v4.1.0

Creates a channel that parses text chunks into CSV records.

Details

Records are arrays of cells, or objects keyed by the first record when headers is true. Parser state is created per channel run.

See

decode
decodeSchemaString

Signature

declare const decodeString: {
  <IE = never, Done = unknown>(options: DecodeOptions & {
    readonly headers: true;
  }): Channel<readonly [ReadonlyRecord<string, string>, ReadonlyRecord<string, string>], CsvError | IE, Done, readonly [string, string], IE, Done>;
  <IE = never, Done = unknown>(options?: DecodeOptions & {
    readonly headers?: false;
  }): Channel<readonly [Row, Row], CsvError | IE, Done, readonly [string, string], IE, Done>;
  <IE = never, Done = unknown>(options?: DecodeOptions): Channel<readonly [ReadonlyRecord<string, string> | Row, ReadonlyRecord<string, string> | Row], CsvError | IE, Done, readonly [string, string], IE, Done>;
}

Example Parsing header records

import { Effect, Stream } from "effect"
import { Csv } from "effect/encoding"

const records = Stream.make("name,city\nAlice,\"San\nFrancisco\"\n").pipe(
  Stream.pipeThroughChannel(Csv.decodeString({ headers: true })),
  Stream.runCollect
)
await Effect.runPromise(records) // => [{ name: "Alice", city: "San\nFrancisco" }]

Encoding

encode

unstable

Since v4.1.0

Creates a channel that writes CSV records as UTF-8 bytes, one byte array per input chunk.

See

encodeString

Signature

declare const encode: {
  <IE = never, Done = unknown>(options: EncodeOptions & {
    readonly columns: ReadonlyArray<string>;
  }): Channel<readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], IE, Done, readonly [ReadonlyRecord<string, string | undefined>, ReadonlyRecord<string, string | undefined>], IE, Done>;
  <IE = never, Done = unknown>(options?: EncodeOptions & {
    readonly columns?: undefined;
    readonly headers?: false;
  }): Channel<readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], IE, Done, readonly [Row, Row], IE, Done>;
  <IE = never, Done = unknown>(options?: EncodeOptions): Channel<readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], IE, Done, readonly [Row | ReadonlyRecord<string, string | undefined>, Row | ReadonlyRecord<string, string | undefined>], IE, Done>;
}

encodeSchema

unstable

Since v4.1.0

Creates a channel that encodes values with a schema and writes them as UTF-8 CSV bytes.

Gotchas

When the bytes stream an HTTP response body, for example through HttpApiSchema.StreamUint8Array, an encoding failure ends the body; it cannot change a status that was already sent.

See

encodeSchemaString

Signature

declare function encodeSchema<S extends Constraint>(schema: S): <IE = never, Done = unknown>(options?: EncodeSchemaOptions<S>) => Channel<readonly [Uint8Array<ArrayBufferLike>, Uint8Array<ArrayBufferLike>], SchemaError | IE, Done, readonly [S["Type"], S["Type"]], IE, Done, S["EncodingServices"]>

encodeSchemaString

unstable

Since v4.1.0

Creates a channel that encodes values with a schema and writes them as CSV text, one string per record.

Details

Columns default to the encoded keys of a struct schema, including keys renamed with Schema.encodeKeys. Missing or undefined values become empty cells. Schema failures are SchemaErrors whose issue paths start with the output record number, counting the header row.

See

decodeSchemaString
encodeSchema

Signature

declare function encodeSchemaString<S extends Constraint>(schema: S): <IE = never, Done = unknown>(options?: EncodeSchemaOptions<S>) => Channel<readonly [string, string], SchemaError | IE, Done, readonly [S["Type"], S["Type"]], IE, Done, S["EncodingServices"]>

Example Encoding renamed headers

import { Effect, Schema, Stream } from "effect"
import { Csv } from "effect/encoding"

const Person = Schema.Struct({
  name: Schema.String,
  age: Schema.Number,
  joined: Schema.Date
}).pipe(Schema.encodeKeys({ name: "Full Name", age: "Age (years)" }))

const csv = await Effect.runPromise(
  Stream.make({ name: "Alice", age: 32, joined: new Date(0) }).pipe(
    Stream.pipeThroughChannel(Csv.encodeSchemaString(Person)({ headers: true, newline: "\n" })),
    Stream.runCollect,
    Effect.map((parts) => parts.join(""))
  )
)
csv.split("\n") // => ["Full Name,Age (years),joined", "Alice,32,1970-01-01T00:00:00.000Z", ""]

encodeString

unstable

Since v4.1.0

Creates a channel that writes CSV records as text, one string per record.

Details

Without columns, records are arrays of cells. With columns, records are objects encoded by those keys. A non-string cell is a defect.

See

encode
encodeSchemaString

Signature

declare const encodeString: {
  <IE = never, Done = unknown>(options: EncodeOptions & {
    readonly columns: ReadonlyArray<string>;
  }): Channel<readonly [string, string], IE, Done, readonly [ReadonlyRecord<string, string | undefined>, ReadonlyRecord<string, string | undefined>], IE, Done>;
  <IE = never, Done = unknown>(options?: EncodeOptions & {
    readonly columns?: undefined;
    readonly headers?: false;
  }): Channel<readonly [string, string], IE, Done, readonly [Row, Row], IE, Done>;
  <IE = never, Done = unknown>(options?: EncodeOptions): Channel<readonly [string, string], IE, Done, readonly [Row | ReadonlyRecord<string, string | undefined>, Row | ReadonlyRecord<string, string | undefined>], IE, Done>;
}

Example Writing header records

import { Effect, Stream } from "effect"
import { Csv } from "effect/encoding"

const csv = Stream.make({ name: "Smith, Charlie", city: "London" }).pipe(
  Stream.pipeThroughChannel(Csv.encodeString({ columns: ["name", "city"], headers: true })),
  Stream.runCollect,
  Effect.map((parts) => parts.join(""))
)
await Effect.runPromise(csv) // => 'name,city\r\n"Smith, Charlie",London\r\n'

Errors

CsvError

unstable

Since v4.1.0

A CSV decoding failure.

Details

record is one-based and counts the header row and blank lines. line is the one-based physical line where the record starts. column is the one-based cell index of a syntax error, and undefined otherwise.

Signature

declare class CsvError extends YieldableError<this> & {
  readonly _tag: "CsvError";
} & Readonly<{
  readonly column?: number;
  readonly kind: "Syntax" | "RecordSize" | "Header";
  readonly line: number;
  readonly message: string;
  readonly record: number;
}> {
  constructor(args: {
    readonly column?: number;
    readonly kind: "Syntax" | "RecordSize" | "Header";
    readonly line: number;
    readonly message: string;
    readonly record: number;
  });
  readonly "~effect/encoding/Csv/CsvError": "~effect/encoding/Csv/CsvError";
}

Guards

isCsvError

unstable

Since v4.1.0

Checks whether a value is a CsvError.

Signature

declare function isCsvError(u: unknown): u is CsvError

Models

Row

type unstable

Since v4.1.0

A record as an array of string cells.

Signature

type Row = ReadonlyArray<string>