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('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('hyperdb').info('open database %r'%db['database'])
135 try:
136 conn = psycopg.connect(**db)
137 except psycopg.OperationalError, message:
138 raise hyperdb.DatabaseError(message)
140 cursor = conn.cursor()
142 return (conn, cursor)
144 def open_connection(self):
145 if not db_exists(self.config):
146 db_create(self.config)
148 self.conn, self.cursor = self.sql_open_connection()
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()
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)')
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)''')
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)')
187 def fix_version_2_tables(self):
188 # Convert journal date column to TIMESTAMP, params column to TEXT
189 self._convert_journal_tables()
191 # Convert all String properties to TEXT
192 self._convert_string_properties()
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))
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)''')
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)
215 def __repr__(self):
216 return '<roundpsycopgsql 0x%x>' % id(self)
218 def sql_commit(self, fail_ok=False):
219 ''' Actually commit to the database.
220 '''
221 logging.getLogger('hyperdb').info('commit')
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
233 # open a new cursor for subsequent work
234 self.cursor = self.conn.cursor()
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]
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]
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)
252 return rdbms_common.Database.create_class_table(self, spec)
254 def drop_class_table(self, cn):
255 sql = 'drop table _%s'%cn
256 self.sql(sql)
258 sql = 'drop sequence _%s_ids'%cn
259 self.sql(sql)
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])
266 def setid(self, classname, setid):
267 sql = "select setval('_%s_ids', %s) from dual"%(classname, int(setid))
268 self.sql(sql)
270 def clear(self):
271 rdbms_common.Database.clear(self)
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)
278 class PostgresqlClass:
279 order_by_null_values = '(%s is not NULL)'
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
288 # vim: set et sts=4 sw=4 :