d43659e44bcb6a0c6927b3ab790b80399fa8f196
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'
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
23 from roundup import hyperdb, date
24 from roundup.backends import rdbms_common
25 from roundup.backends import sessions_rdbms
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
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)
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)
48 if os.path.exists(config.DATABASE):
49 shutil.rmtree(config.DATABASE)
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'
58 try:
59 conn = psycopg.connect(**template1)
60 except psycopg.OperationalError, message:
61 raise hyperdb.DatabaseError(message)
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')
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.
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
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
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()
123 class Database(rdbms_common.Database):
124 arg = '%s'
126 # used by some code to switch styles of query
127 implements_intersect = 1
129 def getSessionManager(self):
130 return Sessions(self)
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)
141 cursor = conn.cursor()
143 return (conn, cursor)
145 def open_connection(self):
146 if not db_exists(self.config):
147 db_create(self.config)
149 self.conn, self.cursor = self.sql_open_connection()
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()
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)')
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)''')
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)')
188 def fix_version_2_tables(self):
189 # Convert journal date column to TIMESTAMP, params column to TEXT
190 self._convert_journal_tables()
192 # Convert all String properties to TEXT
193 self._convert_string_properties()
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))
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)''')
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)
216 def __repr__(self):
217 return '<roundpsycopgsql 0x%x>' % id(self)
219 def sql_commit(self, fail_ok=False):
220 ''' Actually commit to the database.
221 '''
222 logging.getLogger('roundup.hyperdb').info('commit')
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
235 # open a new cursor for subsequent work
236 self.cursor = self.conn.cursor()
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]
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]
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)
254 return rdbms_common.Database.create_class_table(self, spec)
256 def drop_class_table(self, cn):
257 sql = 'drop table _%s'%cn
258 self.sql(sql)
260 sql = 'drop sequence _%s_ids'%cn
261 self.sql(sql)
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])
268 def setid(self, classname, setid):
269 sql = "select setval('_%s_ids', %s) from dual"%(classname, int(setid))
270 self.sql(sql)
272 def clear(self):
273 rdbms_common.Database.clear(self)
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)
280 class PostgresqlClass:
281 order_by_null_values = '(%s is not NULL)'
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
290 # vim: set et sts=4 sw=4 :