DB バッチ書き込み

chunks(100) で最大100件ずつにまとめ、各書き込みの完了を待って次のバッチへ進みます。時間でも区切りたい場合は bufferTime を使います。

import { bufferTime, chunks, from } from "mogami-stream";

export async function writeBatches<T>(
  rows: AsyncIterable<T>,
  insertMany: (batch: T[], signal: AbortSignal) => Promise<void>,
  signal: AbortSignal,
) {
  await from(rows)
    .pipe(chunks(100))
    .forEach((batch, { signal }) => insertMany(batch, signal), { signal });
}

export async function writeTimedBatches<T>(
  rows: AsyncIterable<T>,
  insertMany: (batch: T[], signal: AbortSignal) => Promise<void>,
  signal: AbortSignal,
) {
  await from(rows)
    .pipe(bufferTime(1_000, { maxItems: 100 }))
    .forEach((batch, { signal }) => insertMany(batch, signal), { signal });
}

正常終了時は端数のバッチも書き込みます。失敗やキャンセル時は未完了のバッチを出力しません。

chunks · bufferTime

DB バッチ書き込み ​

DB バッチ書き込み