MongoDB connector for Google Data Studio

appsscript.json

{
  "exceptionLogging": "STACKDRIVER",
  "dataStudio": {
    "name": "MongoDB via mlab API - acc40",
    "company": "Account 4.0",
    "companyUrl": "https://acc40.com",
    "logoUrl": "https://next-scraft.appspot.com/images/scraft.ico",
    "addonUrl": "https://medium.com/@jsat66/mongodb-connector-for-google-data-studio-part-1-%E0%B8%AA%E0%B8%B3%E0%B8%A3%E0%B8%A7%E0%B8%88-c35eac7f2bf4",
    "supportUrl": "https://gist.github.com/jojosati/89652770b39fd147a2484e4baf13a5ee",
    "description": "Universal MongoDB connector."
  }
}

configService.gs

function configService(config) {
  var fn = { };
  
  fn.get = function (key) {
    return config[key] || '';
  }
  
  fn.eval = function (key) {
    return eval('(' + fn.get(key) + ')');     
  }
  
  fn.list = function (key) {
    return fn.get(key).split(/(?:\s*[\,\n]\s*)+/);
  }
  
  fn.pipeline = function (dateRange) {
    var query = fn.eval('query')
    
    // query can be Array (pipeline) or Object ($match stage)
    var pipeline = query;
    
    if (!query.map) {
      // convert object to $match stage in pipeline
      pipeline = [{$match: pipeline}];
    }
    
    // https://developers.google.com/datastudio/connector/date-range#getdata_behavior_when_daterangerequired_is_true
    if (dateRange && dateRange.startDate) {
      var dateRangeField = fn.get('dateRangeField');
      var sameDate = (dateRange.startDate === dateRange.endDate);
      
      var dateRangeQuery = sameDate? dateRange.startDate : {$gte: dateRange.startDate, $lte: dateRange.endDate};
      
      pipeline.unshift({$match: newObj(dateRangeField, dateRangeQuery)});
    }
        
    return pipeline;
  }

  fn.fetchData = function (pipeline) {
    // https://docs.mlab.com/data-api/#commands
    // https://mongodb.github.io/node-mongodb-native/3.3/api/Collection.html#aggregate
    var url = fn.get('mlabUrl');

    var payload = {
      aggregate: fn.get('collection'),
      pipeline: pipeline,
    };
    
    if (/api\.mlab\.com\/api\/1/.test(url)) {
      payload.cursor = {};
      payload.allowDiskUse = true;
    }
      
    var options = {
      method: 'post',
      contentType: 'application/json',
      payload: JSON.stringify(payload),
    }

    // https://developers.google.com/apps-script/reference/url-fetch/url-fetch-app#fetchurl,-params
    // return httpResponse - https://developers.google.com/apps-script/reference/url-fetch/http-response.html
    var result =  JSON.parse(UrlFetchApp.fetch(url, options).getContentText());
    
    if (result.errmsg && result.code===9) {
      // aggregate error - The 'cursor' option is required'
      // retry again
      payload.cursor = {batchSize: 1000};
      payload.allowDiskUse = true;
      options.payload = JSON.stringify(payload);
      result =  JSON.parse(UrlFetchApp.fetch(url, options).getContentText());
    }
    
    // error handling - https://developers.google.com/datastudio/connector/error-handling#user-facing-errors   
    if (!result.ok && result.errmsg) {
      console.error(result.errmsg, options);
      showError(result.errmsg);
    }
    
    // mlab result
    if (result.cursor)
      return result.cursor.firstBatch;

    // mserver result
    return result.values || [];
  }
  
  return fn;  
}

filterQuery.gs

// https://developers.google.com/datastudio/connector/reference#dimensionsfilters
// https://developers.google.com/datastudio/connector/reference#filteroperator
function isInclusive (_type) {
  return _type !== 'EXCLUDE';
}

var opsFN = {};

opsFN.EQUALS = function (filter) {
  var fieldName = filter.fieldName;
  var inclusive = isInclusive(filter.type);
  
  var values = filter.values ;
  
  if (values.length === 0) {
    return newObj(fieldName, inclusive? null : {$ne: null});
  }
  
  if (values.length === 1) {
    return newObj(fieldName, inclusive? values[0] : {$ne: values[0]});
  }
  
  return newObj(fieldName, inclusive? {$in: filter.values} : {$nin: filter.values[0]});
}

