bin/sql_db.py

   1 # -*- coding: utf-8 -*-
   2 ##############################################################################
   3 #
   4 #    OpenERP, Open Source Management Solution
   5 #    Copyright (C) 2004-2009 Tiny SPRL (<http://tiny.be>).
   6 #
   7 #    This program is free software: you can redistribute it and/or modify
   8 #    it under the terms of the GNU Affero General Public License as
   9 #    published by the Free Software Foundation, either version 3 of the
  10 #    License, or (at your option) any later version.
  11 #
  12 #    This program is distributed in the hope that it will be useful,
  13 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 #    GNU Affero General Public License for more details.
  16 #
  17 #    You should have received a copy of the GNU Affero General Public License
  18 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19 #
  20 ##############################################################################
  21
  22 __all__ = ['db_connect', 'close_db']
  23
  24 import netsvc
  25 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT, ISOLATION_LEVEL_READ_COMMITTED, ISOLATION_LEVEL_SERIALIZABLE
  26 from psycopg2.psycopg1 import cursor as psycopg1cursor
  27 from psycopg2.pool import PoolError
  28
  29 import psycopg2.extensions
  30
  31 psycopg2.extensions.register_type(psycopg2.extensions.UNICODE)
  32
  33 types_mapping = {
  34     'date': (1082,),
  35     'time': (1083,),
  36     'datetime': (1114,),
  37 }
  38
  39 def unbuffer(symb, cr):
  40     if symb is None: return None
  41     return str(symb)
  42
  43 def undecimalize(symb, cr):
  44     if symb is None: return None
  45     return float(symb)
  46
  47 for name, typeoid in types_mapping.items():
  48     psycopg2.extensions.register_type(psycopg2.extensions.new_type(typeoid, name, lambda x, cr: x))
  49 psycopg2.extensions.register_type(psycopg2.extensions.new_type((700, 701, 1700,), 'float', undecimalize))
  50
  51
  52 import tools
  53 from tools.func import wraps
  54 from datetime import datetime as mdt
  55 import threading
  56
  57 import re
  58 re_from = re.compile('.* from "?([a-zA-Z_0-9]+)"? .*$');
  59 re_into = re.compile('.* into "?([a-zA-Z_0-9]+)"? .*$');
  60
  61
  62 def log(msg, lvl=netsvc.LOG_DEBUG2):
  63     logger = netsvc.Logger()
  64     logger.notifyChannel('sql', lvl, msg)
  65
  66 sql_counter = 0
  67
  68 class Cursor(object):
  69     IN_MAX = 1000
  70
  71     def check(f):
  72         @wraps(f)
  73         def wrapper(self, *args, **kwargs):
  74             if self.__closed:
  75                 raise psycopg2.ProgrammingError('Unable to use the cursor after having closed it')
  76             return f(self, *args, **kwargs)
  77         return wrapper
  78
  79     def __init__(self, pool, dbname, serialized=False):
  80         self.sql_from_log = {}
  81         self.sql_into_log = {}
  82         self.sql_log = False
  83         self.sql_log_count = 0
  84
  85         self.__closed = True    # avoid the call of close() (by __del__) if an exception
  86                                 # is raised by any of the following initialisations
  87         self._pool = pool
  88         self.dbname = dbname
  89         self._serialized = serialized
  90         self._cnx = pool.borrow(dsn(dbname))
  91         self._obj = self._cnx.cursor(cursor_factory=psycopg1cursor)
  92         self.__closed = False   # real initialisation value
  93         self.autocommit(False)
  94
  95         if tools.config['log_level'] in (netsvc.LOG_DEBUG, netsvc.LOG_DEBUG_RPC):
  96             from inspect import stack
  97             self.__caller = tuple(stack()[2][1:3])
  98
  99     def __del__(self):
 100         if not self.__closed:
 101             # Oops. 'self' has not been closed explicitly.
 102             # The cursor will be deleted by the garbage collector,
 103             # but the database connection is not put back into the connection
 104             # pool, preventing some operation on the database like dropping it.
 105             # This can also lead to a server overload.
 106             if tools.config['log_level'] in (netsvc.LOG_DEBUG, netsvc.LOG_DEBUG_RPC):
 107                 msg = "Cursor not closed explicitly\n"  \
 108                       "Cursor was created at %s:%s" % self.__caller
 109                 log(msg, netsvc.LOG_WARNING)
 110             self.close()
 111
 112     @check
 113     def execute(self, query, params=None):
 114         if '%d' in query or '%f' in query:
 115             log(query, netsvc.LOG_WARNING)
 116             log("SQL queries cannot contain %d or %f anymore. Use only %s", netsvc.LOG_WARNING)
 117             if params:
 118                 query = query.replace('%d', '%s').replace('%f', '%s')
 119
 120         if self.sql_log:
 121             now = mdt.now()
 122
 123         try:
 124             params = params or None
 125             res = self._obj.execute(query, params)
 126         except psycopg2.ProgrammingError, pe:
 127             logger= netsvc.Logger()
 128             logger.notifyChannel('sql_db', netsvc.LOG_ERROR, "Programming error: %s, in query %s" % (pe, query))
 129             raise
 130         except Exception, e:
 131             log("bad query: %s" % self._obj.query)
 132             log(e)
 133             raise
 134
 135         if self.sql_log:
 136             log("query: %s" % self._obj.query)
 137             self.sql_log_count+=1
 138             res_from = re_from.match(query.lower())
 139             if res_from:
 140                 self.sql_from_log.setdefault(res_from.group(1), [0, 0])
 141                 self.sql_from_log[res_from.group(1)][0] += 1
 142                 self.sql_from_log[res_from.group(1)][1] += mdt.now() - now
 143             res_into = re_into.match(query.lower())
 144             if res_into:
 145                 self.sql_into_log.setdefault(res_into.group(1), [0, 0])
 146                 self.sql_into_log[res_into.group(1)][0] += 1
 147                 self.sql_into_log[res_into.group(1)][1] += mdt.now() - now
 148         return res
 149
 150     def print_log(self):
 151         global sql_counter
 152         sql_counter += self.sql_log_count
 153         if not self.sql_log:
 154             return
 155         def process(type):
 156             sqllogs = {'from':self.sql_from_log, 'into':self.sql_into_log}
 157             sum = 0
 158             if sqllogs[type]:
 159                 sqllogitems = sqllogs[type].items()
 160                 sqllogitems.sort(key=lambda k: k[1][1])
 161                 log("SQL LOG %s:" % (type,))
 162                 for r in sqllogitems:
 163                     log("table: %s: %s/%s" %(r[0], str(r[1][1]), r[1][0]))
 164                     sum+= r[1][1]
 165                 sqllogs[type].clear()
 166             log("SUM %s:%s/%d [%d]" % (type, sum, self.sql_log_count,sql_counter))
 167         process('from')
 168         process('into')
 169         self.sql_log_count = 0
 170         self.sql_log = False
 171
 172     @check
 173     def close(self):
 174         if not self._obj:
 175             return
 176
 177         self.print_log()
 178
 179         if not self._serialized:
 180             self.rollback() # Ensure we close the current transaction.
 181
 182         self._obj.close()
 183
 184         # This force the cursor to be freed, and thus, available again. It is
 185         # important because otherwise we can overload the server very easily
 186         # because of a cursor shortage (because cursors are not garbage
 187         # collected as fast as they should). The problem is probably due in
 188         # part because browse records keep a reference to the cursor.
 189         del self._obj
 190         self.__closed = True
 191         self._pool.give_back(self._cnx)
 192
 193     @check
 194     def autocommit(self, on):
 195         offlevel = [ISOLATION_LEVEL_READ_COMMITTED, ISOLATION_LEVEL_SERIALIZABLE][bool(self._serialized)]
 196         self._cnx.set_isolation_level([offlevel, ISOLATION_LEVEL_AUTOCOMMIT][bool(on)])
 197
 198     @check
 199     def commit(self):
 200         return self._cnx.commit()
 201
 202     @check
 203     def rollback(self):
 204         return self._cnx.rollback()
 205
 206     @check
 207     def __getattr__(self, name):
 208         return getattr(self._obj, name)
 209
 210
 211 class ConnectionPool(object):
 212
 213     def locked(fun):
 214         @wraps(fun)
 215         def _locked(self, *args, **kwargs):
 216             self._lock.acquire()
 217             try:
 218                 return fun(self, *args, **kwargs)
 219             finally:
 220                 self._lock.release()
 221         return _locked
 222
 223
 224     def __init__(self, maxconn=64):
 225         self._connections = []
 226         self._maxconn = max(maxconn, 1)
 227         self._lock = threading.Lock()
 228         self._logger = netsvc.Logger()
 229
 230     def _log(self, msg):
 231         #self._logger.notifyChannel('ConnectionPool', netsvc.LOG_INFO, msg)
 232         pass
 233     def _debug(self, msg):
 234         #self._logger.notifyChannel('ConnectionPool', netsvc.LOG_DEBUG, msg)
 235         pass
 236
 237     @locked
 238     def borrow(self, dsn):
 239         self._log('Borrow connection to %s' % (dsn,))
 240
 241         result = None
 242         for i, (cnx, used) in enumerate(self._connections):
 243             if not used and cnx.dsn == dsn:
 244                 self._debug('Existing connection found at index %d' % i)
 245
 246                 self._connections.pop(i)
 247                 self._connections.append((cnx, True))
 248
 249                 result = cnx
 250                 break
 251         if result:
 252             return result
 253
 254         if len(self._connections) >= self._maxconn:
 255             # try to remove the older connection not used
 256             for i, (cnx, used) in enumerate(self._connections):
 257                 if not used:
 258                     self._debug('Removing old connection at index %d: %s' % (i, cnx.dsn))
 259                     self._connections.pop(i)
 260                     break
 261             else:
 262                 # note: this code is called only if the for loop has completed (no break)
 263                 raise PoolError('Connection Pool Full')
 264
 265         self._debug('Create new connection')
 266         result = psycopg2.connect(dsn=dsn)
 267         self._connections.append((result, True))
 268         return result
 269
 270     @locked
 271     def give_back(self, connection):
 272         self._log('Give back connection to %s' % (connection.dsn,))
 273         for i, (cnx, used) in enumerate(self._connections):
 274             if cnx is connection:
 275                 self._connections.pop(i)
 276                 self._connections.append((cnx, False))
 277                 break
 278         else:
 279             raise PoolError('This connection does not below to the pool')
 280
 281     @locked
 282     def close_all(self, dsn):
 283         for i, (cnx, used) in tools.reverse_enumerate(self._connections):
 284             if cnx.dsn == dsn:
 285                 cnx.close()
 286                 self._connections.pop(i)
 287
 288
 289 class Connection(object):
 290     __LOCKS = {}
 291
 292     def __init__(self, pool, dbname, unique=False):
 293         self.dbname = dbname
 294         self._pool = pool
 295         self._unique = unique
 296         if unique:
 297             if dbname not in self.__LOCKS:
 298                 self.__LOCKS[dbname] = threading.Lock()
 299             self.__LOCKS[dbname].acquire()
 300
 301     def __del__(self):
 302         if self._unique:
 303             self.__LOCKS[self.dbname].release()
 304
 305     def cursor(self, serialized=False):
 306         return Cursor(self._pool, self.dbname, serialized=serialized)
 307
 308     def serialized_cursor(self):
 309         return self.cursor(True)
 310
 311
 312 _dsn = ''
 313 for p in ('host', 'port', 'user', 'password'):
 314     cfg = tools.config['db_' + p]
 315     if cfg:
 316         _dsn += '%s=%s ' % (p, cfg)
 317
 318 def dsn(db_name):
 319     return '%sdbname=%s' % (_dsn, db_name)
 320
 321
 322 _Pool = ConnectionPool(int(tools.config['db_maxconn']))
 323
 324 def db_connect(db_name):
 325     unique = db_name in ['template1', 'template0']
 326     return Connection(_Pool, db_name, unique)
 327
 328 def close_db(db_name):
 329     _Pool.close_all(dsn(db_name))
 330     tools.cache.clean_caches_for_db(db_name)
 331
 332
 333 # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
 334