[auf_roundup.git] / build / lib / roundup / backends / back_postgresql.py

#$Id: back_postgresql.py,v 1.44 2008-08-07 05:50:03 richard Exp $
#
# Copyright (c) 2003 Martynas Sklyzmantas, Andrey Lebedev <andrey@micro.lt>
#
# This module is free software, and you may redistribute it and/or modify
# under the same terms as Python, so long as this copyright message and
# disclaimer are retained in their original form.
#
'''Postgresql backend via psycopg for Roundup.'''
__docformat__ = 'restructuredtext'

import os, shutil, time
try:
    import psycopg
    from psycopg import QuotedString
    from psycopg import ProgrammingError
except:
    from psycopg2 import psycopg1 as psycopg
    from psycopg2.extensions import QuotedString
    from psycopg2.psycopg1 import ProgrammingError
import logging

from roundup import hyperdb, date
from roundup.backends import rdbms_common
from roundup.backends import sessions_rdbms

def connection_dict(config, dbnamestr=None):
    ''' read_default_group is MySQL-specific, ignore it '''
    d = rdbms_common.connection_dict(config, dbnamestr)
    if 'read_default_group' in d:
        del d['read_default_group']
    if 'read_default_file' in d:
        del d['read_default_file']
    return d

def db_create(config):
    """Clear all database contents and drop database itself"""
    command = "CREATE DATABASE %s WITH ENCODING='UNICODE'"%config.RDBMS_NAME
    logging.getLogger('hyperdb').info(command)
    db_command(config, command)

def db_nuke(config, fail_ok=0):
    """Clear all database contents and drop database itself"""
    command = 'DROP DATABASE %s'% config.RDBMS_NAME
    logging.getLogger('hyperdb').info(command)
    db_command(config, command)

    if os.path.exists(config.DATABASE):
        shutil.rmtree(config.DATABASE)

def db_command(config, command):
    '''Perform some sort of database-level command. Retry 10 times if we
    fail by conflicting with another user.
    '''
    template1 = connection_dict(config)
    template1['database'] = 'template1'

    try:
        conn = psycopg.connect(**template1)
    except psycopg.OperationalError, message:
        raise hyperdb.DatabaseError(message)

    conn.set_isolation_level(0)
    cursor = conn.cursor()
    try:
        for n in range(10):
            if pg_command(cursor, command):
                return
    finally:
        conn.close()
    raise RuntimeError('10 attempts to create database failed')

def pg_command(cursor, command):
    '''Execute the postgresql command, which may be blocked by some other
    user connecting to the database, and return a true value if it succeeds.

    If there is a concurrent update, retry the command.
    '''
    try:
        cursor.execute(command)
    except psycopg.ProgrammingError, err:
        response = str(err).split('\n')[0]
        if response.find('FATAL') != -1:
            raise RuntimeError(response)
        else:
            msgs = [
                'is being accessed by other users',
                'could not serialize access due to concurrent update',
            ]
            can_retry = 0
            for msg in msgs:
                if response.find(msg) == -1:
                    can_retry = 1
            if can_retry:
                time.sleep(1)
                return 0
            raise RuntimeError(response)
    return 1

def db_exists(config):
    """Check if database already exists"""
    db = connection_dict(config, 'database')
    try:
        conn = psycopg.connect(**db)
        conn.close()
        return 1
    except:
        return 0

class Sessions(sessions_rdbms.Sessions):
    def set(self, *args, **kwargs):
        try:
            sessions_rdbms.Sessions.set(self, *args, **kwargs)
        except ProgrammingError, err:
            response = str(err).split('\n')[0]
            if -1 != response.find('ERROR') and \
               -1 != response.find('could not serialize access due to concurrent update'):
                # another client just updated, and we're running on
                # serializable isolation.
                # see http://www.postgresql.org/docs/7.4/interactive/transaction-iso.html
                self.db.rollback()