opsFN.IN_LIST = opsFN.EQUALS;
opsFN.IS_NULL = opsFN.EQUALS;
  
opsFN.CONTAINS = function(filter) {
  var fieldName = filter.fieldName;
  var inclusive = isInclusive(filter.type);
  
  var values = filter.values.map(function(v){
    return {$regex: escapeRegExp(v), $options: 'i'};
  })
  
  if (values.length === 0)
    return;
  
  if (values.length === 1)
    return newObj(fieldName, inclusive? values[0] : {$not: values[0]});
  
  return newObj(fieldName, inclusive? {$in: values} : {$nin: values[0]});
}
  
opsFN.REGEXP_PARTIAL_MATCH = function(filter) {
  var fieldName = filter.fieldName;
  var inclusive = isInclusive(filter.type);
  
  var values = filter.values.map(function(v){
    return {$regex: v};
  })
  
  if (values.length === 0)
    return;
  
  if (values.length === 1)
    return newObj(fieldName, inclusive? values[0] : {$not: values[0]});
  
  return newObj(fieldName, inclusive? {$in: values} : {$nin: values[0]});
}
  
opsFN.REGEXP_EXACT_MATCH = function(filter) {
  var fieldName = filter.fieldName;
  var inclusive = isInclusive(filter.type);
  
  var values = filter.values.map(function(v){
    return {$regex: '^' + v + '$'};
  })
  
  if (values.length === 0)
    return;
  
  if (values.length === 1)
    return newObj(fieldName, inclusive? values[0] : {$not: values[0]});
  
  return newObj(fieldName, inclusive? {$in: values} : {$nin: values[0]});
}
  
opsFN.BETWEEN = function(filter) {
  var fieldName = filter.fieldName;
  var inclusive = isInclusive(filter.type);
  
  var values = filter.values ;
  
  var qry =  {$gte: values[0], $lte: values[1]};
  
  return newObj(fieldName, inclusive? qry : {$not: qry});  
}
  
opsFN.NUMERIC_GREATER_THAN = function(filter) {
  var fieldName = filter.fieldName;
  var inclusive = isInclusive(filter.type);
  
  var values = filter.values;
  var num = +(values[0]);
  
  return newObj(fieldName, inclusive? {$gt: num} : {$lte: num});    
}

opsFN.NUMERIC_GREATER_THAN_OR_EQUAL = function(filter) {
  var fieldName = filter.fieldName;
  var inclusive = isInclusive(filter.type);
  
  var values = filter.values;
  var num = +(values[0]);
  
  return newObj(fieldName, inclusive? {$gte: num} : {$lt: num});    
}

opsFN.NUMERIC_LESS_THAN = function(filter) {
  var fieldName = filter.fieldName;
  var inclusive = isInclusive(filter.type);
  
  var values = filter.values;
  
  return newObj(fieldName, inclusive? {$lt: +(values[0])} : {$gte: +(values[0])});
}
  
opsFN.NUMERIC_LESS_THAN_OR_EQUAL = function(filter) {
  var fieldName = filter.fieldName;
  var inclusive = isInclusive(filter.type);
  
  var values = filter.values;
  
  return newObj(fieldName, inclusive? {$lte: +(values[0])} : {$gt: +(values[0])});
}

function filterQuery(dimensionsFilters) {
  var rootq = [];
  
  dimensionsFilters.forEach(function(filters) {
    var subq = [];
    
    filters.forEach(function(filter) {
      var fn = opsFN[filter.operator];
      
      if (fn) {
        var qry = fn(filter);
        
        if (qry)
          subq.push(qry);
      }
    })
    
    if (subq.length)
      rootq.push(subq.length === 1? subq[0] : {$or: subq});
  })
  
  
  if (rootq.length)
    return (rootq.length === 1)? rootq[0] : {$and: rootq};
}

getAuthType.gs

// ==== getAuthType ====
// https://developers.google.com/datastudio/connector/build#define_authentication_type_in_getauthtype
// legacy response - https://developers.google.com/datastudio/connector/reference#getauthtype
// code - legacy
function getAuthType() {
  return { type: 'NONE' };
}

