csvReader follows dbgate stream api

This commit is contained in:
Jan Prochazka 2020-06-11 10:19:38 +02:00
parent f68bdafd9f
commit dc7c44b797
2 changed files with 55 additions and 16 deletions

View File

@ -1,15 +1,54 @@
const _ = require('lodash');
const csv = require('csv');
const fs = require('fs');
const stream = require('stream');
async function csvReader({ fileName, encoding = 'utf-8', ...options }) {
class CsvPrepareStream extends stream.Transform {
constructor({ header }) {
super({ objectMode: true });
this.structure = null;
this.header = header;
}
_transform(chunk, encoding, done) {
if (this.structure) {
this.push(
_.zipObject(
this.structure.columns.map((x) => x.columnName),
chunk
)
);
done();
} else {
if (this.header) {
this.structure = { columns: chunk.map((columnName) => ({ columnName })) };
this.push(this.structure);
} else {
this.structure = { columns: chunk.map((value, index) => ({ columnName: `col${index + 1}` })) };
this.push(this.structure);
this.push(
_.zipObject(
this.structure.columns.map((x) => x.columnName),
chunk
)
);
}
done();
}
}
}
async function csvReader({ fileName, encoding = 'utf-8', header = true, delimiter, quoted }) {
console.log(`Reading file ${fileName}`);
const csvStream = csv.parse({
columns: true,
...options,
// @ts-ignore
delimiter,
quoted,
});
const fileStream = fs.createReadStream(fileName, encoding);
const csvPrepare = new CsvPrepareStream({ header });
fileStream.pipe(csvStream);
return csvStream;
csvStream.pipe(csvPrepare);
return csvPrepare;
}
module.exports = csvReader;

View File

@ -3,21 +3,21 @@ const dbgateApi = require('@dbgate/api');
async function run() {
const csvReader = await dbgateApi.csvReader({
fileName: 'test.csv',
header: true,
// header: false,
});
const tableWriter = await dbgateApi.tableWriter({
connection: {
server: 'localhost',
engine: 'mysql',
user: 'root',
password: 'test',
port: '3307',
database: 'Chinook',
},
pureName: 'importedTable'
});
// const tableWriter = await dbgateApi.tableWriter({
// connection: {
// server: 'localhost',
// engine: 'mysql',
// user: 'root',
// password: 'test',
// port: '3307',
// database: 'Chinook',
// },
// pureName: 'importedTable'
// });
const consoleWriter = await dbgateApi.consoleObjectWriter();