class Database(rdbms_common.Database):
    arg = '%s'

    # used by some code to switch styles of query
    implements_intersect = 1

    def getSessionManager(self):
        return Sessions(self)

    def sql_open_connection(self):
        db = connection_dict(self.config, 'database')
        logging.getLogger('hyperdb').info('open database %r'%db['database'])
        try:
            conn = psycopg.connect(**db)
        except psycopg.OperationalError, message:
            raise hyperdb.DatabaseError(message)

        cursor = conn.cursor()

        return (conn, cursor)

    def open_connection(self):
        if not db_exists(self.config):
            db_create(self.config)

        self.conn, self.cursor = self.sql_open_connection()

        try:
            self.load_dbschema()
        except psycopg.ProgrammingError, message:
            if str(message).find('schema') == -1:
                raise
            self.rollback()
            self.init_dbschema()
            self.sql("CREATE TABLE schema (schema TEXT)")
            self.sql("CREATE TABLE dual (dummy integer)")
            self.sql("insert into dual values (1)")
            self.create_version_2_tables()

    def create_version_2_tables(self):
        # OTK store
        self.sql('''CREATE TABLE otks (otk_key VARCHAR(255),
            otk_value TEXT, otk_time REAL)''')
        self.sql('CREATE INDEX otks_key_idx ON otks(otk_key)')

        # Sessions store
        self.sql('''CREATE TABLE sessions (
            session_key VARCHAR(255), session_time REAL,
            session_value TEXT)''')
        self.sql('''CREATE INDEX sessions_key_idx ON
            sessions(session_key)''')

        # full-text indexing store
        self.sql('CREATE SEQUENCE ___textids_ids')
        self.sql('''CREATE TABLE __textids (
            _textid integer primary key, _class VARCHAR(255),
            _itemid VARCHAR(255), _prop VARCHAR(255))''')
        self.sql('''CREATE TABLE __words (_word VARCHAR(30),
            _textid integer)''')
        self.sql('CREATE INDEX words_word_idx ON __words(_word)')
        self.sql('CREATE INDEX words_by_id ON __words (_textid)')
        self.sql('CREATE UNIQUE INDEX __textids_by_props ON '
                 '__textids (_class, _itemid, _prop)')

    def fix_version_2_tables(self):
        # Convert journal date column to TIMESTAMP, params column to TEXT
        self._convert_journal_tables()

        # Convert all String properties to TEXT
        self._convert_string_properties()

        # convert session / OTK *_time columns to REAL
        for name in ('otk', 'session'):
            self.sql('drop index %ss_key_idx'%name)
            self.sql('drop table %ss'%name)
            self.sql('''CREATE TABLE %ss (%s_key VARCHAR(255),
                %s_value VARCHAR(255), %s_time REAL)'''%(name, name, name,
                name))
            self.sql('CREATE INDEX %ss_key_idx ON %ss(%s_key)'%(name, name,
                name))

    def fix_version_3_tables(self):
        rdbms_common.Database.fix_version_3_tables(self)
        self.sql('''CREATE INDEX words_both_idx ON public.__words
            USING btree (_word, _textid)''')

    def add_actor_column(self):
        # update existing tables to have the new actor column
        tables = self.database_schema['tables']
        for name in tables:
            self.sql('ALTER TABLE _%s add __actor VARCHAR(255)'%name)

    def __repr__(self):
        return '<roundpsycopgsql 0x%x>' % id(self)

    def sql_commit(self, fail_ok=False):
        ''' Actually commit to the database.
        '''
        logging.getLogger('hyperdb').info('commit')

        try:
            self.conn.commit()
        except psycopg.ProgrammingError, message:
            # we've been instructed that this commit is allowed to fail
            if fail_ok and str(message).endswith('could not serialize '
                    'access due to concurrent update'):
                logging.getLogger('hyperdb').info('commit FAILED, but fail_ok')
            else:
                raise

        # open a new cursor for subsequent work
        self.cursor = self.conn.cursor()

    def sql_stringquote(self, value):
        ''' psycopg.QuotedString returns a "buffer" object with the
            single-quotes around it... '''
        return str(QuotedString(str(value)))[1:-1]

    def sql_index_exists(self, table_name, index_name):
        sql = 'select count(*) from pg_indexes where ' \
            'tablename=%s and indexname=%s'%(self.arg, self.arg)
        self.sql(sql, (table_name, index_name))
        return self.cursor.fetchone()[0]

    def create_class_table(self, spec, create_sequence=1):
        if create_sequence:
            sql = 'CREATE SEQUENCE _%s_ids'%spec.classname
            self.sql(sql)

        return rdbms_common.Database.create_class_table(self, spec)

    def drop_class_table(self, cn):
        sql = 'drop table _%s'%cn
        self.sql(sql)

        sql = 'drop sequence _%s_ids'%cn
        self.sql(sql)

    def newid(self, classname):
        sql = "select nextval('_%s_ids') from dual"%classname
        self.sql(sql)
        return str(self.cursor.fetchone()[0])

    def setid(self, classname, setid):
        sql = "select setval('_%s_ids', %s) from dual"%(classname, int(setid))
        self.sql(sql)

    def clear(self):
        rdbms_common.Database.clear(self)

        # reset the sequences
        for cn in self.classes:
            self.cursor.execute('DROP SEQUENCE _%s_ids'%cn)
            self.cursor.execute('CREATE SEQUENCE _%s_ids'%cn)

