roundup/backends/back_postgresql.py

   1 #$Id: back_postgresql.py,v 1.44 2008-08-07 05:50:03 richard Exp $
   2 #
   3 # Copyright (c) 2003 Martynas Sklyzmantas, Andrey Lebedev <andrey@micro.lt>
   4 #
   5 # This module is free software, and you may redistribute it and/or modify
   6 # under the same terms as Python, so long as this copyright message and
   7 # disclaimer are retained in their original form.
   8 #
   9 '''Postgresql backend via psycopg for Roundup.'''
  10 __docformat__ = 'restructuredtext'
  11
  12 import os, shutil, time
  13 try:
  14     import psycopg
  15     from psycopg import QuotedString
  16     from psycopg import ProgrammingError
  17 except:
  18     from psycopg2 import psycopg1 as psycopg
  19     from psycopg2.extensions import QuotedString
  20     from psycopg2.psycopg1 import ProgrammingError
  21 import logging
  22
  23 from roundup import hyperdb, date
  24 from roundup.backends import rdbms_common
  25 from roundup.backends import sessions_rdbms
  26
  27 def connection_dict(config, dbnamestr=None):
  28     ''' read_default_group is MySQL-specific, ignore it '''
  29     d = rdbms_common.connection_dict(config, dbnamestr)
  30     if 'read_default_group' in d:
  31         del d['read_default_group']
  32     if 'read_default_file' in d:
  33         del d['read_default_file']
  34     return d
  35
  36 def db_create(config):
  37     """Clear all database contents and drop database itself"""
  38     command = "CREATE DATABASE %s WITH ENCODING='UNICODE'"%config.RDBMS_NAME
  39     logging.getLogger('roundup.hyperdb').info(command)
  40     db_command(config, command)
  41
  42 def db_nuke(config, fail_ok=0):
  43     """Clear all database contents and drop database itself"""
  44     command = 'DROP DATABASE %s'% config.RDBMS_NAME
  45     logging.getLogger('roundup.hyperdb').info(command)
  46     db_command(config, command)
  47
  48     if os.path.exists(config.DATABASE):
  49         shutil.rmtree(config.DATABASE)
  50
  51 def db_command(config, command):
  52     '''Perform some sort of database-level command. Retry 10 times if we
  53     fail by conflicting with another user.
  54     '''
  55     template1 = connection_dict(config)
  56     template1['database'] = 'template1'
  57
  58     try:
  59         conn = psycopg.connect(**template1)
  60     except psycopg.OperationalError, message:
  61         raise hyperdb.DatabaseError(message)
  62
  63     conn.set_isolation_level(0)
  64     cursor = conn.cursor()
  65     try:
  66         for n in range(10):
  67             if pg_command(cursor, command):
  68                 return
  69     finally:
  70         conn.close()
  71     raise RuntimeError('10 attempts to create database failed')
  72
  73 def pg_command(cursor, command):
  74     '''Execute the postgresql command, which may be blocked by some other
  75     user connecting to the database, and return a true value if it succeeds.
  76
  77     If there is a concurrent update, retry the command.
  78     '''
  79     try:
  80         cursor.execute(command)
  81     except psycopg.ProgrammingError, err:
  82         response = str(err).split('\n')[0]
  83         if response.find('FATAL') != -1:
  84             raise RuntimeError(response)
  85         else:
  86             msgs = [
  87                 'is being accessed by other users',
  88                 'could not serialize access due to concurrent update',
  89             ]
  90             can_retry = 0
  91             for msg in msgs:
  92                 if response.find(msg) == -1:
  93                     can_retry = 1
  94             if can_retry:
  95                 time.sleep(1)
  96                 return 0
  97             raise RuntimeError(response)
  98     return 1
  99
 100 def db_exists(config):
 101     """Check if database already exists"""
 102     db = connection_dict(config, 'database')
 103     try:
 104         conn = psycopg.connect(**db)
 105         conn.close()
 106         return 1
 107     except:
 108         return 0
 109
 110 class Sessions(sessions_rdbms.Sessions):
 111     def set(self, *args, **kwargs):
 112         try:
 113             sessions_rdbms.Sessions.set(self, *args, **kwargs)
 114         except ProgrammingError, err:
 115             response = str(err).split('\n')[0]
 116             if -1 != response.find('ERROR') and \
 117                -1 != response.find('could not serialize access due to concurrent update'):
 118                 # another client just updated, and we're running on
 119                 # serializable isolation.
 120                 # see http://www.postgresql.org/docs/7.4/interactive/transaction-iso.html
 121                 self.db.rollback()
 122
 123 class Database(rdbms_common.Database):
 124     arg = '%s'
 125
 126     # used by some code to switch styles of query
 127     implements_intersect = 1
 128
 129     def getSessionManager(self):
 130         return Sessions(self)
 131
 132     def sql_open_connection(self):
 133         db = connection_dict(self.config, 'database')
 134         logging.getLogger('roundup.hyperdb').info(
 135             'open database %r'%db['database'])
 136         try:
 137             conn = psycopg.connect(**db)
 138         except psycopg.OperationalError, message:
 139             raise hyperdb.DatabaseError(message)
 140
 141         cursor = conn.cursor()
 142
 143         return (conn, cursor)
 144
 145     def open_connection(self):
 146         if not db_exists(self.config):
 147             db_create(self.config)
 148
 149         self.conn, self.cursor = self.sql_open_connection()
 150
 151         try:
 152             self.load_dbschema()
 153         except psycopg.ProgrammingError, message:
 154             if str(message).find('schema') == -1:
 155                 raise
 156             self.rollback()
 157             self.init_dbschema()
 158             self.sql("CREATE TABLE schema (schema TEXT)")
 159             self.sql("CREATE TABLE dual (dummy integer)")
 160             self.sql("insert into dual values (1)")
 161             self.create_version_2_tables()
 162
 163     def create_version_2_tables(self):
 164         # OTK store
 165         self.sql('''CREATE TABLE otks (otk_key VARCHAR(255),
 166             otk_value TEXT, otk_time REAL)''')
 167         self.sql('CREATE INDEX otks_key_idx ON otks(otk_key)')
 168
 169         # Sessions store
 170         self.sql('''CREATE TABLE sessions (
 171             session_key VARCHAR(255), session_time REAL,
 172             session_value TEXT)''')
 173         self.sql('''CREATE INDEX sessions_key_idx ON
 174             sessions(session_key)''')
 175
 176         # full-text indexing store
 177         self.sql('CREATE SEQUENCE ___textids_ids')
 178         self.sql('''CREATE TABLE __textids (
 179             _textid integer primary key, _class VARCHAR(255),
 180             _itemid VARCHAR(255), _prop VARCHAR(255))''')
 181         self.sql('''CREATE TABLE __words (_word VARCHAR(30),
 182             _textid integer)''')
 183         self.sql('CREATE INDEX words_word_idx ON __words(_word)')
 184         self.sql('CREATE INDEX words_by_id ON __words (_textid)')
 185         self.sql('CREATE UNIQUE INDEX __textids_by_props ON '
 186                  '__textids (_class, _itemid, _prop)')
 187
 188     def fix_version_2_tables(self):
 189         # Convert journal date column to TIMESTAMP, params column to TEXT
 190         self._convert_journal_tables()
 191
 192         # Convert all String properties to TEXT
 193         self._convert_string_properties()
 194
 195         # convert session / OTK *_time columns to REAL
 196         for name in ('otk', 'session'):
 197             self.sql('drop index %ss_key_idx'%name)
 198             self.sql('drop table %ss'%name)
 199             self.sql('''CREATE TABLE %ss (%s_key VARCHAR(255),
 200                 %s_value VARCHAR(255), %s_time REAL)'''%(name, name, name,
 201                 name))
 202             self.sql('CREATE INDEX %ss_key_idx ON %ss(%s_key)'%(name, name,
 203                 name))
 204
 205     def fix_version_3_tables(self):
 206         rdbms_common.Database.fix_version_3_tables(self)
 207         self.sql('''CREATE INDEX words_both_idx ON public.__words
 208             USING btree (_word, _textid)''')
 209
 210     def add_actor_column(self):
 211         # update existing tables to have the new actor column
 212         tables = self.database_schema['tables']
 213         for name in tables:
 214             self.sql('ALTER TABLE _%s add __actor VARCHAR(255)'%name)
 215
 216     def __repr__(self):
 217         return '<roundpsycopgsql 0x%x>' % id(self)
 218
 219     def sql_commit(self, fail_ok=False):
 220         ''' Actually commit to the database.
 221         '''
 222         logging.getLogger('roundup.hyperdb').info('commit')
 223
 224         try:
 225             self.conn.commit()
 226         except psycopg.ProgrammingError, message:
 227             # we've been instructed that this commit is allowed to fail
 228             if fail_ok and str(message).endswith('could not serialize '
 229                     'access due to concurrent update'):
 230                 logging.getLogger('roundup.hyperdb').info(
 231                     'commit FAILED, but fail_ok')
 232             else:
 233                 raise
 234
 235         # open a new cursor for subsequent work
 236         self.cursor = self.conn.cursor()
 237
 238     def sql_stringquote(self, value):
 239         ''' psycopg.QuotedString returns a "buffer" object with the
 240             single-quotes around it... '''
 241         return str(QuotedString(str(value)))[1:-1]
 242
 243     def sql_index_exists(self, table_name, index_name):
 244         sql = 'select count(*) from pg_indexes where ' \
 245             'tablename=%s and indexname=%s'%(self.arg, self.arg)
 246         self.sql(sql, (table_name, index_name))
 247         return self.cursor.fetchone()[0]
 248
 249     def create_class_table(self, spec, create_sequence=1):
 250         if create_sequence:
 251             sql = 'CREATE SEQUENCE _%s_ids'%spec.classname
 252             self.sql(sql)
 253
 254         return rdbms_common.Database.create_class_table(self, spec)
 255
 256     def drop_class_table(self, cn):
 257         sql = 'drop table _%s'%cn
 258         self.sql(sql)
 259
 260         sql = 'drop sequence _%s_ids'%cn
 261         self.sql(sql)
 262
 263     def newid(self, classname):
 264         sql = "select nextval('_%s_ids') from dual"%classname
 265         self.sql(sql)
 266         return str(self.cursor.fetchone()[0])
 267
 268     def setid(self, classname, setid):
 269         sql = "select setval('_%s_ids', %s) from dual"%(classname, int(setid))
 270         self.sql(sql)
 271
 272     def clear(self):
 273         rdbms_common.Database.clear(self)
 274
 275         # reset the sequences
 276         for cn in self.classes:
 277             self.cursor.execute('DROP SEQUENCE _%s_ids'%cn)
 278             self.cursor.execute('CREATE SEQUENCE _%s_ids'%cn)
 279
 280 class PostgresqlClass:
 281     order_by_null_values = '(%s is not NULL)'
 282
 283 class Class(PostgresqlClass, rdbms_common.Class):
 284     pass
 285 class IssueClass(PostgresqlClass, rdbms_common.IssueClass):
 286     pass
 287 class FileClass(PostgresqlClass, rdbms_common.FileClass):
 288     pass
 289
 290 # vim: set et sts=4 sw=4 :