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 from datetime import timedelta
  56 import threading
  57
  58 import re
  59 re_from = re.compile('.* from "?([a-zA-Z_0-9]+)"? .*$');
  60 re_into = re.compile('.* into "?([a-zA-Z_0-9]+)"? .*$');
  61
  62
  63 def log(msg, lvl=netsvc.LOG_DEBUG2):
  64     logger = netsvc.Logger()
  65     logger.notifyChannel('sql', lvl, msg)
  66
  67 sql_counter = 0
  68
  69 class Cursor(object):
  70     IN_MAX = 1000
  71
  72     def check(f):
  73         @wraps(f)
  74         def wrapper(self, *args, **kwargs):
  75             if self.__closed:
  76                 raise psycopg2.ProgrammingError('Unable to use the cursor after having closed it')
  77             return f(self, *args, **kwargs)
  78         return wrapper
  79
  80     def __init__(self, pool, dbname, serialized=False):
  81         self.sql_from_log = {}
  82         self.sql_into_log = {}
  83         self.sql_log = False
  84         self.sql_log_count = 0
  85
  86         self.__closed = True    # avoid the call of close() (by __del__) if an exception
  87                                 # is raised by any of the following initialisations
  88         self._pool = pool
  89         self.dbname = dbname
  90         self._serialized = serialized
  91         self._cnx = pool.borrow(dsn(dbname))
  92         self._obj = self._cnx.cursor(cursor_factory=psycopg1cursor)
  93         self.__closed = False   # real initialisation value
  94         self.autocommit(False)
  95
  96         if tools.config['log_level'] in (netsvc.LOG_DEBUG, netsvc.LOG_DEBUG_RPC):
  97             from inspect import stack
  98             self.__caller = tuple(stack()[2][1:3])
  99
 100     def __del__(self):
 101         if not self.__closed:
 102             # Oops. 'self' has not been closed explicitly.
 103             # The cursor will be deleted by the garbage collector,
 104             # but the database connection is not put back into the connection
 105             # pool, preventing some operation on the database like dropping it.
 106             # This can also lead to a server overload.
 107             if tools.config['log_level'] in (netsvc.LOG_DEBUG, netsvc.LOG_DEBUG_RPC):
 108                 msg = "Cursor not closed explicitly\n"  \
 109                       "Cursor was created at %s:%s" % self.__caller
 110                 log(msg, netsvc.LOG_WARNING)
 111             self.close()
 112
 113     @check
 114     def execute(self, query, params=None):
 115         if '%d' in query or '%f' in query:
 116             log(query, netsvc.LOG_WARNING)
 117             log("SQL queries cannot contain %d or %f anymore. Use only %s", netsvc.LOG_WARNING)
 118             if params:
 119                 query = query.replace('%d', '%s').replace('%f', '%s')
 120
 121         if self.sql_log:
 122             now = mdt.now()
 123
 124         try:
 125             params = params or None
 126             res = self._obj.execute(query, params)
 127         except psycopg2.ProgrammingError, pe:
 128             logger= netsvc.Logger()
 129             logger.notifyChannel('sql_db', netsvc.LOG_ERROR, "Programming error: %s, in query %s" % (pe, query))
 130             raise
 131         except Exception, e:
 132             log("bad query: %s" % self._obj.query)
 133             log(e)
 134             raise
 135
 136         if self.sql_log:
 137             delay = mdt.now() - now
 138             delay = delay.seconds * 1E6 + delay.microseconds
 139
 140             log("query: %s" % self._obj.query)
 141             self.sql_log_count+=1
 142             res_from = re_from.match(query.lower())
 143             if res_from:
 144                 self.sql_from_log.setdefault(res_from.group(1), [0, 0])
 145                 self.sql_from_log[res_from.group(1)][0] += 1
 146                 self.sql_from_log[res_from.group(1)][1] += delay
 147             res_into = re_into.match(query.lower())
 148             if res_into:
 149                 self.sql_into_log.setdefault(res_into.group(1), [0, 0])
 150                 self.sql_into_log[res_into.group(1)][0] += 1
 151                 self.sql_into_log[res_into.group(1)][1] += delay
 152         return res
 153
 154     def print_log(self):
 155         global sql_counter
 156         sql_counter += self.sql_log_count
 157         if not self.sql_log:
 158             return
 159         def process(type):
 160             sqllogs = {'from':self.sql_from_log, 'into':self.sql_into_log}
 161             sum = 0
 162             if sqllogs[type]:
 163                 sqllogitems = sqllogs[type].items()
 164                 sqllogitems.sort(key=lambda k: k[1][1])
 165                 log("SQL LOG %s:" % (type,))
 166                 for r in sqllogitems:
 167                     delay = timedelta(microseconds=r[1][1])
 168                     log("table: %s: %s/%s" %(r[0], str(delay), r[1][0]))
 169                     sum+= r[1][1]
 170                 sqllogs[type].clear()
 171             sum = timedelta(microseconds=sum)
 172             log("SUM %s:%s/%d [%d]" % (type, str(sum), self.sql_log_count, sql_counter))
 173             sqllogs[type].clear()
 174         process('from')
 175         process('into')
 176         self.sql_log_count = 0
 177         self.sql_log = False
 178
 179     @check
 180     def close(self):
 181         if not self._obj:
 182             return
 183
 184         self.print_log()
 185
 186         if not self._serialized:
 187             self.rollback() # Ensure we close the current transaction.
 188
 189         self._obj.close()
 190
 191         # This force the cursor to be freed, and thus, available again. It is
 192         # important because otherwise we can overload the server very easily
 193         # because of a cursor shortage (because cursors are not garbage
 194         # collected as fast as they should). The problem is probably due in
 195         # part because browse records keep a reference to the cursor.
 196         del self._obj
 197         self.__closed = True
 198         self._pool.give_back(self._cnx)
 199
 200     @check
 201     def autocommit(self, on):
 202         offlevel = [ISOLATION_LEVEL_READ_COMMITTED, ISOLATION_LEVEL_SERIALIZABLE][bool(self._serialized)]
 203         self._cnx.set_isolation_level([offlevel, ISOLATION_LEVEL_AUTOCOMMIT][bool(on)])
 204
 205     @check
 206     def commit(self):
 207         return self._cnx.commit()
 208
 209     @check
 210     def rollback(self):
 211         return self._cnx.rollback()
 212
 213     @check
 214     def __getattr__(self, name):
 215         return getattr(self._obj, name)
 216
 217
 218 class ConnectionPool(object):
 219
 220     def locked(fun):
 221         @wraps(fun)
 222         def _locked(self, *args, **kwargs):
 223             self._lock.acquire()
 224             try:
 225                 return fun(self, *args, **kwargs)
 226             finally:
 227                 self._lock.release()
 228         return _locked
 229
 230
 231     def __init__(self, maxconn=64):
 232         self._connections = []
 233         self._maxconn = max(maxconn, 1)
 234         self._lock = threading.Lock()
 235         self._logger = netsvc.Logger()
 236
 237     def _log(self, msg):
 238         #self._logger.notifyChannel('ConnectionPool', netsvc.LOG_INFO, msg)
 239         pass
 240     def _debug(self, msg):
 241         #self._logger.notifyChannel('ConnectionPool', netsvc.LOG_DEBUG, msg)
 242         pass
 243
 244     @locked
 245     def borrow(self, dsn):
 246         self._log('Borrow connection to %s' % (dsn,))
 247
 248         result = None
 249         for i, (cnx, used) in enumerate(self._connections):
 250             if not used and dsn_are_equals(cnx.dsn, dsn):
 251                 self._debug('Existing connection found at index %d' % i)
 252
 253                 self._connections.pop(i)
 254                 self._connections.append((cnx, True))
 255
 256                 result = cnx
 257                 break
 258         if result:
 259             return result
 260
 261         if len(self._connections) >= self._maxconn:
 262             # try to remove the older connection not used
 263             for i, (cnx, used) in enumerate(self._connections):
 264                 if not used:
 265                     self._debug('Removing old connection at index %d: %s' % (i, cnx.dsn))
 266                     self._connections.pop(i)
 267                     break
 268             else:
 269                 # note: this code is called only if the for loop has completed (no break)
 270                 raise PoolError('Connection Pool Full')
 271
 272         self._debug('Create new connection')
 273         result = psycopg2.connect(dsn=dsn)
 274         self._connections.append((result, True))
 275         return result
 276
 277     @locked
 278     def give_back(self, connection):
 279         self._log('Give back connection to %s' % (connection.dsn,))
 280         for i, (cnx, used) in enumerate(self._connections):
 281             if cnx is connection:
 282                 self._connections.pop(i)
 283                 self._connections.append((cnx, False))
 284                 break
 285         else:
 286             raise PoolError('This connection does not below to the pool')
 287
 288     @locked
 289     def close_all(self, dsn):
 290         for i, (cnx, used) in tools.reverse_enumerate(self._connections):
 291             if dsn_are_equals(cnx.dsn, dsn):
 292                 cnx.close()
 293                 self._connections.pop(i)
 294
 295
 296 class Connection(object):
 297     __LOCKS = {}
 298
 299     def __init__(self, pool, dbname, unique=False):
 300         self.dbname = dbname
 301         self._pool = pool
 302         self._unique = unique
 303         if unique:
 304             if dbname not in self.__LOCKS:
 305                 self.__LOCKS[dbname] = threading.Lock()
 306             self.__LOCKS[dbname].acquire()
 307
 308     def __del__(self):
 309         if self._unique:
 310             close_db(self.dbname)
 311             self.__LOCKS[self.dbname].release()
 312
 313     def cursor(self, serialized=False):
 314         return Cursor(self._pool, self.dbname, serialized=serialized)
 315
 316     def serialized_cursor(self):
 317         return self.cursor(True)
 318
 319     def __nonzero__(self):
 320         """Check if connection is possible"""
 321         try:
 322             cr = self.cursor()
 323             cr.close()
 324             return True
 325         except:
 326             return False
 327
 328
 329 _dsn = ''
 330 for p in ('host', 'port', 'user', 'password'):
 331     cfg = tools.config['db_' + p]
 332     if cfg:
 333         _dsn += '%s=%s ' % (p, cfg)
 334
 335 def dsn(db_name):
 336     return '%sdbname=%s' % (_dsn, db_name)
 337
 338 def dsn_are_equals(first, second):
 339     def key(dsn):
 340         k = dict(x.split('=', 1) for x in dsn.strip().split())
 341         k.pop('password', None) # password is not relevant
 342         return k
 343     return key(first) == key(second)
 344
 345
 346 _Pool = ConnectionPool(int(tools.config['db_maxconn']))
 347
 348 def db_connect(db_name):
 349     unique = db_name in ['template1', 'template0']
 350     return Connection(_Pool, db_name, unique)
 351
 352 def close_db(db_name):
 353     _Pool.close_all(dsn(db_name))
 354     tools.cache.clean_caches_for_db(db_name)
 355
 356
 357 # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4:
 358