class PostgresqlClass:
    order_by_null_values = '(%s is not NULL)'

class Class(PostgresqlClass, rdbms_common.Class):
    pass
class IssueClass(PostgresqlClass, rdbms_common.IssueClass):
    pass
class FileClass(PostgresqlClass, rdbms_common.FileClass):
    pass

# vim: set et sts=4 sw=4 :
Commit	Line	Data
c638d827 CR	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('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('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('hyperdb').info('open database %r'%db['database'])
135	try:
136	conn = psycopg.connect(**db)
137	except psycopg.OperationalError, message:
138	raise hyperdb.DatabaseError(message)
139
140	cursor = conn.cursor()
141
142	return (conn, cursor)
143
144	def open_connection(self):
145	if not db_exists(self.config):
146	db_create(self.config)
147
148	self.conn, self.cursor = self.sql_open_connection()
149
150	try:
151	self.load_dbschema()
152	except psycopg.ProgrammingError, message:
153	if str(message).find('schema') == -1:
154	raise
155	self.rollback()
156	self.init_dbschema()
157	self.sql("CREATE TABLE schema (schema TEXT)")
158	self.sql("CREATE TABLE dual (dummy integer)")
159	self.sql("insert into dual values (1)")
160	self.create_version_2_tables()
161
162	def create_version_2_tables(self):
163	# OTK store
164	self.sql('''CREATE TABLE otks (otk_key VARCHAR(255),
165	otk_value TEXT, otk_time REAL)''')
166	self.sql('CREATE INDEX otks_key_idx ON otks(otk_key)')
167
168	# Sessions store
169	self.sql('''CREATE TABLE sessions (
170	session_key VARCHAR(255), session_time REAL,
171	session_value TEXT)''')
172	self.sql('''CREATE INDEX sessions_key_idx ON
173	sessions(session_key)''')
174
175	# full-text indexing store
176	self.sql('CREATE SEQUENCE ___textids_ids')
177	self.sql('''CREATE TABLE __textids (
178	_textid integer primary key, _class VARCHAR(255),
179	_itemid VARCHAR(255), _prop VARCHAR(255))''')
180	self.sql('''CREATE TABLE __words (_word VARCHAR(30),
181	_textid integer)''')
182	self.sql('CREATE INDEX words_word_idx ON __words(_word)')
183	self.sql('CREATE INDEX words_by_id ON __words (_textid)')
184	self.sql('CREATE UNIQUE INDEX __textids_by_props ON '
185	'__textids (_class, _itemid, _prop)')
186
187	def fix_version_2_tables(self):
188	# Convert journal date column to TIMESTAMP, params column to TEXT
189	self._convert_journal_tables()
190
191	# Convert all String properties to TEXT
192	self._convert_string_properties()
193
194	# convert session / OTK *_time columns to REAL
195	for name in ('otk', 'session'):
196	self.sql('drop index %ss_key_idx'%name)
197	self.sql('drop table %ss'%name)
198	self.sql('''CREATE TABLE %ss (%s_key VARCHAR(255),
199	%s_value VARCHAR(255), %s_time REAL)'''%(name, name, name,
200	name))
201	self.sql('CREATE INDEX %ss_key_idx ON %ss(%s_key)'%(name, name,
202	name))
203
204	def fix_version_3_tables(self):
205	rdbms_common.Database.fix_version_3_tables(self)
206	self.sql('''CREATE INDEX words_both_idx ON public.__words
207	USING btree (_word, _textid)''')
208
209	def add_actor_column(self):
210	# update existing tables to have the new actor column
211	tables = self.database_schema['tables']
212	for name in tables:
213	self.sql('ALTER TABLE _%s add __actor VARCHAR(255)'%name)
214
215	def __repr__(self):
216	return '<roundpsycopgsql 0x%x>' % id(self)
217
218	def sql_commit(self, fail_ok=False):
219	''' Actually commit to the database.
220	'''
221	logging.getLogger('hyperdb').info('commit')
222
223	try:
224	self.conn.commit()
225	except psycopg.ProgrammingError, message:
226	# we've been instructed that this commit is allowed to fail
227	if fail_ok and str(message).endswith('could not serialize '
228	'access due to concurrent update'):
229	logging.getLogger('hyperdb').info('commit FAILED, but fail_ok')
230	else:
231	raise
232
233	# open a new cursor for subsequent work
234	self.cursor = self.conn.cursor()
235
236	def sql_stringquote(self, value):
237	''' psycopg.QuotedString returns a "buffer" object with the
238	single-quotes around it... '''
239	return str(QuotedString(str(value)))[1:-1]
240
241	def sql_index_exists(self, table_name, index_name):
242	sql = 'select count(*) from pg_indexes where ' \
243	'tablename=%s and indexname=%s'%(self.arg, self.arg)
244	self.sql(sql, (table_name, index_name))
245	return self.cursor.fetchone()[0]
246
247	def create_class_table(self, spec, create_sequence=1):
248	if create_sequence:
249	sql = 'CREATE SEQUENCE _%s_ids'%spec.classname
250	self.sql(sql)
251
252	return rdbms_common.Database.create_class_table(self, spec)
253
254	def drop_class_table(self, cn):
255	sql = 'drop table _%s'%cn
256	self.sql(sql)
257
258	sql = 'drop sequence _%s_ids'%cn
259	self.sql(sql)
260
261	def newid(self, classname):
262	sql = "select nextval('_%s_ids') from dual"%classname
263	self.sql(sql)
264	return str(self.cursor.fetchone()[0])
265
266	def setid(self, classname, setid):
267	sql = "select setval('_%s_ids', %s) from dual"%(classname, int(setid))
268	self.sql(sql)
269
270	def clear(self):
271	rdbms_common.Database.clear(self)
272
273	# reset the sequences
274	for cn in self.classes:
275	self.cursor.execute('DROP SEQUENCE _%s_ids'%cn)
276	self.cursor.execute('CREATE SEQUENCE _%s_ids'%cn)
277
278	class PostgresqlClass:
279	order_by_null_values = '(%s is not NULL)'
280
281	class Class(PostgresqlClass, rdbms_common.Class):
282	pass
283	class IssueClass(PostgresqlClass, rdbms_common.IssueClass):
284	pass
285	class FileClass(PostgresqlClass, rdbms_common.FileClass):
286	pass
287
288	# vim: set et sts=4 sw=4 :