sqlite3 readable stream

SqliteReadable.ts

import sqlite, { Statement } from 'sqlite3'
import { Readable, ReadableOptions } from 'stream'

interface Props extends ReadableOptions {
  dbPath: string
  query: string
}

export class SqliteReadable extends Readable {
  private stmt: Statement

  constructor({ dbPath, query, ...readableOptions }: Props) {
    super(readableOptions)
    const db = new sqlite.Database(dbPath)
    this.stmt = db.prepare(query)
    this.on('end', () =>
      this.stmt.finalize(() => {
        db.close()
      }),
    )
    this.on('error', console.error)
  }

  _read() {
    this.stmt.get((err, result) => {
      err ? this.emit('error', err) : this.push(result || null)
    })
  }
}