pg封装为Promise

  1 var pg = require(‘pg‘);
  2 var util = require(‘util‘);
  3 //var co = require(‘co‘);
  4 var ConnectionParameters = require(‘pg/lib/connection-parameters‘);
  5 //var http = require(‘http‘);
  6 var slice = [].slice;
  7 /*
  8 var cfg = {
  9     user: ‘postgres‘,
 10     password: ‘aaaaaa‘,
 11     port: 5432,
 12     host: ‘localhost‘,
 13     database: "testdb",
 14 };*/
 15 
 16 var getConSql = (cfg) => {
 17     return "postgres://" + cfg.user + ":" + cfg.password +
 18         "@" + cfg.host + (cfg.port ? ":" + cfg.port : "") +
 19         "/" + cfg.database + "";
 20 };
 21 
 22 var cfg = app.config.connect.xhc_pg;
 23 var master = getConSql(cfg);
 24 
 25 console.dir(master);
 26 
 27 var slaves = master;
 28 if (cfg.slaves && util.isArray(cfg.slaves) && cfg.slaves.length > 0) {
 29     slaves = getConSql(cfg.slaves[0]);
 30 }
 31 
 32 if (cfg.pool && cfg.poolSize) {
 33     pg.defaults.poolSize = cfg.poolSize;
 34 }
 35 
 36 var Pool = (function() {
 37     var queryType = false;
 38     var isObject = function(obj) {
 39         var type = typeof obj;
 40         return type === ‘function‘ || type === ‘object‘ && !!obj;
 41     };
 42 
 43     var isArray = Array.isArray || function(obj) {
 44         return toString.call(obj) === ‘[object Array]‘;
 45     };
 46     //console.log(format(‘INSERT INTO info(id, date, json) VALUES (:id, :date, :json)‘,{id:4,date:"2015-05-14 12:33:33",json:{ab:"d‘d‘df",data:{a:‘df""d‘,b:"sddf"}}}));
 47     //INSERT INTO info(id, date, json) VALUES (3, ‘2015-05-14 12:33:33‘, ‘{"ab":"d‘‘d‘‘df","data":{"a":"df\"\"d","b":"hello"}}‘)
 48     var format = (sql, params) => {
 49         params = params || {};
 50         return sql.replace(/: ?(\w+)/g, function(item, name) {
 51             if (params && params.hasOwnProperty(name)) {
 52                 var value = params[name];
 53                 switch (typeof(value)) {
 54                     case "boolean":
 55                         return String(value);
 56                     case "object":
 57                         return `‘${JSON.stringify(value).replace(/\‘/g, "‘‘")}‘`;
 58                     case "number":
 59                         return value;
 60                     case "string":
 61                         value = value.replace("‘", "‘‘");
 62                         return `‘${value}‘`;
 63                 }
 64             }
 65             return item;
 66         });
 67     };
 68 
 69     var query = function(sql, params, optfun) {
 70         params = params || {};
 71         optfun = optfun || (result => {
 72             return result;
 73         });
 74         return new Promise((resolve, reject) => {
 75             var consql = /insert|update|delete/.test(sql.toLowerCase()) ? master : slaves;
 76             //console.log(consql);
 77             pg.connect(consql, function(err, client, done) {
 78                 if (err) {
 79                     reject(new Error(‘error fetching client from pool‘));
 80                     return;
 81                 }
 82                 if (queryType) {
 83                     if (/:(\w+)/.test(sql)) sql = format(sql, params);
 84                     client.query(sql, function(err, result) {
 85                         done();
 86                         if (err) {
 87                             client.end();
 88                             reject(new Error(‘error running query‘));
 89                             return;
 90                         }
 91                         resolve(optfun(result));
 92 
 93                     });
 94                 } else {
 95 
 96                     var index = 1,
 97                         myobj = {},
 98                         values = [];
 99                     for (var key in params) {
100                         myobj[key] = index;
101                         index++;
102                         values.push(params[key]);
103                     }
104                     if(values.length>0)
105                     sql = sql.replace(/: ?(\w+)/g, function(item, name) {
106                           return "$" + myobj[name];
107                     });
108 
109                     //console.dir(sql);
110                     // console.dir(values);
111                     client.query(sql, values, function(err, result) {
112                         done();
113                         if (err) {
114                             client.end();
115                             reject(new Error(‘error running query‘));
116                             return;
117                         }
118                         //console.dir(result);
119                         resolve(optfun(result));
120                     });
121                 }
122 
123 
124             });
125         });
126     };
127 
128 
129 
130     return {
131         "query": query,
132         "get": function() {
133             var args = slice.call(arguments);
134             if (args.length == 1) args.push(null);
135             args.push(result => {
136                 if (result.rowCount > 0) return result.rows[0];
137                 return [];
138             });
139             return query.apply(null, args);
140         },
141         "select": function() {
142             var args = slice.call(arguments);
143             if (args.length == 1) args.push(null);
144             args.push(result => {
145                 if (result.rowCount > 0) return result.rows;
146                 return [];
147             });
148             return query.apply(null, args);
149         },
150         "format": format,
151         "createSql": function(tname, params, type) {
152             type = type || "insert";
153             if (!tname || tname == "") return "";
154             if (!isObject(params)) return "";
155             var keys = Object.keys(params);
156             var sql = null;
157             switch (type) {
158                 case "insert":
159                     sql = util.format(`insert into ${tname}(%s) values(%s)`, keys.map(a => {
160                         return a;
161                     }).join(‘,‘), keys.map(a => {
162                         return ": " + a;
163                     }).join(‘,‘));
164                     break;
165                 case "update":
166                     sql = util.format(`update ${tname} set %s `, keys.map(function(a) {
167                         return `${a}=: ${a}`;
168                     }).join(‘,‘));
169                     break;
170             }
171             return sql || "";
172         },
173         "writeFailed": function(context, err) {
174             context.body = {
175                 "success": false,
176                 "msg": err
177             };
178         }
179     };
180 })();
181 
182 var api = new Map();
183 api.set("xhcpg", Pool);
184 module.exports = api;

//使用事例
router.post(‘/topic_banner_update‘, function*(next) {
        var param = this.param;
        var parm = {
            "tb_id": param["tb_id"],
            "tb_type": param["tb_type"],
            "tb_url": param["tb_url"],
            "tb_image": _tool.imageUrl(param["tb_image"]),
            "tb_time_start": _tool.dateStrToNum(param["tb_time_start"]),
            "tb_time_stop": _tool.dateStrToNum(param["tb_time_stop"])
        };
        var sql = Format("update topic_banner set tb_type=: tb_type,tb_url=: tb_url,tb_image=: tb_image,tb_time_start=: tb_time_start,tb_time_stop=: tb_time_stop where tb_id=: tb_id", parm);
        console.dir(sql);
        yield app.services.xhcpg.query(sql);
        this.body = {
            success: true
        };
    });

    router.post(‘/topic_banner_list‘, function*(next) {
        var param = this.param;
        var page = param["page"] || 1;
        var rows = param["rows"] || 10;
        var sql = `
        with a as(
        select * from  topic_banner  where t_id = ${param["t_id"]} and tb_status<>-1  order by tb_id desc)
        select  json_agg(a)  json from a`;
        // console.dir(sql);
        var result = yield app.services.xhcpg.select(sql);
        var rows = result[0]["json"] || [];

        rows.forEach(function(row) {
            row["tb_time_start"] = _tool.numToDateStr2(row["tb_time_start"]);
            row["tb_time_stop"] = _tool.numToDateStr2(row["tb_time_stop"]);
        });

        this.body = {
            "total": rows.length,
            "rows": rows
        };
    });