Last active
March 24, 2025 13:12
-
-
Save rmela/a3bed669ad6194fb2d9670789541b0c7 to your computer and use it in GitHub Desktop.
SQLite query results as nodejs readable stream
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
echo 'create table foo ( name string, value int )' | sqlite3 foo.db | |
for idx in {0..1000} | |
do | |
echo "insert into foo( name, value ) values( $idx, 'abc${idx}' );" | |
done | sqlite3 foo.db |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
/* | |
* | |
* Return SQLite3 query results as a Nodejs stream, sensitive to backpressure. | |
* | |
* Assumes a foo.db file with a lot of rows - I used 1000 with a number & a string | |
*/ | |
const stream = require('stream'); | |
const sqlite = require('sqlite3'); | |
class DBStream extends stream.Readable { | |
constructor( opts ) { | |
super( { objectMode: true } ); | |
this.sql = opts.sql; | |
this.db = new sqlite.Database( opts.db ); | |
this.stmt = this.db.prepare( this.sql ); | |
this.on( 'end', () => this.stmt.finalize( () => this.db.close() )); | |
} | |
_read() { | |
let strm = this; | |
this.stmt.get( function(err,result) { | |
// If result is undefined, push null, which will end the stream. | |
/* | |
* Should have no backpressure problems, | |
* since _read is only called when the downstream is | |
* ready to fetch data | |
*/ | |
err ? | |
strm.emit('error', err ) : | |
strm.push( result || null); | |
}) | |
} | |
} | |
/* | |
* simple test | |
*/ | |
stream.pipeline( | |
[ | |
new DBStream( { sql:'select * from foo' } ), | |
new stream.Transform( { objectMode: true, transform:( data, enc, cb ) => cb( null, JSON.stringify( data ) ) } ), | |
process.stdout | |
], | |
err => { err && console.error(err); process.exit(0) } | |
) |
Thanks a lot! I took inspiration from your snipped and created my version of it:
import { ReadableTyped } from '@naturalcycles/nodejs-lib'
import { Database, Statement } from 'sqlite'
import { Readable } from 'stream'
/**
* Based on: https://gist.github.com/rmela/a3bed669ad6194fb2d9670789541b0c7
*/
export class SqliteReadable<T = any> extends Readable implements ReadableTyped<T> {
constructor(private stmt: Statement) {
super( { objectMode: true } );
// might be unnecessary
// this.on( 'end', () => {
// console.log(`SQLiteStream end`)
// void this.stmt.finalize()
// })
}
static async create<T = any>(db: Database, sql: string): Promise<SqliteReadable<T>> {
const stmt = await db.prepare(sql)
return new SqliteReadable<T>(stmt)
}
/**
* Necessary to call it, otherwise this error might occur on `db.close()`:
* SQLITE_BUSY: unable to close due to unfinalized statements or unfinished backups
*/
async close(): Promise<void> {
await this.stmt.finalize()
}
// count = 0 // use for debugging
override async _read(): Promise<void> {
// console.log(`read ${++this.count}`) // debugging
try {
const r = await this.stmt.get<T>()
this.push(r || null)
} catch(err) {
console.log(err) // todo: check if it's necessary
this.emit('error', err)
}
}
}
Thank you!
This is a nice improvement!
I'm just now starting to learn typescript.
…On Sat, Aug 21, 2021 at 9:16 AM Kirill Groshkov ***@***.***> wrote:
***@***.**** commented on this gist.
------------------------------
Thanks a lot! I took inspiration from your snipped and created my version
of it:
import { ReadableTyped } from ***@***.***/nodejs-lib'import { Database, Statement } from 'sqlite'import { Readable } from 'stream'
/** * Based on: https://gist.github.com/rmela/a3bed669ad6194fb2d9670789541b0c7 */export class SqliteReadable<T = any> extends Readable implements ReadableTyped<T> {
constructor(private stmt: Statement) {
super( { objectMode: true } );
// might be unnecessary
// this.on( 'end', () => {
// console.log(`SQLiteStream end`)
// void this.stmt.finalize()
// })
}
static async create<T = any>(db: Database, sql: string): Promise<SqliteReadable<T>> {
const stmt = await db.prepare(sql)
return new SqliteReadable<T>(stmt)
}
/** * Necessary to call it, otherwise this error might occur on `db.close()`: * SQLITE_BUSY: unable to close due to unfinalized statements or unfinished backups */
async close(): Promise<void> {
await this.stmt.finalize()
}
// count = 0 // use for debugging
override async _read(): Promise<void> {
// console.log(`read ${++this.count}`) // debugging
try {
const r = await this.stmt.get<T>()
this.push(r || null)
} catch(err) {
console.log(err) // todo: check if it's necessary
this.emit('error', err)
}
}
}
—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub
<https://gist.github.com/a3bed669ad6194fb2d9670789541b0c7#gistcomment-3867899>,
or unsubscribe
<https://github.com/notifications/unsubscribe-auth/AADP6E345UFAXXDP2S4SX6DT56RKHANCNFSM465BWTTA>
.
Triage notifications on the go with GitHub Mobile for iOS
<https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675>
or Android
<https://play.google.com/store/apps/details?id=com.github.android&utm_campaign=notification-email>
.
good stuff
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Thank you so much! I needed to stream a result to a CSV file. This does the job 👌🏻