Configurable QueryTablesAPI to call directly postgresql using cartodb-psql
or to keep using a request to the SQL API
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
var sqlApi = require('../sql/sql_api');
|
||||
var sqlApi = require('../sql/sql_api'),
|
||||
PSQL = require('cartodb-psql');
|
||||
|
||||
function QueryTablesApi() {
|
||||
}
|
||||
@@ -33,7 +34,7 @@ QueryTablesApi.prototype.getLastUpdatedTime = function (username, api_key, table
|
||||
});
|
||||
};
|
||||
|
||||
QueryTablesApi.prototype.getAffectedTablesInQuery = function (username, api_key, sql, callback) {
|
||||
QueryTablesApi.prototype.getAffectedTablesInQuery = function (username, options, sql, callback) {
|
||||
// Replace mapnik tokens
|
||||
sql = sql
|
||||
.replace(affectedTableRegexCache.bbox, 'ST_MakeEnvelope(0,0,0,0)')
|
||||
@@ -45,20 +46,32 @@ QueryTablesApi.prototype.getAffectedTablesInQuery = function (username, api_key,
|
||||
sql = 'SELECT CDB_QueryTables($windshaft$' + sql + '$windshaft$)';
|
||||
|
||||
// call sql api
|
||||
sqlApi.query(username, api_key, sql, function(err, rows){
|
||||
if (err){
|
||||
var msg = err.message ? err.message : err;
|
||||
callback(new Error('could not fetch source tables: ' + msg));
|
||||
return;
|
||||
}
|
||||
var qtables = rows[0].cdb_querytables;
|
||||
var tableNames = qtables.split(/^\{(.*)\}$/)[1];
|
||||
tableNames = tableNames ? tableNames.split(',') : [];
|
||||
callback(null, tableNames);
|
||||
});
|
||||
if (shouldQueryPostgresDirectly()) {
|
||||
var psql = new PSQL(options);
|
||||
psql.query(sql, function(err, resultSet) {
|
||||
var rows = resultSet.rows || [];
|
||||
handleAffectedTablesInQueryRows(err, rows, callback);
|
||||
});
|
||||
} else {
|
||||
sqlApi.query(username, options.api_key, sql, function(err, rows) {
|
||||
handleAffectedTablesInQueryRows(err, rows, callback);
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
QueryTablesApi.prototype.getAffectedTablesAndLastUpdatedTime = function (username, api_key, sql, callback) {
|
||||
function handleAffectedTablesInQueryRows(err, rows, callback) {
|
||||
if (err){
|
||||
var msg = err.message ? err.message : err;
|
||||
callback(new Error('could not fetch source tables: ' + msg));
|
||||
return;
|
||||
}
|
||||
var qtables = rows[0].cdb_querytables;
|
||||
var tableNames = qtables.split(/^\{(.*)\}$/)[1];
|
||||
tableNames = tableNames ? tableNames.split(',') : [];
|
||||
callback(null, tableNames);
|
||||
}
|
||||
|
||||
QueryTablesApi.prototype.getAffectedTablesAndLastUpdatedTime = function (username, options, sql, callback) {
|
||||
sql = sql
|
||||
.replace(affectedTableRegexCache.bbox, 'ST_MakeEnvelope(0,0,0,0)')
|
||||
.replace(affectedTableRegexCache.pixel_width, '1')
|
||||
@@ -72,23 +85,42 @@ QueryTablesApi.prototype.getAffectedTablesAndLastUpdatedTime = function (usernam
|
||||
'WHERE m.tabname = any ((SELECT tablenames from querytables)::regclass[])'
|
||||
].join(' ');
|
||||
|
||||
sqlApi.query(username, api_key, query, function(err, rows){
|
||||
if (err || rows.length === 0) {
|
||||
var msg = err.message ? err.message : err;
|
||||
callback(new Error('could not fetch affected tables and last updated time: ' + msg));
|
||||
return;
|
||||
}
|
||||
|
||||
var result = rows[0];
|
||||
|
||||
var tableNames = result.tablenames.split(/^\{(.*)\}$/)[1];
|
||||
tableNames = tableNames ? tableNames.split(',') : [];
|
||||
|
||||
var lastUpdatedTime = result.max || 0;
|
||||
|
||||
callback(null, {
|
||||
affectedTables: tableNames,
|
||||
lastUpdatedTime: lastUpdatedTime * 1000
|
||||
if (shouldQueryPostgresDirectly()) {
|
||||
var psql = new PSQL(options);
|
||||
psql.query(query, function(err, resultSet) {
|
||||
var rows = resultSet.rows || [];
|
||||
handleAffectedTablesAndLastUpdatedTimeRows(err, rows, callback);
|
||||
});
|
||||
});
|
||||
} else {
|
||||
sqlApi.query(username, options.api_key, query, function(err, rows) {
|
||||
handleAffectedTablesAndLastUpdatedTimeRows(err, rows, callback);
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
function handleAffectedTablesAndLastUpdatedTimeRows(err, rows, callback) {
|
||||
if (err || rows.length === 0) {
|
||||
var msg = err.message ? err.message : err;
|
||||
callback(new Error('could not fetch affected tables and last updated time: ' + msg));
|
||||
return;
|
||||
}
|
||||
|
||||
var result = rows[0];
|
||||
|
||||
var tableNames = result.tablenames.split(/^\{(.*)\}$/)[1];
|
||||
tableNames = tableNames ? tableNames.split(',') : [];
|
||||
|
||||
var lastUpdatedTime = result.max || 0;
|
||||
|
||||
callback(null, {
|
||||
affectedTables: tableNames,
|
||||
lastUpdatedTime: lastUpdatedTime * 1000
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
function shouldQueryPostgresDirectly() {
|
||||
return global.environment
|
||||
&& global.environment.enabledFeatures
|
||||
&& global.environment.enabledFeatures.cdbQueryTablesFromPostgres;
|
||||
}
|
||||
|
||||
@@ -191,7 +191,14 @@ module.exports = function(){
|
||||
if ( req.profiler ) req.profiler.done('getSignerMapKey');
|
||||
key = data;
|
||||
}
|
||||
queryTablesApi.getAffectedTablesInQuery(user, key, sql, this); // in addCacheChannel
|
||||
queryTablesApi.getAffectedTablesInQuery(user, {
|
||||
user: req.params.dbuser,
|
||||
pass: req.params.dbpass,
|
||||
host: req.params.dbhost,
|
||||
port: req.params.dbport,
|
||||
dbname: req.params.dbname,
|
||||
api_key: key
|
||||
}, sql, this); // in addCacheChannel
|
||||
},
|
||||
function finish(err, data) {
|
||||
next(err,data);
|
||||
@@ -316,7 +323,14 @@ module.exports = function(){
|
||||
|
||||
Step(
|
||||
function getAffectedTablesAndLastUpdatedTime() {
|
||||
queryTablesApi.getAffectedTablesAndLastUpdatedTime(usr, key, sql, this);
|
||||
queryTablesApi.getAffectedTablesAndLastUpdatedTime(usr, {
|
||||
user: req.params.dbuser,
|
||||
pass: req.params.dbpass,
|
||||
host: req.params.dbhost,
|
||||
port: req.params.dbport,
|
||||
dbname: req.params.dbname,
|
||||
api_key: key
|
||||
}, sql, this);
|
||||
},
|
||||
function handleAffectedTablesAndLastUpdatedTime(err, result) {
|
||||
if (req.profiler) req.profiler.done('queryTablesAndLastUpdated');
|
||||
|
||||
Reference in New Issue
Block a user