--- /dev/null
+# ZPsycopgDA/db.py - query execution
+#
+# Copyright (C) 2004 Federico Di Gregorio <fog@initd.org>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by the
+# Free Software Foundation; either version 2, or (at your option) any later
+# version.
+#
+# Or, at your option this program (ZPsycopgDA) can be distributed under the
+# Zope Public License (ZPL) Version 1.0, as published on the Zope web site,
+# http://www.zope.org/Resources/ZPL.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTIBILITY
+# or FITNESS FOR A PARTICULAR PURPOSE.
+#
+# See the LICENSE file for details.
+
+from Shared.DC.ZRDB.TM import TM
+from Shared.DC.ZRDB import dbi_db
+
+from ZODB.POSException import ConflictError
+
+import site
+import pool
+
+import psycopg2
+from psycopg2.extensions import INTEGER, LONGINTEGER, FLOAT, BOOLEAN, DATE, TIME
+from psycopg2 import NUMBER, STRING, ROWID, DATETIME
+
+
+# the DB object, managing all the real query work
+
+class DB(TM, dbi_db.DB):
+
+ _p_oid = _p_changed = _registered = None
+
+ def __init__(self, dsn, tilevel, enc='utf-8'):
+ self.dsn = dsn
+ self.tilevel = tilevel
+ self.encoding = enc
+ self.failures = 0
+ self.calls = 0
+ self.make_mappings()
+
+ def getconn(self, create=True):
+ conn = pool.getconn(self.dsn)
+ conn.set_isolation_level(int(self.tilevel))
+ return conn
+
+ def putconn(self, close=False):
+ try:
+ conn = pool.getconn(self.dsn, False)
+ except AttributeError:
+ pass
+ pool.putconn(self.dsn, conn, close)
+
+ def getcursor(self):
+ conn = self.getconn()
+ return conn.cursor()
+
+ def _finish(self, *ignored):
+ try:
+ conn = self.getconn(False)
+ conn.commit()
+ self.putconn()
+ except AttributeError:
+ pass
+
+ def _abort(self, *ignored):
+ try:
+ conn = self.getconn(False)
+ conn.rollback()
+ self.putconn()
+ except AttributeError:
+ pass
+
+ def open(self):
+ # this will create a new pool for our DSN if not already existing,
+ # then get and immediately release a connection
+ self.getconn()
+ self.putconn()
+
+ def close(self):
+ # FIXME: if this connection is closed we flush all the pool associated
+ # with the current DSN; does this makes sense?
+ pool.flushpool(self.dsn)
+
+ def sortKey(self):
+ return 1
+
+ def make_mappings(self):
+ """Generate the mappings used later by self.convert_description()."""
+ self.type_mappings = {}
+ for t, s in [(INTEGER,'i'), (LONGINTEGER, 'i'), (NUMBER, 'n'),
+ (BOOLEAN,'n'), (ROWID, 'i'),
+ (DATETIME, 'd'), (DATE, 'd'), (TIME, 'd')]:
+ for v in t.values:
+ self.type_mappings[v] = (t, s)
+
+ def convert_description(self, desc, use_psycopg_types=False):
+ """Convert DBAPI-2.0 description field to Zope format."""
+ items = []
+ for name, typ, width, ds, p, scale, null_ok in desc:
+ m = self.type_mappings.get(typ, (STRING, 's'))
+ items.append({
+ 'name': name,
+ 'type': use_psycopg_types and m[0] or m[1],
+ 'width': width,
+ 'precision': p,
+ 'scale': scale,
+ 'null': null_ok,
+ })
+ return items
+
+ ## tables and rows ##
+
+ def tables(self, rdb=0, _care=('TABLE', 'VIEW')):
+ self._register()
+ c = self.getcursor()
+ c.execute(
+ "SELECT t.tablename AS NAME, 'TABLE' AS TYPE "
+ " FROM pg_tables t WHERE tableowner <> 'postgres' "
+ "UNION SELECT v.viewname AS NAME, 'VIEW' AS TYPE "
+ " FROM pg_views v WHERE viewowner <> 'postgres' "
+ "UNION SELECT t.tablename AS NAME, 'SYSTEM_TABLE\' AS TYPE "
+ " FROM pg_tables t WHERE tableowner = 'postgres' "
+ "UNION SELECT v.viewname AS NAME, 'SYSTEM_TABLE' AS TYPE "
+ "FROM pg_views v WHERE viewowner = 'postgres'")
+ res = []
+ for name, typ in c.fetchall():
+ if typ in _care:
+ res.append({'TABLE_NAME': name, 'TABLE_TYPE': typ})
+ self.putconn()
+ return res
+
+ def columns(self, table_name):
+ self._register()
+ c = self.getcursor()
+ try:
+ r = c.execute('SELECT * FROM "%s" WHERE 1=0' % table_name)
+ except:
+ return ()
+ self.putconn()
+ return self.convert_description(c.description, True)
+
+ ## query execution ##
+
+ def query(self, query_string, max_rows=None, query_data=None):
+ self._register()
+ self.calls = self.calls+1
+
+ desc = ()
+ res = []
+ nselects = 0
+
+ c = self.getcursor()
+
+ try:
+ for qs in [x for x in query_string.split('\0') if x]:
+ if type(qs) == unicode:
+ if self.encoding:
+ qs = qs.encode(self.encoding)
+ try:
+ if query_data:
+ c.execute(qs, query_data)
+ else:
+ c.execute(qs)
+ except psycopg2.OperationalError, e:
+ try:
+ self.close()
+ except:
+ pass
+ self.open()
+ try:
+ if query_data:
+ c.execute(qs, query_data)
+ else:
+ c.execute(qs)
+ except (psycopg2.ProgrammingError,
+ psycopg2.IntegrityError), e:
+ if e.args[0].find("concurrent update") > -1:
+ raise ConflictError
+ raise e
+ except (psycopg2.ProgrammingError, psycopg2.IntegrityError), e:
+ if e.args[0].find("concurrent update") > -1:
+ raise ConflictError
+ raise e
+ if c.description is not None:
+ nselects += 1
+ if c.description != desc and nselects > 1:
+ raise psycopg2.ProgrammingError(
+ 'multiple selects in single query not allowed')
+ if max_rows:
+ res = c.fetchmany(max_rows)
+ else:
+ res = c.fetchall()
+ desc = c.description
+ self.failures = 0
+
+ except StandardError, err:
+ self._abort()
+ raise err
+
+ return self.convert_description(desc), res
git://git.onelab.eu / plcapi.git / blobdiff