無印　ハイネックカットソー　ネイビー　130

Node.js FileStreamの読み込みを一時停止しつつ、一括登録を実行する

最初のサンプル

const fs = require('fs');
const readline = require('readline')
const { Connection, TYPES } = require('tedious')
// 設定に従いデータベースへ接続
function create_connection(config) {
    
    const connection = new Connection(config)
    // Promiseをnewした時点で引数のfunctionが実行される
    const p = new Promise(function(resolve, reject) {
        connection.on('connect', err => {
            if (err) {
                reject(err)
            } else {
                resolve(connection)
            }
        });
        connection.connect()
    });
    return p
    
}
// BulkLoadの実行
function execBulkLoad(connection, rows) {
    const p = new Promise(function(resolve, reject) {
        //　BulkLoad用の設定
        const options = {}
        const bulkLoad = connection.newBulkLoad('postal_code', options, function (error, rowCount) {
            resolve('inserted ' + rowCount + ' rows');
        })
        bulkLoad.addColumn('code', TYPES.Char, { nullable: false })
        bulkLoad.addColumn('address', TYPES.NVarChar, { length: 100, nullable: false })
        rows.forEach((row) => {
            // { code: '郵便番号', address: '住所'} の形式のデータをaddRow
            bulkLoad.addRow(row)
        })
        // バルクロード実行
        connection.execBulkLoad(bulkLoad)
    });
    return p
}
async function main() {
    // データベースに接続
    const config = {
        authentication: {
            options: {
                userName: 'sa',
                password: 'P@ssw0rd'
            },
            type: 'default'
        },
        server: 'localhost',
        options: {
            database: 'sample',
            encrypt: false
        }
    }
    const connection = await create_connection(config);
    const stream = fs.createReadStream('./KEN_ALL_UTF8.CSV', 'utf8')
    const reader = readline.createInterface({ input: stream })
    let rows = []
    let rowCount = 0
    reader.on('line', async (data) => {
        // 郵便番号情報を取得
        const item = data.split(',').map((value) => { return value.replace(/^"+|"+$/g,'') })
        const code = item[2]
        const address = item[6] + item[7] + item[8]
        rows.push({code: code, address: address})
        if (rows.length == 10000) {
            const msg = await execBulkLoad(connection, rows)
            console.log(msg)
            rows = []
        }
        rowCount++
    })
    reader.on('close', async () => {
        // 登録実行
        if (rows.length) {
            const msg = await execBulkLoad(connection, rows)
            console.log(msg)
        }
        console.log(rowCount)
        connection.close()
    })
}
main()

pause / resume

    reader.on('line', async (data) => {
        // 郵便番号情報を取得
        const item = data.split(',').map((value) => { return value.replace(/^"+|"+$/g,'') })
        const code = item[2]
        const address = item[6] + item[7] + item[8]
        rows.push({code: code, address: address})
        if (rows.length == 10000) {
            reader.pause() // 読み込みを一旦停止
            const msg = await execBulkLoad(connection, rows)
            console.log(msg)
            rows = []
            reader.resume() // 読み込み再開
        }
        rowCount++
    })

await pause

    reader.on('line', async (data) => {
        // 郵便番号情報を取得
        const item = data.split(',').map((value) => { return value.replace(/^"+|"+$/g,'') })
        const code = item[2]
        const address = item[6] + item[7] + item[8]
        rows.push({code: code, address: address})
        if (rows.length == 10000) {
            await reader.pause() // 読み込みを一旦停止
            const msg = await execBulkLoad(connection, rows)
            console.log(msg)
            rows = []
            reader.resume() // 読み込み再開
        }
        rowCount++
    })

pause event

const fs = require('fs');
const readline = require('readline')
const { Connection, TYPES } = require('tedious')
// 設定に従いデータベースへ接続
function create_connection(config) {
    
    const connection = new Connection(config)
    // Promiseをnewした時点で引数のfunctionが実行される
    const p = new Promise(function(resolve, reject) {
        connection.on('connect', err => {
            if (err) {
                reject(err)
            } else {
                resolve(connection)
            }
        });
        connection.connect()
    });
    return p
    
}
// BulkLoadの実行
function execBulkLoad(connection, rows) {
    const p = new Promise(function(resolve, reject) {
        //　BulkLoad用の設定
        const options = {}
        const bulkLoad = connection.newBulkLoad('postal_code', options, function (error, rowCount) {
            resolve('inserted ' + rowCount + ' rows (input:' + rows.length+')');
        })
        bulkLoad.addColumn('code', TYPES.Char, { nullable: false })
        bulkLoad.addColumn('address', TYPES.NVarChar, { length: 100, nullable: false })
        rows.forEach((row) => {
            // { code: '郵便番号', address: '住所'} の形式のデータをaddRow
            bulkLoad.addRow(row)
        })
        // バルクロード実行
        connection.execBulkLoad(bulkLoad)
    });
    return p
}
async function main() {
    // データベースに接続
    const config = {
        authentication: {
            options: {
                userName: 'sa',
                password: 'P@ssw0rd'
            },
            type: 'default'
        },
        server: 'localhost',
        options: {
            database: 'sample',
            encrypt: false
        }
    }
    const connection = await create_connection(config);
    const stream = fs.createReadStream('./KEN_ALL_UTF8.CSV', 'utf8')
    const reader = readline.createInterface({ input: stream })
    let rows = []
    let bulkRows = []
    let rowCount = 0
    reader.on('line', (data) => {
        // 郵便番号情報を取得
        const item = data.split(',').map((value) => { return value.replace(/^"+|"+$/g,'') })
        const code = item[2]
        const address = item[6] + item[7] + item[8]
        rows.push({code: code, address: address})
        
        if (rows.length == 10000) {
            bulkRows.push(rows)
            rows = []
            reader.pause() // 読み込みを一旦停止
        }
        rowCount++
    })
    reader.on('pause', async () => {
        // reader.pauseに加え、closeイベントの前にも呼び出される
        // 登録内容が存在しない場合はスキップ
        if (bulkRows.length == 0) {
            return
        } 
        const msg = await execBulkLoad(connection, bulkRows.pop())
        console.log(msg)
        reader.resume() // 読み込み再開
    })
    reader.on('close', async () => {
        // 登録実行
        if (rows.length > 0) {
            const msg = await execBulkLoad(connection, rows)
            console.log(msg)
        }
        console.log(rowCount)
        connection.close()
    })
}
main()