getConfig.gs

// ==== getConfig ====
// https://developers.google.com/datastudio/connector/build#define_configuration_via_getconfig
// legacy response - https://developers.google.com/datastudio/connector/reference#getconfig
// steppedConfig - https://developers.google.com/datastudio/connector/stepped-configuration
// code - legacy
function getConfig(request) {
  var rconfig = request.configParams;
  var response = {};
  
  response.configParams = [
    
    {
      type: 'INFO',
      name: 'info',
      text: 'This connector use mlab API runCommmand/aggregate spec - https://docs.mlab.com/data-api/#commands',
    },
    {
      type: 'TEXTINPUT',
      name: 'mlabUrl',
      displayName: 'mLab url (runCommand API with apiKey)',
      helpText: 'e.g. https://api.mlab.com/api/1/databases/my-db/runCommand?apiKey=my-apikey',
    },
    {
      type: 'TEXTINPUT',
      name: 'collection',
      displayName: 'collection name',
      helpText: 'e.g. my-coll',
    },
    {
      type: 'TEXTAREA',
      name: 'query',
      displayName: 'embedded MongoDB query or pipeline stages (optional)',
      helpText: 'e.g. [{$match:{zone:"ASIA"}]',
    },
    {
      type: 'TEXTAREA',
      name: 'schemaFields',
      displayName: 'schema fields list (optional)',
      helpText: 'e.g. country,city,count:number',
    },
    {
      type: 'TEXTINPUT',
      name: 'dateRangeField',
      displayName: 'field to enable dateRange in report (optional).',
      helpText: 'e.g. date',
      isDynamic: true,
    }
    
  ];
    
  response.isSteppedConfig = true;
  
  if (rconfig) {
   
    response.isSteppedConfig = false;
    response.dateRangeRequired = Boolean(rconfig.dateRangeField);
    
    
    response.configParams.push(
      {
        type: 'INFO',
        name: 'validate',
        text: 'dataRangeRequired is ' + (response.dateRangeRequired? 'enabled' : 'disabled') ,
      }
    );

    var errors = []
    if (!rconfig.mlabUrl) {
      errors.push('mLab url is not defined.')
    }
    
    if (!rconfig.collection) {
      errors.push('collection name is not defined.')
    }
    
    if (rconfig.query) {
      try {
        configService(rconfig).eval('query');
      }
      catch(e) {
        errors.push('query error - ' + e.message);
      }
    }
    
    if (errors.length) {
      response.isSteppedConfig = true;
      response.configParams.push(
        {
          type: 'INFO',
          name: 'validate',
          text: errors.join('\n\n') ,
        }
      );  
    }
  }
  
  return response;
}

getData.gs

