Daily Coding - Write data from an AWS Lambda call to sqlite3

In this post I will explain the process of writing data from an AWS lambda call to a local sqlite database.

The use case I have is as follows. I have an AWS lambda function that runs a scraping job to get data from a service. I'd like the lambda call to be triggered from my server machine using a cron job at certain points.

Once I get the response from the lambda function, I would like to use the data returned from the AWS Lambda call and then parse and write the data to a local sequel database using Node.js.

The following script accomplishes the same. I'm using sqlite3 and sequelize from npm.

Further as part of this demonstration, it shows the utility of using a lambda function to run parallel independent jobs. And aggregate the data once the responses are received from the function.

'use strict';
require('dotenv').config();

var AWS = require('aws-sdk');
const fs = require('fs');
const rison = require('rison');

const moment = require("moment");
const _ = require('underscore');
const async = require("async");

const uuidv4 = require('uuid/v4');

var Sequelize = require("sequelize");
var sequelize = new Sequelize('main', '', '', {
    // sqlite! now!
    dialect: 'sqlite',

    // the storage engine for sqlite
    // - default ':memory:'
    storage: './data/'+moment().format('YYYYMMDD')+'-marketwatch'+'.sqlite3',
    logging:false
});

const Option = sequelize.define('option', {
    call_ask: Sequelize.NUMBER,
    call_bid: Sequelize.NUMBER,
    call_change: Sequelize.NUMBER,
    call_last: Sequelize.NUMBER,
    call_openint: Sequelize.NUMBER,
    call_quote: Sequelize.STRING,
    call_vol: Sequelize.NUMBER,
    expiry: Sequelize.STRING,
    put_ask: Sequelize.NUMBER,
    put_bid: Sequelize.NUMBER,
    put_change: Sequelize.NUMBER,
    put_last: Sequelize.NUMBER,
    put_openint: Sequelize.NUMBER,
    put_quote: Sequelize.STRING,
    put_vol: Sequelize.NUMBER,
    strike: Sequelize.NUMBER,
    ticker: Sequelize.STRING,
    underlying: Sequelize.NUMBER,
    timestamp: Sequelize.STRING,
    timestamp_int: Sequelize.NUMBER,
    id: { type: Sequelize.STRING, primaryKey: true }
}, {

});

sequelize
    .authenticate()
    .then(function(err) {
        console.log('Connection has been established successfully.');
    }, function(err) {
        console.log('Unable to connect to the database:', err);
    });


var timestamp = moment().toISOString();
var timestamp_int = moment().unix();

async function writeResult(result) {
    let data = result.Payload;

    // remove the quotes
    data = String(data);
    data = data.substring(1, data.length - 1);
    var options = rison.decode_array(data);

    // console.log(options);

    var items = [];

    _.each(options, async function(o) {
        o.underlying = Number(o.underlying);
        o.expiry = moment(o.expiry).toDate().toISOString();
        o.timestamp = timestamp;
        o.timestamp_int = timestamp_int;
        o.id = uuidv4();

        // console.log(o);
        items.push(o);

        await Option.create(o);
    });

    // console.log(items);
}

const invokeLambda = (lambda, params) => new Promise((resolve, reject) => {
    lambda.invoke(params, (error, data) => {
        if (error) {
            reject(error);
        } else {
            resolve(data);
        }
    });
});

(async () => {

    let region = process.env.AWS_REGION;
    let functionURN = process.env.AWS_FUNCTION_URN;

    // You shouldn't hard-code your keys in production!
    // http://docs.aws.amazon.com/AWSJavaScriptSDK/guide/node-configuring.html
    AWS.config.update({
        accessKeyId: process.env.AWS_ACCESS_KEY,
        secretAccessKey: process.env.AWS_SECRET_KEY,
        region: region,
    });

    const lambda = new AWS.Lambda();

    let keywords = ['ASPS', 'FB', 'TSLA', 'OCN', 'PTON', 'TWTR', 'SPOT'];
    // keywords = ['ASPS'];
    // keywords = ['ASPS', 'FB'];

    let promises = [];

    for (let kw of keywords) {
        let event = { keyword: kw };
        let params = {
            FunctionName: functionURN,
            InvocationType: "RequestResponse",
            Payload: JSON.stringify(event),
        };
        console.log(params);
        promises.push(
            invokeLambda(lambda, params)
        )
    }

    console.log(`Invoked ${promises.length} lambda requests!`);

    var start = new Date();
    let results = await Promise.all(promises);
    var end = new Date() - start;

    console.log(`invokeLambda() in region ${region} took ${end/1000} seconds`);

    await sequelize.sync({});
    console.log("All models were synchronized successfully.");

    const initialTimestamp = new Date();

    await Promise.all(
        results.map(async result => {
            await writeResult(result);
        })
    );

    console.log(
        `All Completed! ${Number(new Date()) - Number(initialTimestamp)}ms.`
    );
})();