fastify データベースを検索した結果をhtmlで表示する

サンプル

const fastify = require('fastify')({ logger: true })
const postal = require('./module/postal')()
fastify.register(require('point-of-view'), {
    engine: {
        ejs: require('ejs')
    }
})
fastify.get('/', async (req, reply) => {
    // 検索結果をviewに渡す
    const rows = await postal.search('銀座')
    reply.view('/views/index.ejs', { rows: rows })
})
fastify.listen(3000, '0.0.0.0', err => {
    if (err) throw err
    console.log(`server listening on ${fastify.server.address().port}`)
})

const { Connection, Request } = require('tedious')
module.exports = () => {
    // 接続情報
    const config = {
        authentication: {
            options: {
                userName: 'sa',
                password: 'P@ssw0rd'
            },
            type: 'default'
        },
        server: 'localhost',
        options: {
            database: 'sample',
            encrypt: false,
            rowCollectionOnRequestCompletion : true
        }
    }
    // 設定に従いデータベースへ接続
    this.create_connection = (config) => {
    
        const connection = new Connection(config)
        // Promiseをnewした時点で引数のfunctionが実行される
        const p = new Promise(function(resolve, reject) {
            connection.on('connect', err => {
                if (err) {
                    reject(err)
                } else {
                    resolve(connection)
                }
            })
            connection.connect()
        })
        return p
        
    }
    // SQLの実行
    this.execute = (connection, sql) => {
        const p = new Promise(function(resolve, reject) {
            const request = new Request(sql, (err, rowCount, columns) => {
                if (err) {
                    reject(err)
                    return
                }
                let rows = []
                columns.forEach(column => {
                    let row = {}
                    column.forEach(field => {
                        row[field.metadata.colName] = field.value
                    });
                    rows.push(row)
                });
                
                resolve(rows)
            })
            connection.execSql(request)
        });
        return p
    }
    // 住所の一部でデータを検索
    this.search = async (address) => {
        const connection = await this.create_connection(config)
        const rows = await this.execute(connection, "SELECT * FROM postal_code WHERE address LIKE '%" + address + "%' ORDER BY code")
        connection.close()
        return rows
    }
    return this
}

<html lang="ja">
<head>
<meta charset="UTF-8">
<title>fastifyサンプル</title>
<style>
table {
    border: solid 2px;
    border-collapse: collapse;
}
th, td {
    border: solid 2px;
    padding: 4px;
}
</style>
</head>
<body>
    <table>
        <thead>
            <tr>
                <th>郵便番号</th>
                <th>住所</th>
            </tr>
        </thead>
        <tbody>
            <% rows.forEach(row => { %>
                <tr>
                    <td><%= row.code %></td>
                    <td><%= row.address %></td>
                </tr>
            <% }) %>
        </tbody>
    </table>
</body>
</html>

fastify 入力データの検証(validation)

const fastify = require('fastify')({ logger: true })
const opts = {
    schema: {
        body: {
            type: 'object',
            required: [ // addressを必須項目に
                'address'
            ],
            properties: {
                id: { type: 'number'},
                address: { type: 'string' }
            }
        }
    }
}
fastify.post('/', opts, async (req, reply) => {
    // jsonデータはreq.bodyに設定される
    return req.body
})
fastify.listen(3000, '0.0.0.0', err => {
    if (err) throw err
    console.log(`server listening on ${fastify.server.address().port}`)
})

fastify json形式のPOSTデータ受け取り

プログラム修正

{"address":"銀座"}

const fastify = require('fastify')({ logger: true })
const postal = require('./module/postal')()
fastify.post('/', async (req, reply) => {
    // jsonデータはreq.bodyに設定される
    const address = req.body.address
    const rows = await postal.search(address)
    return rows
})
fastify.listen(3000, '0.0.0.0', err => {
    if (err) throw err
    console.log(`server listening on ${fastify.server.address().port}`)
})

fastifyでPOSTデータ受信時、FST_ERR_CTP_INVALID_MEDIA_TYPE

最初の実装

const fastify = require('fastify')({ logger: true })
const postal = require('./module/postal')()
/*
fastify.get('/', async (req, reply) => {
    // クエリーパラメーター取得
    const address = req.query.address
    const rows = await postal.search(address)
    return rows
})
*/
// postに変更
fastify.post('/', async (req, reply) => {
    const address = req.query.address
    const rows = await postal.search(address)
    return rows
})
fastify.listen(3000, '0.0.0.0', err => {
    if (err) throw err
    console.log(`server listening on ${fastify.server.address().port}`)
})

POST対応

const fastify = require('fastify')({ logger: true })
const postal = require('./module/postal')()
// POST対応
fastify.register(require('fastify-formbody'))
// postに変更
fastify.post('/', async (req, reply) => {
    // req.bodyからPOSTデータを取得するよう変更
    //const address = req.query.address
    const address = req.body.address
    const rows = await postal.search(address)
    return rows
})
fastify.listen(3000, '0.0.0.0', err => {
    if (err) throw err
    console.log(`server listening on ${fastify.server.address().port}`)
})