// ==== getData ====
// https://developers.google.com/datastudio/connector/build#fetch_and_return_data_with_getdata
function getData(request) {
  // https://developers.google.com/datastudio/connector/reference#request_3
  var rconfig = request.configParams;
  var cs = configService(rconfig);
  
  var pipeline = cs.pipeline(request.dateRange);
  var filtersApplied = false;
  var batchSize = 1000;
  var rowsMax = 10000;
  
  if (/mlab.com/.test(rconfig.mlabUrl)) {
    batchSize = 100;
  }
  
  // https://developers.google.com/datastudio/connector/reference#dimensionsfilters
  // https://developers.google.com/datastudio/connector/reference#filteroperator
  console.log(JSON.stringify(request));
  if (request.dimensionsFilters && request.dimensionsFilters.length) {
    console.log('dimensionsFilters', JSON.stringify(request.dimensionsFilters))
    var qry = filterQuery(request.dimensionsFilters);
    if (qry) {
      pipeline.push({$match: qry});
      filtersApplied = true;
    }
  }
    
  var projection = {};
  request.fields.forEach(function(fld) {
    if (fld.forFilterOnly && filtersApplied)
      return;
    
    projection[fld.name] = true
  })
  pipeline.push({$project: projection})
  
  var schema ;
  var fieldNames = Object.keys(projection);
  var schemaFields = schemaFromConfig(cs.list('schemaFields'))
  
  if (schemaFields.length) {
    schema = schemaFields.filter(function(scf) {
      return fieldNames.indexOf(scf.name) !== -1;
    })
  }
  
  var rows = [];
  var readMore = true;
    
  // https://developers.google.com/datastudio/connector/reference#scriptparams
  var isSample = request.scriptParams && request.scriptParams.sampleExtraction;
  var limit = isSample? 10 : batchSize;
  console.log(JSON.stringify(pipeline));
              
  while (readMore) {
    var _pipeline = pipeline.concat([]);
    
    if (rows.length)
      _pipeline.push({$skip: rows.length});
    
    _pipeline.push({$limit: limit});
    
    var _jsonData = [];
    try {
      _jsonData = cs.fetchData(_pipeline);
    }
    catch (e) {
      console.error(e)
      showError('getData fail.', e);
      break;
    }
    
    if (!schema) {
      schema = schemaFromSample(_jsonData, fieldNames);
    }
    
    var _rows = _jsonData.map(function(doc){
      
      var values = schema.map(function(scf) {
        return objResolve(doc, scf.name)
      });
      
      return {values: values};
    });
    
    rows.push.apply(rows, _rows);
    
    readMore = (_jsonData.length >= batchSize && rows.length < rowsMax);
  }
  
  console.log('total rows = ' + rows.length);   
  return {schema: schema, rows: rows, filtersApplied: filtersApplied};
}

getSchema.gs

// ==== getSchema ====
// https://developers.google.com/datastudio/connector/build#define_the_fields_with_getschema
// legacy response - https://developers.google.com/datastudio/connector/reference#getschema
function schemaFromSample(samples, _fields) {
  var names = [];
  var dataTypes = {};
  
  var doc
  for (doc in samples) {
    doc = samples[doc];
    
    if (_fields && names.length === _fields.length)
      break;
    
    ;(_fields || Object.keys(doc)).forEach(function(k){
      
      if (names.indexOf(k) !== -1)
        return;
      
      var t = objResolve(doc, k)
      if (['number', 'string', 'boolean'].indexOf(t) !== -1) {
        names.push(k);
        dataTypes[k] = t.toUpperCase();
      }
    })
  }
  
  // https://developers.google.com/datastudio/connector/reference#field
  // https://developers.google.com/datastudio/connector/semantics#semantic-type-detection
  return (_fields || names).map(function(k) {
    var fld = {name: k, dataType: dataTypes[k] || 'STRING'}
    
//    var conceptType = (dataTypes[k] === 'NUMBER') ? 'METRIC' : 'DIMENSION';
//    fld.semantics = {conceptType: conceptType};
    
    return fld;
  })
}

function schemaFromConfig (schemaFields) {
  return schemaFields.map(function(nt) {
    // split name:type
    var ntsegs = nt.split(':');
    
    var field = {name: ntsegs[0], dataType: (ntsegs[1] || 'STRING').toUpperCase()};
    
    return field;
  });
}

function getSchema(request) {
  var rconfig = request.configParams
  var cs = configService(rconfig);
  var schema = schemaFromConfig(cs.list('schemaFields'))
  
  if (schema.length)
    return {schema: schema};
  
  // dynamic schema by fetching sample data
  var pipeline = cs.pipeline(request.dateRange);
  
  pipeline.push({$limit: 10}); // limit sample
  
  var samples = cs.fetchData(pipeline);
  schema = schemaFromSample(samples)
  return {schema: schema};
}

utils.gs

// https://developers.google.com/datastudio/connector/debug
function isAdminUser() {
  return true;
}

// https://developers.google.com/datastudio/connector/error-handling#user-facing-errors
function showError(errorText, debugText) {
  var cc = DataStudioApp.createCommunityConnector();
  
  cc.newUserError()
  .setText(errorText)
  .setDebugText(debugText)
  .throwException();
}

function escapeRegExp(string) {
  return string.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); // $& means the whole matched string
}

function newObj(key, val) {
  var o = {};
  o[key] = val;
  
  return o;
}

function objResolve(obj, path) {
  return path.split(".").reduce(function(o, p) {
    return o && o[p];
  }, obj);
}