Project

General

Profile

1 11 aaronmk
# Database access
2
3 1869 aaronmk
import copy
4 11 aaronmk
import re
5 3238 aaronmk
import time
6 865 aaronmk
import warnings
7 11 aaronmk
8 300 aaronmk
import exc
9 1909 aaronmk
import dicts
10 1893 aaronmk
import iters
11 1960 aaronmk
import lists
12 3241 aaronmk
import profiling
13 1889 aaronmk
from Proxy import Proxy
14 1872 aaronmk
import rand
15 2217 aaronmk
import sql_gen
16 862 aaronmk
import strings
17 131 aaronmk
import util
18 11 aaronmk
19 832 aaronmk
##### Exceptions
20
21 2804 aaronmk
def get_cur_query(cur, input_query=None):
22 2168 aaronmk
    raw_query = None
23
    if hasattr(cur, 'query'): raw_query = cur.query
24
    elif hasattr(cur, '_last_executed'): raw_query = cur._last_executed
25 2170 aaronmk
26
    if raw_query != None: return raw_query
27 2804 aaronmk
    else: return '[input] '+strings.ustr(input_query)
28 14 aaronmk
29 2170 aaronmk
def _add_cursor_info(e, *args, **kw_args):
30
    '''For params, see get_cur_query()'''
31 2771 aaronmk
    exc.add_msg(e, 'query: '+strings.ustr(get_cur_query(*args, **kw_args)))
32 135 aaronmk
33 300 aaronmk
class DbException(exc.ExceptionWithCause):
34 14 aaronmk
    def __init__(self, msg, cause=None, cur=None):
35 2145 aaronmk
        exc.ExceptionWithCause.__init__(self, msg, cause, cause_newline=True)
36 14 aaronmk
        if cur != None: _add_cursor_info(self, cur)
37
38 2143 aaronmk
class ExceptionWithName(DbException):
39
    def __init__(self, name, cause=None):
40 4491 aaronmk
        DbException.__init__(self, 'for name: '
41
            +strings.as_tt(strings.ustr(name)), cause)
42 2143 aaronmk
        self.name = name
43 360 aaronmk
44 3109 aaronmk
class ExceptionWithValue(DbException):
45
    def __init__(self, value, cause=None):
46 4492 aaronmk
        DbException.__init__(self, 'for value: '
47
            +strings.as_tt(strings.urepr(value)), cause)
48 2240 aaronmk
        self.value = value
49
50 2945 aaronmk
class ExceptionWithNameType(DbException):
51
    def __init__(self, type_, name, cause=None):
52 4491 aaronmk
        DbException.__init__(self, 'for type: '+strings.as_tt(strings.ustr(
53
            type_))+'; name: '+strings.as_tt(name), cause)
54 2945 aaronmk
        self.type = type_
55
        self.name = name
56
57 2306 aaronmk
class ConstraintException(DbException):
58 3345 aaronmk
    def __init__(self, name, cond, cols, cause=None):
59
        msg = 'Violated '+strings.as_tt(name)+' constraint'
60 5447 aaronmk
        if cond != None: msg += ' with condition '+strings.as_tt(cond)
61 3345 aaronmk
        if cols != []: msg += ' on columns: '+strings.as_tt(', '.join(cols))
62
        DbException.__init__(self, msg, cause)
63 2306 aaronmk
        self.name = name
64 3345 aaronmk
        self.cond = cond
65 468 aaronmk
        self.cols = cols
66 11 aaronmk
67 2523 aaronmk
class MissingCastException(DbException):
68 4139 aaronmk
    def __init__(self, type_, col=None, cause=None):
69
        msg = 'Missing cast to type '+strings.as_tt(type_)
70
        if col != None: msg += ' on column: '+strings.as_tt(col)
71
        DbException.__init__(self, msg, cause)
72 2523 aaronmk
        self.type = type_
73
        self.col = col
74
75 5576 aaronmk
class EncodingException(ExceptionWithName): pass
76
77 2306 aaronmk
class DuplicateKeyException(ConstraintException): pass
78 13 aaronmk
79 2306 aaronmk
class NullValueException(ConstraintException): pass
80 13 aaronmk
81 3346 aaronmk
class CheckException(ConstraintException): pass
82
83 3109 aaronmk
class InvalidValueException(ExceptionWithValue): pass
84 2239 aaronmk
85 7112 aaronmk
class InvalidTypeException(ExceptionWithNameType): pass
86
87 2945 aaronmk
class DuplicateException(ExceptionWithNameType): pass
88 2143 aaronmk
89 3419 aaronmk
class DoesNotExistException(ExceptionWithNameType): pass
90
91 89 aaronmk
class EmptyRowException(DbException): pass
92
93 865 aaronmk
##### Warnings
94
95
class DbWarning(UserWarning): pass
96
97 1930 aaronmk
##### Result retrieval
98
99
def col_names(cur): return (col[0] for col in cur.description)
100
101
def rows(cur): return iter(lambda: cur.fetchone(), None)
102
103
def consume_rows(cur):
104
    '''Used to fetch all rows so result will be cached'''
105
    iters.consume_iter(rows(cur))
106
107
def next_row(cur): return rows(cur).next()
108
109
def row(cur):
110
    row_ = next_row(cur)
111
    consume_rows(cur)
112
    return row_
113
114
def next_value(cur): return next_row(cur)[0]
115
116
def value(cur): return row(cur)[0]
117
118
def values(cur): return iters.func_iter(lambda: next_value(cur))
119
120
def value_or_none(cur):
121
    try: return value(cur)
122
    except StopIteration: return None
123
124 2762 aaronmk
##### Escaping
125 2101 aaronmk
126 2573 aaronmk
def esc_name_by_module(module, name):
127
    if module == 'psycopg2' or module == None: quote = '"'
128 2101 aaronmk
    elif module == 'MySQLdb': quote = '`'
129
    else: raise NotImplementedError("Can't escape name for "+module+' database')
130 2500 aaronmk
    return sql_gen.esc_name(name, quote)
131 2101 aaronmk
132
def esc_name_by_engine(engine, name, **kw_args):
133
    return esc_name_by_module(db_engines[engine][0], name, **kw_args)
134
135
def esc_name(db, name, **kw_args):
136
    return esc_name_by_module(util.root_module(db.db), name, **kw_args)
137
138
def qual_name(db, schema, table):
139
    def esc_name_(name): return esc_name(db, name)
140
    table = esc_name_(table)
141
    if schema != None: return esc_name_(schema)+'.'+table
142
    else: return table
143
144 1869 aaronmk
##### Database connections
145 1849 aaronmk
146 2097 aaronmk
db_config_names = ['engine', 'host', 'user', 'password', 'database', 'schemas']
147 1926 aaronmk
148 1869 aaronmk
db_engines = {
149
    'MySQL': ('MySQLdb', {'password': 'passwd', 'database': 'db'}),
150
    'PostgreSQL': ('psycopg2', {}),
151
}
152
153
DatabaseErrors_set = set([DbException])
154
DatabaseErrors = tuple(DatabaseErrors_set)
155
156
def _add_module(module):
157
    DatabaseErrors_set.add(module.DatabaseError)
158
    global DatabaseErrors
159
    DatabaseErrors = tuple(DatabaseErrors_set)
160
161
def db_config_str(db_config):
162
    return db_config['engine']+' database '+db_config['database']
163
164 2448 aaronmk
log_debug_none = lambda msg, level=2: None
165 1901 aaronmk
166 1849 aaronmk
class DbConn:
167 2923 aaronmk
    def __init__(self, db_config, autocommit=True, caching=True,
168 3183 aaronmk
        log_debug=log_debug_none, debug_temp=False, src=None):
169 2915 aaronmk
        '''
170
        @param debug_temp Whether temporary objects should instead be permanent.
171
            This assists in debugging the internal objects used by the program.
172 3183 aaronmk
        @param src In autocommit mode, will be included in a comment in every
173
            query, to help identify the data source in pg_stat_activity.
174 2915 aaronmk
        '''
175 1869 aaronmk
        self.db_config = db_config
176 2190 aaronmk
        self.autocommit = autocommit
177
        self.caching = caching
178 1901 aaronmk
        self.log_debug = log_debug
179 2193 aaronmk
        self.debug = log_debug != log_debug_none
180 2915 aaronmk
        self.debug_temp = debug_temp
181 3183 aaronmk
        self.src = src
182 3074 aaronmk
        self.autoanalyze = False
183 3269 aaronmk
        self.autoexplain = False
184
        self.profile_row_ct = None
185 1869 aaronmk
186 3124 aaronmk
        self._savepoint = 0
187 3120 aaronmk
        self._reset()
188 1869 aaronmk
189
    def __getattr__(self, name):
190
        if name == '__dict__': raise Exception('getting __dict__')
191
        if name == 'db': return self._db()
192
        else: raise AttributeError()
193
194
    def __getstate__(self):
195
        state = copy.copy(self.__dict__) # shallow copy
196 1915 aaronmk
        state['log_debug'] = None # don't pickle the debug callback
197 1869 aaronmk
        state['_DbConn__db'] = None # don't pickle the connection
198
        return state
199
200 3118 aaronmk
    def clear_cache(self): self.query_results = {}
201
202 3120 aaronmk
    def _reset(self):
203 3118 aaronmk
        self.clear_cache()
204 3124 aaronmk
        assert self._savepoint == 0
205 3118 aaronmk
        self._notices_seen = set()
206
        self.__db = None
207
208 2165 aaronmk
    def connected(self): return self.__db != None
209
210 3116 aaronmk
    def close(self):
211 3119 aaronmk
        if not self.connected(): return
212
213 3135 aaronmk
        # Record that the automatic transaction is now closed
214 3136 aaronmk
        self._savepoint -= 1
215 3135 aaronmk
216 3119 aaronmk
        self.db.close()
217 3120 aaronmk
        self._reset()
218 3116 aaronmk
219 3125 aaronmk
    def reconnect(self):
220
        # Do not do this in test mode as it would roll back everything
221
        if self.autocommit: self.close()
222
        # Connection will be reopened automatically on first query
223
224 1869 aaronmk
    def _db(self):
225
        if self.__db == None:
226
            # Process db_config
227
            db_config = self.db_config.copy() # don't modify input!
228 2097 aaronmk
            schemas = db_config.pop('schemas', None)
229 1869 aaronmk
            module_name, mappings = db_engines[db_config.pop('engine')]
230
            module = __import__(module_name)
231
            _add_module(module)
232
            for orig, new in mappings.iteritems():
233
                try: util.rename_key(db_config, orig, new)
234
                except KeyError: pass
235
236
            # Connect
237
            self.__db = module.connect(**db_config)
238
239 3161 aaronmk
            # Record that a transaction is already open
240
            self._savepoint += 1
241
242 1869 aaronmk
            # Configure connection
243 2906 aaronmk
            if hasattr(self.db, 'set_isolation_level'):
244
                import psycopg2.extensions
245
                self.db.set_isolation_level(
246
                    psycopg2.extensions.ISOLATION_LEVEL_READ_COMMITTED)
247 2101 aaronmk
            if schemas != None:
248 2893 aaronmk
                search_path = [self.esc_name(s) for s in schemas.split(',')]
249
                run_query(self, 'SET search_path TO '+(','.join(search_path)),
250
                    log_level=3)
251 1869 aaronmk
252
        return self.__db
253 1889 aaronmk
254 1891 aaronmk
    class DbCursor(Proxy):
255 1927 aaronmk
        def __init__(self, outer):
256 1891 aaronmk
            Proxy.__init__(self, outer.db.cursor())
257 2191 aaronmk
            self.outer = outer
258 1927 aaronmk
            self.query_results = outer.query_results
259 1894 aaronmk
            self.query_lookup = None
260 1891 aaronmk
            self.result = []
261 1889 aaronmk
262 2802 aaronmk
        def execute(self, query):
263 2764 aaronmk
            self._is_insert = query.startswith('INSERT')
264 2797 aaronmk
            self.query_lookup = query
265 2148 aaronmk
            try:
266 3162 aaronmk
                try: cur = self.inner.execute(query)
267 2802 aaronmk
                finally: self.query = get_cur_query(self.inner, query)
268 1904 aaronmk
            except Exception, e:
269
                self.result = e # cache the exception as the result
270
                self._cache_result()
271
                raise
272 3004 aaronmk
273
            # Always cache certain queries
274 3183 aaronmk
            query = sql_gen.lstrip(query)
275 3004 aaronmk
            if query.startswith('CREATE') or query.startswith('ALTER'):
276 3007 aaronmk
                # structural changes
277 3040 aaronmk
                # Rest of query must be unique in the face of name collisions,
278
                # so don't cache ADD COLUMN unless it has distinguishing comment
279
                if query.find('ADD COLUMN') < 0 or query.endswith('*/'):
280 3007 aaronmk
                    self._cache_result()
281 3004 aaronmk
            elif self.rowcount == 0 and query.startswith('SELECT'): # empty
282 2800 aaronmk
                consume_rows(self) # fetch all rows so result will be cached
283 3004 aaronmk
284 2762 aaronmk
            return cur
285 1894 aaronmk
286 1891 aaronmk
        def fetchone(self):
287
            row = self.inner.fetchone()
288 1899 aaronmk
            if row != None: self.result.append(row)
289
            # otherwise, fetched all rows
290 1904 aaronmk
            else: self._cache_result()
291
            return row
292
293
        def _cache_result(self):
294 2948 aaronmk
            # For inserts that return a result set, don't cache result set since
295
            # inserts are not idempotent. Other non-SELECT queries don't have
296
            # their result set read, so only exceptions will be cached (an
297
            # invalid query will always be invalid).
298 1930 aaronmk
            if self.query_results != None and (not self._is_insert
299 1906 aaronmk
                or isinstance(self.result, Exception)):
300
301 1894 aaronmk
                assert self.query_lookup != None
302 1916 aaronmk
                self.query_results[self.query_lookup] = self.CacheCursor(
303
                    util.dict_subset(dicts.AttrsDictView(self),
304
                    ['query', 'result', 'rowcount', 'description']))
305 1906 aaronmk
306 1916 aaronmk
        class CacheCursor:
307
            def __init__(self, cached_result): self.__dict__ = cached_result
308
309 1927 aaronmk
            def execute(self, *args, **kw_args):
310 1916 aaronmk
                if isinstance(self.result, Exception): raise self.result
311
                # otherwise, result is a rows list
312
                self.iter = iter(self.result)
313
314
            def fetchone(self):
315
                try: return self.iter.next()
316
                except StopIteration: return None
317 1891 aaronmk
318 2212 aaronmk
    def esc_value(self, value):
319 2663 aaronmk
        try: str_ = self.mogrify('%s', [value])
320
        except NotImplementedError, e:
321
            module = util.root_module(self.db)
322
            if module == 'MySQLdb':
323
                import _mysql
324
                str_ = _mysql.escape_string(value)
325
            else: raise e
326 2374 aaronmk
        return strings.to_unicode(str_)
327 2212 aaronmk
328 2347 aaronmk
    def esc_name(self, name): return esc_name(self, name) # calls global func
329
330 2814 aaronmk
    def std_code(self, str_):
331
        '''Standardizes SQL code.
332
        * Ensures that string literals are prefixed by `E`
333
        '''
334
        if str_.startswith("'"): str_ = 'E'+str_
335
        return str_
336
337 2665 aaronmk
    def can_mogrify(self):
338 2663 aaronmk
        module = util.root_module(self.db)
339 2665 aaronmk
        return module == 'psycopg2'
340 2663 aaronmk
341 2665 aaronmk
    def mogrify(self, query, params=None):
342
        if self.can_mogrify(): return self.db.cursor().mogrify(query, params)
343
        else: raise NotImplementedError("Can't mogrify query")
344
345 5579 aaronmk
    def set_encoding(self, encoding):
346
        encoding_str = sql_gen.Literal(encoding)
347
        run_query(self, 'SET NAMES '+encoding_str.to_str(self))
348
349 2671 aaronmk
    def print_notices(self):
350 2725 aaronmk
        if hasattr(self.db, 'notices'):
351
            for msg in self.db.notices:
352
                if msg not in self._notices_seen:
353
                    self._notices_seen.add(msg)
354
                    self.log_debug(msg, level=2)
355 2671 aaronmk
356 2793 aaronmk
    def run_query(self, query, cacheable=False, log_level=2,
357 2464 aaronmk
        debug_msg_ref=None):
358 2445 aaronmk
        '''
359 2464 aaronmk
        @param log_ignore_excs The log_level will be increased by 2 if the query
360
            throws one of these exceptions.
361 2664 aaronmk
        @param debug_msg_ref If specified, the log message will be returned in
362
            this instead of being output. This allows you to filter log messages
363
            depending on the result of the query.
364 2445 aaronmk
        '''
365 2167 aaronmk
        assert query != None
366
367 3183 aaronmk
        if self.autocommit and self.src != None:
368 3206 aaronmk
            query = sql_gen.esc_comment(self.src)+'\t'+query
369 3183 aaronmk
370 2047 aaronmk
        if not self.caching: cacheable = False
371 1903 aaronmk
        used_cache = False
372 2664 aaronmk
373 3242 aaronmk
        if self.debug:
374
            profiler = profiling.ItersProfiler(start_now=True, iter_text='row')
375 1903 aaronmk
        try:
376 1927 aaronmk
            # Get cursor
377
            if cacheable:
378 3238 aaronmk
                try: cur = self.query_results[query]
379 1927 aaronmk
                except KeyError: cur = self.DbCursor(self)
380 3238 aaronmk
                else: used_cache = True
381 1927 aaronmk
            else: cur = self.db.cursor()
382
383
            # Run query
384 3238 aaronmk
            try: cur.execute(query)
385 3162 aaronmk
            except Exception, e:
386
                _add_cursor_info(e, self, query)
387
                raise
388 3238 aaronmk
            else: self.do_autocommit()
389 1903 aaronmk
        finally:
390 3242 aaronmk
            if self.debug:
391 3244 aaronmk
                profiler.stop(self.profile_row_ct)
392 3242 aaronmk
393
                ## Log or return query
394
395 4491 aaronmk
                query = strings.ustr(get_cur_query(cur, query))
396 3281 aaronmk
                # Put the src comment on a separate line in the log file
397
                query = query.replace('\t', '\n', 1)
398 3239 aaronmk
399 3240 aaronmk
                msg = 'DB query: '
400 3239 aaronmk
401 3240 aaronmk
                if used_cache: msg += 'cache hit'
402
                elif cacheable: msg += 'cache miss'
403
                else: msg += 'non-cacheable'
404 3239 aaronmk
405 3241 aaronmk
                msg += ':\n'+profiler.msg()+'\n'+strings.as_code(query, 'SQL')
406 3240 aaronmk
407 3237 aaronmk
                if debug_msg_ref != None: debug_msg_ref[0] = msg
408
                else: self.log_debug(msg, log_level)
409 3245 aaronmk
410
                self.print_notices()
411 1903 aaronmk
412
        return cur
413 1914 aaronmk
414 2797 aaronmk
    def is_cached(self, query): return query in self.query_results
415 2139 aaronmk
416 2907 aaronmk
    def with_autocommit(self, func):
417 2801 aaronmk
        import psycopg2.extensions
418
419
        prev_isolation_level = self.db.isolation_level
420 2907 aaronmk
        self.db.set_isolation_level(
421
            psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
422 2683 aaronmk
        try: return func()
423 2801 aaronmk
        finally: self.db.set_isolation_level(prev_isolation_level)
424 2683 aaronmk
425 2139 aaronmk
    def with_savepoint(self, func):
426 3137 aaronmk
        top = self._savepoint == 0
427 2171 aaronmk
        savepoint = 'level_'+str(self._savepoint)
428 3137 aaronmk
429 3272 aaronmk
        if self.debug:
430 3273 aaronmk
            self.log_debug('Begin transaction', level=4)
431 3272 aaronmk
            profiler = profiling.ItersProfiler(start_now=True, iter_text='row')
432
433 3160 aaronmk
        # Must happen before running queries so they don't get autocommitted
434
        self._savepoint += 1
435
436 3137 aaronmk
        if top: query = 'START TRANSACTION ISOLATION LEVEL READ COMMITTED'
437
        else: query = 'SAVEPOINT '+savepoint
438
        self.run_query(query, log_level=4)
439
        try:
440
            return func()
441
            if top: self.run_query('COMMIT', log_level=4)
442 2139 aaronmk
        except:
443 3137 aaronmk
            if top: query = 'ROLLBACK'
444
            else: query = 'ROLLBACK TO SAVEPOINT '+savepoint
445
            self.run_query(query, log_level=4)
446
447 2139 aaronmk
            raise
448 2930 aaronmk
        finally:
449
            # Always release savepoint, because after ROLLBACK TO SAVEPOINT,
450
            # "The savepoint remains valid and can be rolled back to again"
451
            # (http://www.postgresql.org/docs/8.3/static/sql-rollback-to.html).
452 3137 aaronmk
            if not top:
453
                self.run_query('RELEASE SAVEPOINT '+savepoint, log_level=4)
454 2930 aaronmk
455
            self._savepoint -= 1
456
            assert self._savepoint >= 0
457
458 3272 aaronmk
            if self.debug:
459
                profiler.stop(self.profile_row_ct)
460 3273 aaronmk
                self.log_debug('End transaction\n'+profiler.msg(), level=4)
461 3272 aaronmk
462 2930 aaronmk
            self.do_autocommit() # OK to do this after ROLLBACK TO SAVEPOINT
463 2191 aaronmk
464
    def do_autocommit(self):
465
        '''Autocommits if outside savepoint'''
466 3135 aaronmk
        assert self._savepoint >= 1
467
        if self.autocommit and self._savepoint == 1:
468 2924 aaronmk
            self.log_debug('Autocommitting', level=4)
469 2191 aaronmk
            self.db.commit()
470 2643 aaronmk
471 3155 aaronmk
    def col_info(self, col, cacheable=True):
472 6900 aaronmk
        module = util.root_module(self.db)
473 6899 aaronmk
        if module == 'psycopg2':
474 6900 aaronmk
            qual_table = sql_gen.Literal(col.table.to_str(self))
475 6899 aaronmk
            col_name_str = sql_gen.Literal(col.name)
476
            try:
477 6900 aaronmk
                type_, is_array, default, nullable = row(run_query(self, '''\
478 6899 aaronmk
SELECT
479
format_type(COALESCE(NULLIF(typelem, 0), pg_type.oid), -1) AS type
480
, typcategory = 'A' AS type_is_array
481 6903 aaronmk
, pg_get_expr(pg_attrdef.adbin, attrelid, true) AS default
482 6899 aaronmk
, NOT pg_attribute.attnotnull AS nullable
483
FROM pg_attribute
484
LEFT JOIN pg_type ON pg_type.oid = atttypid
485
LEFT JOIN pg_attrdef ON adrelid = attrelid AND adnum = attnum
486
WHERE
487 6901 aaronmk
attrelid = '''+qual_table.to_str(self)+'''::regclass
488
AND attname = '''+col_name_str.to_str(self)+'''
489 6899 aaronmk
'''
490 6904 aaronmk
                    , recover=True, cacheable=cacheable, log_level=4))
491 6905 aaronmk
            except (DoesNotExistException, StopIteration):
492
                raise sql_gen.NoUnderlyingTableException(col)
493 6900 aaronmk
            if is_array: type_ = sql_gen.ArrayType(type_)
494 6899 aaronmk
        else:
495
            table = sql_gen.Table('columns', 'information_schema')
496
            cols = [sql_gen.Col('data_type'), sql_gen.Col('udt_name'),
497
                'column_default', sql_gen.Cast('boolean',
498
                sql_gen.Col('is_nullable'))]
499
500
            conds = [('table_name', col.table.name),
501
                ('column_name', strings.ustr(col.name))]
502
            schema = col.table.schema
503
            if schema != None: conds.append(('table_schema', schema))
504
505
            cur = select(self, table, cols, conds, order_by='table_schema',
506
                limit=1, cacheable=cacheable, log_level=4)
507
            try: type_, extra_type, default, nullable = row(cur)
508
            except StopIteration: raise sql_gen.NoUnderlyingTableException(col)
509
            if type_ == 'USER-DEFINED': type_ = extra_type
510
            elif type_ == 'ARRAY':
511
                type_ = sql_gen.ArrayType(strings.remove_prefix('_', extra_type,
512
                    require=True))
513 2643 aaronmk
514 6899 aaronmk
        if default != None: default = sql_gen.as_Code(default, self)
515 2819 aaronmk
        return sql_gen.TypedCol(col.name, type_, default, nullable)
516 2917 aaronmk
517
    def TempFunction(self, name):
518
        if self.debug_temp: schema = None
519
        else: schema = 'pg_temp'
520
        return sql_gen.Function(name, schema)
521 1849 aaronmk
522 1869 aaronmk
connect = DbConn
523
524 832 aaronmk
##### Recoverable querying
525 15 aaronmk
526 5577 aaronmk
def parse_exception(db, e, recover=False):
527
    msg = strings.ustr(e.args[0])
528
    msg = re.sub(r'^(?:PL/Python: )?ValueError: ', r'', msg)
529
530
    match = re.match(r'^invalid byte sequence for encoding "(.+?)":', msg)
531
    if match:
532
        encoding, = match.groups()
533
        raise EncodingException(encoding, e)
534
535 5763 aaronmk
    def make_DuplicateKeyException(constraint, e):
536 5577 aaronmk
        cols = []
537
        cond = None
538
        if recover: # need auto-rollback to run index_cols()
539
            try:
540
                cols = index_cols(db, constraint)
541
                cond = index_cond(db, constraint)
542
            except NotImplementedError: pass
543 9757 aaronmk
            except DatabaseErrors, e2: # e.g. "current transaction is aborted"
544
                exc.print_ex(e2)
545
                raise e
546 5763 aaronmk
        return DuplicateKeyException(constraint, cond, cols, e)
547 5577 aaronmk
548 5763 aaronmk
    match = re.match(r'^duplicate key value violates unique constraint "(.+?)"',
549
        msg)
550
    if match:
551
        constraint, = match.groups()
552
        raise make_DuplicateKeyException(constraint, e)
553
554 5764 aaronmk
    match = re.match(r'^could not create unique index "(.+?)"\n'
555
        r'DETAIL:  Key .+? is duplicated', msg)
556
    if match:
557
        constraint, = match.groups()
558 5823 aaronmk
        raise DuplicateKeyException(constraint, None, [], e)
559 5764 aaronmk
560 5577 aaronmk
    match = re.match(r'^null value in column "(.+?)" violates not-null'
561
        r' constraint', msg)
562
    if match:
563
        col, = match.groups()
564
        raise NullValueException('NOT NULL', None, [col], e)
565
566
    match = re.match(r'^new row for relation "(.+?)" violates check '
567
        r'constraint "(.+?)"', msg)
568
    if match:
569
        table, constraint = match.groups()
570
        constraint = sql_gen.Col(constraint, table)
571
        cond = None
572
        if recover: # need auto-rollback to run constraint_cond()
573
            try: cond = constraint_cond(db, constraint)
574
            except NotImplementedError: pass
575
        raise CheckException(constraint.to_str(db), cond, [], e)
576
577 5723 aaronmk
    match = re.match(r'^(?:invalid input (?:syntax|value)\b[^:]*'
578 5717 aaronmk
        r'|.+? out of range)(?:: "(.+?)")?', msg)
579 5577 aaronmk
    if match:
580
        value, = match.groups()
581 5717 aaronmk
        value = util.do_ignore_none(strings.to_unicode, value)
582
        raise InvalidValueException(value, e)
583 5577 aaronmk
584
    match = re.match(r'^column "(.+?)" is of type (.+?) but expression '
585
        r'is of type', msg)
586
    if match:
587
        col, type_ = match.groups()
588
        raise MissingCastException(type_, col, e)
589
590
    match = re.match(r'^could not determine polymorphic type because '
591
        r'input has type "unknown"', msg)
592
    if match: raise MissingCastException('text', None, e)
593
594 5724 aaronmk
    match = re.match(r'^.+? types (.+?) and (.+?) cannot be matched', msg)
595
    if match:
596
        type0, type1 = match.groups()
597
        raise MissingCastException(type0, None, e)
598 5577 aaronmk
599 7112 aaronmk
    match = re.match(r'^.*?\brelation "(.+?)" is not a (table)', msg)
600
    if match:
601
        name, type_ = match.groups()
602
        raise InvalidTypeException(type_, name, e)
603
604 9458 aaronmk
    typed_name_re = r'^(\S+) ("?)(.+?)\2(?: of relation ".+?")?'
605 9457 aaronmk
        # regexp must be followed with text for .*? to match properly
606 5577 aaronmk
607 9458 aaronmk
    match = re.match(typed_name_re+r'(?:\(.*?)? already exists', msg)
608 5577 aaronmk
    if match:
609 9458 aaronmk
        type_, quote, name = match.groups()
610 5577 aaronmk
        raise DuplicateException(type_, name, e)
611
612
    match = re.match(r'more than one (\S+) named ""(.+?)""', msg)
613
    if match:
614
        type_, name = match.groups()
615
        raise DuplicateException(type_, name, e)
616
617
    match = re.match(typed_name_re+r' does not exist', msg)
618
    if match:
619 9458 aaronmk
        type_, quote, name = match.groups()
620 5683 aaronmk
        if type_ == 'function':
621 5709 aaronmk
            match = re.match(r'^(.+?)\(.*\)$', name)
622 5710 aaronmk
            if match: # includes params, so is call rather than cast to regproc
623 5708 aaronmk
                function_name, = match.groups()
624 5711 aaronmk
                func = sql_gen.Function(function_name)
625 5712 aaronmk
                if function_exists(db, func) and msg.find('CAST') < 0:
626 5683 aaronmk
                    # not found only because of a missing cast
627 10548 aaronmk
                    try: type_ = function_param0_type(db, func)
628
                    except DuplicateException: type_ = 'text'
629 6301 aaronmk
                    col = None
630 5792 aaronmk
                    if type_ == 'anyelement': type_ = 'text'
631 6301 aaronmk
                    elif type_ == 'hstore': # cast just the value param
632
                        type_ = 'text'
633
                        col = 'value'
634
                    raise MissingCastException(type_, col, e)
635 5577 aaronmk
        raise DoesNotExistException(type_, name, e)
636
637
    raise # no specific exception raised
638
639 2139 aaronmk
def with_savepoint(db, func): return db.with_savepoint(func)
640 11 aaronmk
641 2791 aaronmk
def run_query(db, query, recover=None, cacheable=False, log_level=2,
642
    log_ignore_excs=None, **kw_args):
643 2794 aaronmk
    '''For params, see DbConn.run_query()'''
644 830 aaronmk
    if recover == None: recover = False
645 2464 aaronmk
    if log_ignore_excs == None: log_ignore_excs = ()
646
    log_ignore_excs = tuple(log_ignore_excs)
647 3236 aaronmk
    debug_msg_ref = [None]
648 830 aaronmk
649 3267 aaronmk
    query = with_explain_comment(db, query)
650 3258 aaronmk
651 2148 aaronmk
    try:
652 2464 aaronmk
        try:
653 2794 aaronmk
            def run(): return db.run_query(query, cacheable, log_level,
654 2793 aaronmk
                debug_msg_ref, **kw_args)
655 2796 aaronmk
            if recover and not db.is_cached(query):
656 2464 aaronmk
                return with_savepoint(db, run)
657
            else: return run() # don't need savepoint if cached
658 5890 aaronmk
        except Exception, e:
659
            # Give failed EXPLAIN approximately the log_level of its query
660
            if query.startswith('EXPLAIN'): log_level -= 1
661
662
            parse_exception(db, e, recover)
663 2464 aaronmk
    except log_ignore_excs:
664
        log_level += 2
665
        raise
666
    finally:
667 3236 aaronmk
        if debug_msg_ref[0] != None: db.log_debug(debug_msg_ref[0], log_level)
668 830 aaronmk
669 832 aaronmk
##### Basic queries
670
671 3256 aaronmk
def is_explainable(query):
672
    # See <http://www.postgresql.org/docs/8.3/static/sql-explain.html#AEN57749>
673 3257 aaronmk
    return re.match(r'^(?:SELECT|INSERT|UPDATE|DELETE|VALUES|EXECUTE|DECLARE)\b'
674
        , query)
675 3256 aaronmk
676 3263 aaronmk
def explain(db, query, **kw_args):
677
    '''
678
    For params, see run_query().
679
    '''
680 3267 aaronmk
    kw_args.setdefault('log_level', 4)
681 3263 aaronmk
682 3750 aaronmk
    return strings.ustr(strings.join_lines(values(run_query(db,
683
        'EXPLAIN '+query, recover=True, cacheable=True, **kw_args))))
684 3256 aaronmk
        # not a higher log_level because it's useful to see what query is being
685
        # run before it's executed, which EXPLAIN effectively provides
686
687 3265 aaronmk
def has_comment(query): return query.endswith('*/')
688
689
def with_explain_comment(db, query, **kw_args):
690 3269 aaronmk
    if db.autoexplain and not has_comment(query) and is_explainable(query):
691 3265 aaronmk
        query += '\n'+sql_gen.esc_comment(' EXPLAIN:\n'
692
            +explain(db, query, **kw_args))
693
    return query
694
695 2153 aaronmk
def next_version(name):
696 2163 aaronmk
    version = 1 # first existing name was version 0
697 2586 aaronmk
    match = re.match(r'^(.*)#(\d+)$', name)
698 2153 aaronmk
    if match:
699 2586 aaronmk
        name, version = match.groups()
700
        version = int(version)+1
701 2932 aaronmk
    return sql_gen.concat(name, '#'+str(version))
702 2153 aaronmk
703 2899 aaronmk
def lock_table(db, table, mode):
704
    table = sql_gen.as_Table(table)
705
    run_query(db, 'LOCK TABLE '+table.to_str(db)+' IN '+mode+' MODE')
706
707 3303 aaronmk
def run_query_into(db, query, into=None, add_pkey_=False, **kw_args):
708 2085 aaronmk
    '''Outputs a query to a temp table.
709
    For params, see run_query().
710
    '''
711 2789 aaronmk
    if into == None: return run_query(db, query, **kw_args)
712 2790 aaronmk
713
    assert isinstance(into, sql_gen.Table)
714
715 2992 aaronmk
    into.is_temp = True
716 3008 aaronmk
    # "temporary tables cannot specify a schema name", so remove schema
717
    into.schema = None
718 2992 aaronmk
719 2790 aaronmk
    kw_args['recover'] = True
720 2945 aaronmk
    kw_args.setdefault('log_ignore_excs', (DuplicateException,))
721 2790 aaronmk
722 2916 aaronmk
    temp = not db.debug_temp # tables are permanent in debug_temp mode
723 2790 aaronmk
724
    # Create table
725
    while True:
726
        create_query = 'CREATE'
727
        if temp: create_query += ' TEMP'
728
        create_query += ' TABLE '+into.to_str(db)+' AS\n'+query
729 2385 aaronmk
730 2790 aaronmk
        try:
731
            cur = run_query(db, create_query, **kw_args)
732
                # CREATE TABLE AS sets rowcount to # rows in query
733
            break
734 2945 aaronmk
        except DuplicateException, e:
735 2790 aaronmk
            into.name = next_version(into.name)
736
            # try again with next version of name
737
738 5967 aaronmk
    if add_pkey_: add_pkey_or_index(db, into, warn=True)
739 3075 aaronmk
740
    # According to the PostgreSQL doc, "The autovacuum daemon cannot access and
741
    # therefore cannot vacuum or analyze temporary tables. [...] if a temporary
742
    # table is going to be used in complex queries, it is wise to run ANALYZE on
743
    # the temporary table after it is populated."
744
    # (http://www.postgresql.org/docs/9.1/static/sql-createtable.html)
745
    # If into is not a temp table, ANALYZE is useful but not required.
746 3073 aaronmk
    analyze(db, into)
747 2790 aaronmk
748
    return cur
749 2085 aaronmk
750 2120 aaronmk
order_by_pkey = object() # tells mk_select() to order by the pkey
751
752 2199 aaronmk
distinct_on_all = object() # tells mk_select() to SELECT DISTINCT ON all columns
753
754 8333 aaronmk
def has_subset_func(db, table):
755
    return sql_gen.is_underlying_table(table) and function_exists(db, table)
756
757 3420 aaronmk
def mk_select(db, tables=None, fields=None, conds=None, distinct_on=[],
758 3494 aaronmk
    limit=None, start=None, order_by=order_by_pkey, default_table=None,
759
    explain=True):
760 1981 aaronmk
    '''
761 2121 aaronmk
    @param tables The single table to select from, or a list of tables to join
762 2280 aaronmk
        together, with tables after the first being sql_gen.Join objects
763 1981 aaronmk
    @param fields Use None to select all fields in the table
764 2377 aaronmk
    @param conds WHERE conditions: [(compare_left_side, compare_right_side),...]
765 2379 aaronmk
        * container can be any iterable type
766 2399 aaronmk
        * compare_left_side: sql_gen.Code|str (for col name)
767
        * compare_right_side: sql_gen.ValueCond|literal value
768 2199 aaronmk
    @param distinct_on The columns to SELECT DISTINCT ON, or distinct_on_all to
769
        use all columns
770 2786 aaronmk
    @return query
771 1981 aaronmk
    '''
772 2315 aaronmk
    # Parse tables param
773 2964 aaronmk
    tables = lists.mk_seq(tables)
774 2141 aaronmk
    tables = list(tables) # don't modify input! (list() copies input)
775 2315 aaronmk
    table0 = sql_gen.as_Table(tables.pop(0)) # first table is separate
776 2121 aaronmk
777 2315 aaronmk
    # Parse other params
778 2376 aaronmk
    if conds == None: conds = []
779 2650 aaronmk
    elif dicts.is_dict(conds): conds = conds.items()
780 2379 aaronmk
    conds = list(conds) # don't modify input! (list() copies input)
781 3129 aaronmk
    assert limit == None or isinstance(limit, (int, long))
782
    assert start == None or isinstance(start, (int, long))
783 5527 aaronmk
    if limit == 0: order_by = None
784 8334 aaronmk
    if order_by in (None, order_by_pkey) and has_subset_func(db, table0):
785
        # can use subset function for fast querying at large OFFSET values
786 8335 aaronmk
        run_query(db, 'SET LOCAL enable_sort TO off')
787 8334 aaronmk
        table0 = sql_gen.FunctionCall(table0, limit_=limit, offset_=start)
788
        if limit != 0: limit = None # done by function
789
        start = None # done by function
790
        order_by = None # done by function
791
    elif order_by is order_by_pkey:
792 5641 aaronmk
        if lists.is_seq(distinct_on) and distinct_on: order_by = distinct_on[0]
793
        elif table0 != None: order_by = table_order_by(db, table0, recover=True)
794
        else: order_by = None
795 865 aaronmk
796 2315 aaronmk
    query = 'SELECT'
797 2056 aaronmk
798 2315 aaronmk
    def parse_col(col): return sql_gen.as_Col(col, default_table).to_str(db)
799 2056 aaronmk
800 2200 aaronmk
    # DISTINCT ON columns
801 2233 aaronmk
    if distinct_on != []:
802 2467 aaronmk
        query += '\nDISTINCT'
803 2254 aaronmk
        if distinct_on is not distinct_on_all:
804 2200 aaronmk
            query += ' ON ('+(', '.join(map(parse_col, distinct_on)))+')'
805
806
    # Columns
807 3185 aaronmk
    if query.find('\n') >= 0: whitespace = '\n'
808
    else: whitespace = ' '
809
    if fields == None: query += whitespace+'*'
810 2765 aaronmk
    else:
811
        assert fields != []
812 3185 aaronmk
        if len(fields) > 1: whitespace = '\n'
813
        query += whitespace+('\n, '.join(map(parse_col, fields)))
814 2200 aaronmk
815
    # Main table
816 3185 aaronmk
    if query.find('\n') >= 0 or len(tables) > 0: whitespace = '\n'
817
    else: whitespace = ' '
818 3420 aaronmk
    if table0 != None: query += whitespace+'FROM '+table0.to_str(db)
819 865 aaronmk
820 2122 aaronmk
    # Add joins
821 2271 aaronmk
    left_table = table0
822 2263 aaronmk
    for join_ in tables:
823
        table = join_.table
824 2238 aaronmk
825 2343 aaronmk
        # Parse special values
826
        if join_.type_ is sql_gen.filter_out: # filter no match
827 2376 aaronmk
            conds.append((sql_gen.Col(table_not_null_col(db, table), table),
828 2853 aaronmk
                sql_gen.CompareCond(None, '~=')))
829 2343 aaronmk
830 2467 aaronmk
        query += '\n'+join_.to_str(db, left_table)
831 2122 aaronmk
832
        left_table = table
833
834 865 aaronmk
    missing = True
835 2376 aaronmk
    if conds != []:
836 2576 aaronmk
        if len(conds) == 1: whitespace = ' '
837
        else: whitespace = '\n'
838 2578 aaronmk
        query += '\n'+sql_gen.combine_conds([sql_gen.ColValueCond(l, r)
839
            .to_str(db) for l, r in conds], 'WHERE')
840 2227 aaronmk
    if order_by != None:
841 5642 aaronmk
        query += '\nORDER BY '+sql_gen.as_Col(order_by).to_str(db)
842 3297 aaronmk
    if limit != None: query += '\nLIMIT '+str(limit)
843 865 aaronmk
    if start != None:
844 2467 aaronmk
        if start != 0: query += '\nOFFSET '+str(start)
845 865 aaronmk
846 3494 aaronmk
    if explain: query = with_explain_comment(db, query)
847 3266 aaronmk
848 2786 aaronmk
    return query
849 11 aaronmk
850 2054 aaronmk
def select(db, *args, **kw_args):
851
    '''For params, see mk_select() and run_query()'''
852
    recover = kw_args.pop('recover', None)
853
    cacheable = kw_args.pop('cacheable', True)
854 2442 aaronmk
    log_level = kw_args.pop('log_level', 2)
855 2054 aaronmk
856 2791 aaronmk
    return run_query(db, mk_select(db, *args, **kw_args), recover, cacheable,
857
        log_level=log_level)
858 2054 aaronmk
859 2788 aaronmk
def mk_insert_select(db, table, cols=None, select_query=None, returning=None,
860 3181 aaronmk
    embeddable=False, ignore=False, src=None):
861 1960 aaronmk
    '''
862
    @param returning str|None An inserted column (such as pkey) to return
863 2070 aaronmk
    @param embeddable Whether the query should be embeddable as a nested SELECT.
864 2073 aaronmk
        Warning: If you set this and cacheable=True when the query is run, the
865
        query will be fully cached, not just if it raises an exception.
866 3009 aaronmk
    @param ignore Whether to ignore duplicate keys.
867 3181 aaronmk
    @param src Will be included in the name of any created function, to help
868
        identify the data source in pg_stat_activity.
869 1960 aaronmk
    '''
870 2754 aaronmk
    table = sql_gen.remove_table_rename(sql_gen.as_Table(table))
871 2318 aaronmk
    if cols == []: cols = None # no cols (all defaults) = unknown col names
872 3010 aaronmk
    if cols != None: cols = [sql_gen.to_name_only_col(c, table) for c in cols]
873 2063 aaronmk
    if select_query == None: select_query = 'DEFAULT VALUES'
874 2327 aaronmk
    if returning != None: returning = sql_gen.as_Col(returning, table)
875 2063 aaronmk
876 2497 aaronmk
    first_line = 'INSERT INTO '+table.to_str(db)
877 2063 aaronmk
878 3009 aaronmk
    def mk_insert(select_query):
879
        query = first_line
880 3014 aaronmk
        if cols != None:
881
            query += '\n('+(', '.join((c.to_str(db) for c in cols)))+')'
882 3009 aaronmk
        query += '\n'+select_query
883
884
        if returning != None:
885
            returning_name_col = sql_gen.to_name_only_col(returning)
886
            query += '\nRETURNING '+returning_name_col.to_str(db)
887
888
        return query
889 2063 aaronmk
890 3489 aaronmk
    return_type = sql_gen.CustomCode('unknown')
891
    if returning != None: return_type = sql_gen.ColType(returning)
892 3017 aaronmk
893 3009 aaronmk
    if ignore:
894 3017 aaronmk
        # Always return something to set the correct rowcount
895
        if returning == None: returning = sql_gen.NamedCol('NULL', None)
896
897 3009 aaronmk
        embeddable = True # must use function
898 3010 aaronmk
899 3450 aaronmk
        if cols == None: row = [sql_gen.Col(sql_gen.all_cols, 'row')]
900
        else: row = [sql_gen.Col(c.name, 'row') for c in cols]
901 3092 aaronmk
902 3484 aaronmk
        query = sql_gen.RowExcIgnore(sql_gen.RowType(table), select_query,
903 3497 aaronmk
            sql_gen.ReturnQuery(mk_insert(sql_gen.Values(row).to_str(db))),
904
            cols)
905 3009 aaronmk
    else: query = mk_insert(select_query)
906
907 2070 aaronmk
    if embeddable:
908
        # Create function
909 2513 aaronmk
        function_name = sql_gen.clean_name(first_line)
910 3181 aaronmk
        if src != None: function_name = src+': '+function_name
911 2189 aaronmk
        while True:
912
            try:
913 3451 aaronmk
                func = db.TempFunction(function_name)
914 3489 aaronmk
                def_ = sql_gen.FunctionDef(func, sql_gen.SetOf(return_type),
915
                    query)
916 2194 aaronmk
917 3443 aaronmk
                run_query(db, def_.to_str(db), recover=True, cacheable=True,
918 2945 aaronmk
                    log_ignore_excs=(DuplicateException,))
919 2189 aaronmk
                break # this version was successful
920 2945 aaronmk
            except DuplicateException, e:
921 2189 aaronmk
                function_name = next_version(function_name)
922
                # try again with next version of name
923 2070 aaronmk
924 2337 aaronmk
        # Return query that uses function
925 3009 aaronmk
        cols = None
926
        if returning != None: cols = [returning]
927 3451 aaronmk
        func_table = sql_gen.NamedTable('f', sql_gen.FunctionCall(func), cols)
928
            # AS clause requires function alias
929 3298 aaronmk
        return mk_select(db, func_table, order_by=None)
930 2070 aaronmk
931 2787 aaronmk
    return query
932 2066 aaronmk
933 3074 aaronmk
def insert_select(db, table, *args, **kw_args):
934 2085 aaronmk
    '''For params, see mk_insert_select() and run_query_into()
935 2386 aaronmk
    @param into sql_gen.Table with suggested name of temp table to put RETURNING
936
        values in
937 2072 aaronmk
    '''
938 3141 aaronmk
    returning = kw_args.get('returning', None)
939
    ignore = kw_args.get('ignore', False)
940
941 2386 aaronmk
    into = kw_args.pop('into', None)
942
    if into != None: kw_args['embeddable'] = True
943 2066 aaronmk
    recover = kw_args.pop('recover', None)
944 3141 aaronmk
    if ignore: recover = True
945 2066 aaronmk
    cacheable = kw_args.pop('cacheable', True)
946 2673 aaronmk
    log_level = kw_args.pop('log_level', 2)
947 2066 aaronmk
948 3141 aaronmk
    rowcount_only = ignore and returning == None # keep NULL rows on server
949
    if rowcount_only: into = sql_gen.Table('rowcount')
950
951 3074 aaronmk
    cur = run_query_into(db, mk_insert_select(db, table, *args, **kw_args),
952
        into, recover=recover, cacheable=cacheable, log_level=log_level)
953 3141 aaronmk
    if rowcount_only: empty_temp(db, into)
954 3074 aaronmk
    autoanalyze(db, table)
955
    return cur
956 2063 aaronmk
957 2738 aaronmk
default = sql_gen.default # tells insert() to use the default value for a column
958 2066 aaronmk
959 2063 aaronmk
def insert(db, table, row, *args, **kw_args):
960 2085 aaronmk
    '''For params, see insert_select()'''
961 5050 aaronmk
    ignore = kw_args.pop('ignore', False)
962 5094 aaronmk
    if ignore: kw_args.setdefault('recover', True)
963 5050 aaronmk
964 1960 aaronmk
    if lists.is_seq(row): cols = None
965
    else:
966
        cols = row.keys()
967
        row = row.values()
968 2738 aaronmk
    row = list(row) # ensure that "== []" works
969 1960 aaronmk
970 2738 aaronmk
    if row == []: query = None
971
    else: query = sql_gen.Values(row).to_str(db)
972 1961 aaronmk
973 5050 aaronmk
    try: return insert_select(db, table, cols, query, *args, **kw_args)
974 5057 aaronmk
    except (DuplicateKeyException, NullValueException):
975 5050 aaronmk
        if not ignore: raise
976 5163 aaronmk
        return None
977 11 aaronmk
978 3152 aaronmk
def mk_update(db, table, changes=None, cond=None, in_place=False,
979 3153 aaronmk
    cacheable_=True):
980 2402 aaronmk
    '''
981
    @param changes [(col, new_value),...]
982
        * container can be any iterable type
983
        * col: sql_gen.Code|str (for col name)
984
        * new_value: sql_gen.Code|literal value
985
    @param cond sql_gen.Code WHERE condition. e.g. use sql_gen.*Cond objects.
986 3056 aaronmk
    @param in_place If set, locks the table and updates rows in place.
987
        This avoids creating dead rows in PostgreSQL.
988
        * cond must be None
989 3153 aaronmk
    @param cacheable_ Whether column structure information used to generate the
990 3152 aaronmk
        query can be cached
991 2402 aaronmk
    @return str query
992
    '''
993 3057 aaronmk
    table = sql_gen.as_Table(table)
994
    changes = [(sql_gen.to_name_only_col(c, table), sql_gen.as_Value(v))
995
        for c, v in changes]
996
997 3056 aaronmk
    if in_place:
998
        assert cond == None
999 3058 aaronmk
1000 5398 aaronmk
        def col_type(col):
1001
            return sql_gen.canon_type(db.col_info(
1002
                sql_gen.with_default_table(c, table), cacheable_).type)
1003
        changes = [(c, v, col_type(c)) for c, v in changes]
1004 3065 aaronmk
        query = 'ALTER TABLE '+table.to_str(db)+'\n'
1005 5396 aaronmk
        query += ',\n'.join(('ALTER COLUMN '+c.to_str(db)+' TYPE '+t+'\nUSING '
1006
            +v.to_str(db) for c, v, t in changes))
1007 3058 aaronmk
    else:
1008
        query = 'UPDATE '+table.to_str(db)+'\nSET\n'
1009
        query += ',\n'.join((c.to_str(db)+' = '+v.to_str(db)
1010
            for c, v in changes))
1011
        if cond != None: query += '\nWHERE\n'+cond.to_str(db)
1012 3056 aaronmk
1013 3266 aaronmk
    query = with_explain_comment(db, query)
1014
1015 2402 aaronmk
    return query
1016
1017 3074 aaronmk
def update(db, table, *args, **kw_args):
1018 2402 aaronmk
    '''For params, see mk_update() and run_query()'''
1019
    recover = kw_args.pop('recover', None)
1020 3043 aaronmk
    cacheable = kw_args.pop('cacheable', False)
1021 3030 aaronmk
    log_level = kw_args.pop('log_level', 2)
1022 2402 aaronmk
1023 3074 aaronmk
    cur = run_query(db, mk_update(db, table, *args, **kw_args), recover,
1024
        cacheable, log_level=log_level)
1025
    autoanalyze(db, table)
1026
    return cur
1027 2402 aaronmk
1028 3286 aaronmk
def mk_delete(db, table, cond=None):
1029
    '''
1030
    @param cond sql_gen.Code WHERE condition. e.g. use sql_gen.*Cond objects.
1031
    @return str query
1032
    '''
1033
    query = 'DELETE FROM '+table.to_str(db)
1034
    if cond != None: query += '\nWHERE '+cond.to_str(db)
1035
1036
    query = with_explain_comment(db, query)
1037
1038
    return query
1039
1040
def delete(db, table, *args, **kw_args):
1041
    '''For params, see mk_delete() and run_query()'''
1042
    recover = kw_args.pop('recover', None)
1043 3295 aaronmk
    cacheable = kw_args.pop('cacheable', True)
1044 3286 aaronmk
    log_level = kw_args.pop('log_level', 2)
1045
1046
    cur = run_query(db, mk_delete(db, table, *args, **kw_args), recover,
1047
        cacheable, log_level=log_level)
1048
    autoanalyze(db, table)
1049
    return cur
1050
1051 135 aaronmk
def last_insert_id(db):
1052 1849 aaronmk
    module = util.root_module(db.db)
1053 135 aaronmk
    if module == 'psycopg2': return value(run_query(db, 'SELECT lastval()'))
1054
    elif module == 'MySQLdb': return db.insert_id()
1055
    else: return None
1056 13 aaronmk
1057 3490 aaronmk
def define_func(db, def_):
1058
    func = def_.function
1059
    while True:
1060
        try:
1061
            run_query(db, def_.to_str(db), recover=True, cacheable=True,
1062
                log_ignore_excs=(DuplicateException,))
1063
            break # successful
1064
        except DuplicateException:
1065 3495 aaronmk
            func.name = next_version(func.name)
1066 3490 aaronmk
            # try again with next version of name
1067
1068 2394 aaronmk
def mk_flatten_mapping(db, into, cols, preserve=[], as_items=False):
1069 2383 aaronmk
    '''Creates a mapping from original column names (which may have collisions)
1070 2415 aaronmk
    to names that will be distinct among the columns' tables.
1071 2383 aaronmk
    This is meant to be used for several tables that are being joined together.
1072 2415 aaronmk
    @param cols The columns to combine. Duplicates will be removed.
1073
    @param into The table for the new columns.
1074 2394 aaronmk
    @param preserve [sql_gen.Col...] Columns not to rename. Note that these
1075
        columns will be included in the mapping even if they are not in cols.
1076
        The tables of the provided Col objects will be changed to into, so make
1077
        copies of them if you want to keep the original tables.
1078
    @param as_items Whether to return a list of dict items instead of a dict
1079 2383 aaronmk
    @return dict(orig_col=new_col, ...)
1080
        * orig_col: sql_gen.Col(orig_col_name, orig_table)
1081 2392 aaronmk
        * new_col: sql_gen.Col(orig_col_name, into)
1082
        * All mappings use the into table so its name can easily be
1083 2383 aaronmk
          changed for all columns at once
1084
    '''
1085 2415 aaronmk
    cols = lists.uniqify(cols)
1086
1087 2394 aaronmk
    items = []
1088 2389 aaronmk
    for col in preserve:
1089 2390 aaronmk
        orig_col = copy.copy(col)
1090 2392 aaronmk
        col.table = into
1091 2394 aaronmk
        items.append((orig_col, col))
1092
    preserve = set(preserve)
1093
    for col in cols:
1094 2716 aaronmk
        if col not in preserve:
1095 3750 aaronmk
            items.append((col, sql_gen.Col(strings.ustr(col), into, col.srcs)))
1096 2394 aaronmk
1097
    if not as_items: items = dict(items)
1098
    return items
1099 2383 aaronmk
1100 2393 aaronmk
def flatten(db, into, joins, cols, limit=None, start=None, **kw_args):
1101 2391 aaronmk
    '''For params, see mk_flatten_mapping()
1102
    @return See return value of mk_flatten_mapping()
1103
    '''
1104 2394 aaronmk
    items = mk_flatten_mapping(db, into, cols, as_items=True, **kw_args)
1105
    cols = [sql_gen.NamedCol(new.name, old) for old, new in items]
1106 5523 aaronmk
    run_query_into(db, mk_select(db, joins, cols, limit=limit, start=start),
1107
        into=into, add_pkey_=True)
1108 3708 aaronmk
        # don't cache because the temp table will usually be truncated after use
1109 2394 aaronmk
    return dict(items)
1110 2391 aaronmk
1111 3079 aaronmk
##### Database structure introspection
1112 2414 aaronmk
1113 3079 aaronmk
#### Tables
1114
1115 4555 aaronmk
def tables(db, schema_like='public', table_like='%', exact=False,
1116
    cacheable=True):
1117 3079 aaronmk
    if exact: compare = '='
1118
    else: compare = 'LIKE'
1119
1120
    module = util.root_module(db.db)
1121
    if module == 'psycopg2':
1122
        conds = [('schemaname', sql_gen.CompareCond(schema_like, compare)),
1123
            ('tablename', sql_gen.CompareCond(table_like, compare))]
1124
        return values(select(db, 'pg_tables', ['tablename'], conds,
1125 4555 aaronmk
            order_by='tablename', cacheable=cacheable, log_level=4))
1126 3079 aaronmk
    elif module == 'MySQLdb':
1127
        return values(run_query(db, 'SHOW TABLES LIKE '+db.esc_value(table_like)
1128
            , cacheable=True, log_level=4))
1129
    else: raise NotImplementedError("Can't list tables for "+module+' database')
1130
1131 4556 aaronmk
def table_exists(db, table, cacheable=True):
1132 3079 aaronmk
    table = sql_gen.as_Table(table)
1133 4556 aaronmk
    return list(tables(db, table.schema, table.name, True, cacheable)) != []
1134 3079 aaronmk
1135 2426 aaronmk
def table_row_count(db, table, recover=None):
1136 2786 aaronmk
    return value(run_query(db, mk_select(db, table, [sql_gen.row_count],
1137 3298 aaronmk
        order_by=None), recover=recover, log_level=3))
1138 2426 aaronmk
1139 5337 aaronmk
def table_col_names(db, table, recover=None):
1140 5528 aaronmk
    return list(col_names(select(db, table, limit=0, recover=recover,
1141
        log_level=4)))
1142 2414 aaronmk
1143 5383 aaronmk
def table_cols(db, table, *args, **kw_args):
1144
    return [sql_gen.as_Col(strings.ustr(c), table)
1145
        for c in table_col_names(db, table, *args, **kw_args)]
1146
1147 5521 aaronmk
def table_pkey_index(db, table, recover=None):
1148
    table_str = sql_gen.Literal(table.to_str(db))
1149
    try:
1150
        return sql_gen.Table(value(run_query(db, '''\
1151
SELECT relname
1152
FROM pg_index
1153
JOIN pg_class index ON index.oid = indexrelid
1154
WHERE
1155
indrelid = '''+table_str.to_str(db)+'''::regclass
1156
AND indisprimary
1157
'''
1158
            , recover, cacheable=True, log_level=4)), table.schema)
1159
    except StopIteration: raise DoesNotExistException('primary key', '')
1160
1161 5389 aaronmk
def table_pkey_col(db, table, recover=None):
1162 5061 aaronmk
    table = sql_gen.as_Table(table)
1163
1164 5989 aaronmk
    module = util.root_module(db.db)
1165
    if module == 'psycopg2':
1166
        return sql_gen.Col(index_cols(db, table_pkey_index(db, table,
1167
            recover))[0], table)
1168
    else:
1169
        join_cols = ['table_schema', 'table_name', 'constraint_schema',
1170
            'constraint_name']
1171
        tables = [sql_gen.Table('key_column_usage', 'information_schema'),
1172
            sql_gen.Join(
1173
                sql_gen.Table('table_constraints', 'information_schema'),
1174
                dict(((c, sql_gen.join_same_not_null) for c in join_cols)))]
1175
        cols = [sql_gen.Col('column_name')]
1176
1177
        conds = [('constraint_type', 'PRIMARY KEY'), ('table_name', table.name)]
1178
        schema = table.schema
1179
        if schema != None: conds.append(('table_schema', schema))
1180
        order_by = 'position_in_unique_constraint'
1181
1182
        try: return sql_gen.Col(value(select(db, tables, cols, conds,
1183
            order_by=order_by, limit=1, log_level=4)), table)
1184
        except StopIteration: raise DoesNotExistException('primary key', '')
1185 5389 aaronmk
1186 5990 aaronmk
def table_has_pkey(db, table, recover=None):
1187
    try: table_pkey_col(db, table, recover)
1188
    except DoesNotExistException: return False
1189
    else: return True
1190
1191 5389 aaronmk
def pkey_name(db, table, recover=None):
1192
    '''If no pkey, returns the first column in the table.'''
1193 5392 aaronmk
    return pkey_col(db, table, recover).name
1194 832 aaronmk
1195 5390 aaronmk
def pkey_col(db, table, recover=None):
1196 5391 aaronmk
    '''If no pkey, returns the first column in the table.'''
1197 5392 aaronmk
    try: return table_pkey_col(db, table, recover)
1198 5393 aaronmk
    except DoesNotExistException: return table_cols(db, table, recover)[0]
1199 5128 aaronmk
1200 2559 aaronmk
not_null_col = 'not_null_col'
1201 2340 aaronmk
1202
def table_not_null_col(db, table, recover=None):
1203
    '''Name assumed to be the value of not_null_col. If not found, uses pkey.'''
1204 5337 aaronmk
    if not_null_col in table_col_names(db, table, recover): return not_null_col
1205 5388 aaronmk
    else: return pkey_name(db, table, recover)
1206 2340 aaronmk
1207 3348 aaronmk
def constraint_cond(db, constraint):
1208
    module = util.root_module(db.db)
1209
    if module == 'psycopg2':
1210
        table_str = sql_gen.Literal(constraint.table.to_str(db))
1211
        name_str = sql_gen.Literal(constraint.name)
1212
        return value(run_query(db, '''\
1213
SELECT consrc
1214
FROM pg_constraint
1215
WHERE
1216
conrelid = '''+table_str.to_str(db)+'''::regclass
1217
AND conname = '''+name_str.to_str(db)+'''
1218
'''
1219
            , cacheable=True, log_level=4))
1220 5443 aaronmk
    else: raise NotImplementedError("Can't get constraint condition for "
1221
        +module+' database')
1222 3348 aaronmk
1223 5520 aaronmk
def index_exprs(db, index):
1224 3322 aaronmk
    index = sql_gen.as_Table(index)
1225 1909 aaronmk
    module = util.root_module(db.db)
1226
    if module == 'psycopg2':
1227 3322 aaronmk
        qual_index = sql_gen.Literal(index.to_str(db))
1228 5520 aaronmk
        return list(values(run_query(db, '''\
1229 3322 aaronmk
SELECT pg_get_indexdef(indexrelid, generate_series(1, indnatts), true)
1230
FROM pg_index
1231
WHERE indexrelid = '''+qual_index.to_str(db)+'''::regclass
1232 2782 aaronmk
'''
1233
            , cacheable=True, log_level=4)))
1234 5520 aaronmk
    else: raise NotImplementedError()
1235 853 aaronmk
1236 5520 aaronmk
def index_cols(db, index):
1237
    '''Can also use this for UNIQUE constraints, because a UNIQUE index is
1238
    automatically created. When you don't know whether something is a UNIQUE
1239
    constraint or a UNIQUE index, use this function.'''
1240
    return map(sql_gen.parse_expr_col, index_exprs(db, index))
1241
1242 5445 aaronmk
def index_cond(db, index):
1243
    index = sql_gen.as_Table(index)
1244
    module = util.root_module(db.db)
1245
    if module == 'psycopg2':
1246
        qual_index = sql_gen.Literal(index.to_str(db))
1247
        return value(run_query(db, '''\
1248
SELECT pg_get_expr(indpred, indrelid, true)
1249
FROM pg_index
1250
WHERE indexrelid = '''+qual_index.to_str(db)+'''::regclass
1251
'''
1252
            , cacheable=True, log_level=4))
1253
    else: raise NotImplementedError()
1254
1255 5521 aaronmk
def index_order_by(db, index):
1256
    return sql_gen.CustomCode(', '.join(index_exprs(db, index)))
1257
1258
def table_cluster_on(db, table, recover=None):
1259
    '''
1260
    @return The table's cluster index, or its pkey if none is set
1261
    '''
1262
    table_str = sql_gen.Literal(table.to_str(db))
1263
    try:
1264
        return sql_gen.Table(value(run_query(db, '''\
1265
SELECT relname
1266
FROM pg_index
1267
JOIN pg_class index ON index.oid = indexrelid
1268
WHERE
1269
indrelid = '''+table_str.to_str(db)+'''::regclass
1270
AND indisclustered
1271
'''
1272
            , recover, cacheable=True, log_level=4)), table.schema)
1273
    except StopIteration: return table_pkey_index(db, table, recover)
1274
1275
def table_order_by(db, table, recover=None):
1276 8223 aaronmk
    '''
1277
    @return None if table is view, because table_cluster_on() would return None
1278
    '''
1279 5525 aaronmk
    if table.order_by == None:
1280
        try: table.order_by = index_order_by(db, table_cluster_on(db, table,
1281
            recover))
1282
        except DoesNotExistException: pass
1283
    return table.order_by
1284 5521 aaronmk
1285 7242 aaronmk
#### Views
1286
1287
def view_exists(db, view):
1288
    view_str = sql_gen.Literal(view.to_str(db))
1289
    try:
1290
        return value(run_query(db, '''\
1291
SELECT relkind = 'v'
1292
FROM pg_class
1293
WHERE oid = '''+view_str.to_str(db)+'''::regclass
1294
'''
1295
            , cacheable=True, log_level=4))
1296
    except DoesNotExistException: return False
1297
1298 10139 aaronmk
#### Columns
1299
1300
def col_default_value(db, col):
1301
    assert isinstance(col, sql_gen.Col)
1302
    return value(run_query(db,
1303
        'SELECT util.col_default_value('+sql_gen.col2col_ref(db, col)+')'))
1304
1305
def col_is_constant(db, col):
1306
    assert isinstance(col, sql_gen.Col)
1307
    return value(run_query(db,
1308
        'SELECT util.is_constant('+sql_gen.col2col_ref(db, col)+')'))
1309
1310 3079 aaronmk
#### Functions
1311
1312
def function_exists(db, function):
1313 3423 aaronmk
    qual_function = sql_gen.Literal(function.to_str(db))
1314
    try:
1315 3425 aaronmk
        select(db, fields=[sql_gen.Cast('regproc', qual_function)],
1316
            recover=True, cacheable=True, log_level=4)
1317 3423 aaronmk
    except DoesNotExistException: return False
1318 4146 aaronmk
    except DuplicateException: return True # overloaded function
1319 3423 aaronmk
    else: return True
1320 3079 aaronmk
1321 5713 aaronmk
def function_param0_type(db, function):
1322
    qual_function = sql_gen.Literal(function.to_str(db))
1323
    return value(run_query(db, '''\
1324
SELECT proargtypes[0]::regtype
1325
FROM pg_proc
1326
WHERE oid = '''+qual_function.to_str(db)+'''::regproc
1327
'''
1328
        , cacheable=True, log_level=4))
1329
1330 3079 aaronmk
##### Structural changes
1331
1332
#### Columns
1333
1334 5020 aaronmk
def add_col(db, table, col, comment=None, if_not_exists=False, **kw_args):
1335 3079 aaronmk
    '''
1336
    @param col TypedCol Name may be versioned, so be sure to propagate any
1337
        renaming back to any source column for the TypedCol.
1338
    @param comment None|str SQL comment used to distinguish columns of the same
1339
        name from each other when they contain different data, to allow the
1340
        ADD COLUMN query to be cached. If not set, query will not be cached.
1341
    '''
1342
    assert isinstance(col, sql_gen.TypedCol)
1343
1344
    while True:
1345
        str_ = 'ALTER TABLE '+table.to_str(db)+' ADD COLUMN '+col.to_str(db)
1346
        if comment != None: str_ += ' '+sql_gen.esc_comment(comment)
1347
1348
        try:
1349
            run_query(db, str_, recover=True, cacheable=True, **kw_args)
1350
            break
1351
        except DuplicateException:
1352 5020 aaronmk
            if if_not_exists: raise
1353 3079 aaronmk
            col.name = next_version(col.name)
1354
            # try again with next version of name
1355
1356
def add_not_null(db, col):
1357
    table = col.table
1358
    col = sql_gen.to_name_only_col(col)
1359
    run_query(db, 'ALTER TABLE '+table.to_str(db)+' ALTER COLUMN '
1360
        +col.to_str(db)+' SET NOT NULL', cacheable=True, log_level=3)
1361
1362 4443 aaronmk
def drop_not_null(db, col):
1363
    table = col.table
1364
    col = sql_gen.to_name_only_col(col)
1365
    run_query(db, 'ALTER TABLE '+table.to_str(db)+' ALTER COLUMN '
1366
        +col.to_str(db)+' DROP NOT NULL', cacheable=True, log_level=3)
1367
1368 2096 aaronmk
row_num_col = '_row_num'
1369
1370 4997 aaronmk
row_num_col_def = sql_gen.TypedCol('', 'serial', nullable=False,
1371 3079 aaronmk
    constraints='PRIMARY KEY')
1372
1373 4997 aaronmk
def add_row_num(db, table, name=row_num_col):
1374
    '''Adds a row number column to a table. Its definition is in
1375
    row_num_col_def. It will be the primary key.'''
1376
    col_def = copy.copy(row_num_col_def)
1377
    col_def.name = name
1378 5021 aaronmk
    add_col(db, table, col_def, comment='', if_not_exists=True, log_level=3)
1379 3079 aaronmk
1380
#### Indexes
1381
1382
def add_pkey(db, table, cols=None, recover=None):
1383
    '''Adds a primary key.
1384
    @param cols [sql_gen.Col,...] The columns in the primary key.
1385
        Defaults to the first column in the table.
1386
    @pre The table must not already have a primary key.
1387
    '''
1388
    table = sql_gen.as_Table(table)
1389 5388 aaronmk
    if cols == None: cols = [pkey_name(db, table, recover)]
1390 3079 aaronmk
    col_strs = [sql_gen.to_name_only_col(v).to_str(db) for v in cols]
1391
1392
    run_query(db, 'ALTER TABLE '+table.to_str(db)+' ADD PRIMARY KEY ('
1393
        +(', '.join(col_strs))+')', recover=True, cacheable=True, log_level=3,
1394
        log_ignore_excs=(DuplicateException,))
1395
1396 2998 aaronmk
def add_index(db, exprs, table=None, unique=False, ensure_not_null_=True):
1397 2688 aaronmk
    '''Adds an index on column(s) or expression(s) if it doesn't already exist.
1398 3356 aaronmk
    Currently, only function calls and literal values are supported expressions.
1399 2998 aaronmk
    @param ensure_not_null_ If set, translates NULL values to sentinel values.
1400 2847 aaronmk
        This allows indexes to be used for comparisons where NULLs are equal.
1401 2538 aaronmk
    '''
1402 2964 aaronmk
    exprs = lists.mk_seq(exprs)
1403 2538 aaronmk
1404 2688 aaronmk
    # Parse exprs
1405
    old_exprs = exprs[:]
1406
    exprs = []
1407
    cols = []
1408
    for i, expr in enumerate(old_exprs):
1409 2823 aaronmk
        expr = sql_gen.as_Col(expr, table)
1410 2688 aaronmk
1411 2823 aaronmk
        # Handle nullable columns
1412 2998 aaronmk
        if ensure_not_null_:
1413 3164 aaronmk
            try: expr = sql_gen.ensure_not_null(db, expr)
1414 2860 aaronmk
            except KeyError: pass # unknown type, so just create plain index
1415 2823 aaronmk
1416 2688 aaronmk
        # Extract col
1417 3002 aaronmk
        expr = copy.deepcopy(expr) # don't modify input!
1418 3356 aaronmk
        col = expr
1419
        if isinstance(expr, sql_gen.FunctionCall): col = expr.args[0]
1420
        expr = sql_gen.cast_literal(expr)
1421
        if not isinstance(expr, (sql_gen.Expr, sql_gen.Col)):
1422 2688 aaronmk
            expr = sql_gen.Expr(expr)
1423 3356 aaronmk
1424 2688 aaronmk
1425
        # Extract table
1426
        if table == None:
1427
            assert sql_gen.is_table_col(col)
1428
            table = col.table
1429
1430 3356 aaronmk
        if isinstance(col, sql_gen.Col): col.table = None
1431 2688 aaronmk
1432
        exprs.append(expr)
1433
        cols.append(col)
1434 2408 aaronmk
1435 2688 aaronmk
    table = sql_gen.as_Table(table)
1436
1437 3005 aaronmk
    # Add index
1438 3148 aaronmk
    str_ = 'CREATE'
1439
    if unique: str_ += ' UNIQUE'
1440
    str_ += ' INDEX ON '+table.to_str(db)+' ('+(
1441
        ', '.join((v.to_str(db) for v in exprs)))+')'
1442
    run_query(db, str_, recover=True, cacheable=True, log_level=3)
1443 2408 aaronmk
1444 5987 aaronmk
def add_pkey_index(db, table): add_index(db, pkey_col(db, table), table)
1445
1446 5765 aaronmk
def add_pkey_or_index(db, table, cols=None, recover=None, warn=False):
1447
    try: add_pkey(db, table, cols, recover)
1448
    except DuplicateKeyException, e:
1449
        if warn: warnings.warn(UserWarning(exc.str_(e)))
1450 5988 aaronmk
        add_pkey_index(db, table)
1451 5765 aaronmk
1452 3083 aaronmk
already_indexed = object() # tells add_indexes() the pkey has already been added
1453
1454
def add_indexes(db, table, has_pkey=True):
1455
    '''Adds an index on all columns in a table.
1456
    @param has_pkey bool|already_indexed Whether a pkey instead of a regular
1457
        index should be added on the first column.
1458
        * If already_indexed, the pkey is assumed to have already been added
1459
    '''
1460 5337 aaronmk
    cols = table_col_names(db, table)
1461 3083 aaronmk
    if has_pkey:
1462
        if has_pkey is not already_indexed: add_pkey(db, table)
1463
        cols = cols[1:]
1464
    for col in cols: add_index(db, col, table)
1465
1466 3079 aaronmk
#### Tables
1467 2772 aaronmk
1468 3079 aaronmk
### Maintenance
1469 2772 aaronmk
1470 3079 aaronmk
def analyze(db, table):
1471
    table = sql_gen.as_Table(table)
1472
    run_query(db, 'ANALYZE '+table.to_str(db), log_level=3)
1473 2934 aaronmk
1474 3079 aaronmk
def autoanalyze(db, table):
1475
    if db.autoanalyze: analyze(db, table)
1476 2935 aaronmk
1477 3079 aaronmk
def vacuum(db, table):
1478
    table = sql_gen.as_Table(table)
1479
    db.with_autocommit(lambda: run_query(db, 'VACUUM ANALYZE '+table.to_str(db),
1480
        log_level=3))
1481 2086 aaronmk
1482 3079 aaronmk
### Lifecycle
1483
1484 3247 aaronmk
def drop(db, type_, name):
1485
    name = sql_gen.as_Name(name)
1486
    run_query(db, 'DROP '+type_+' IF EXISTS '+name.to_str(db)+' CASCADE')
1487 2889 aaronmk
1488 3247 aaronmk
def drop_table(db, table): drop(db, 'TABLE', table)
1489
1490 3082 aaronmk
def create_table(db, table, cols=[], has_pkey=True, col_indexes=True,
1491
    like=None):
1492 2675 aaronmk
    '''Creates a table.
1493 2681 aaronmk
    @param cols [sql_gen.TypedCol,...] The column names and types
1494
    @param has_pkey If set, the first column becomes the primary key.
1495 2760 aaronmk
    @param col_indexes bool|[ref]
1496
        * If True, indexes will be added on all non-pkey columns.
1497
        * If a list reference, [0] will be set to a function to do this.
1498
          This can be used to delay index creation until the table is populated.
1499 2675 aaronmk
    '''
1500
    table = sql_gen.as_Table(table)
1501
1502 3082 aaronmk
    if like != None:
1503
        cols = [sql_gen.CustomCode('LIKE '+like.to_str(db)+' INCLUDING ALL')
1504
            ]+cols
1505 5525 aaronmk
        table.order_by = like.order_by
1506 2681 aaronmk
    if has_pkey:
1507
        cols[0] = pkey = copy.copy(cols[0]) # don't modify input!
1508 2872 aaronmk
        pkey.constraints = 'PRIMARY KEY'
1509 2681 aaronmk
1510 3085 aaronmk
    temp = table.is_temp and not db.debug_temp
1511
        # temp tables permanent in debug_temp mode
1512 2760 aaronmk
1513 3085 aaronmk
    # Create table
1514 3383 aaronmk
    def create():
1515 3085 aaronmk
        str_ = 'CREATE'
1516
        if temp: str_ += ' TEMP'
1517
        str_ += ' TABLE '+table.to_str(db)+' (\n'
1518
        str_ += '\n, '.join(c.to_str(db) for c in cols)
1519 3126 aaronmk
        str_ += '\n);'
1520 3085 aaronmk
1521 7113 aaronmk
        opts = dict(recover=True, cacheable=True, log_level=2,
1522 3383 aaronmk
            log_ignore_excs=(DuplicateException,))
1523 7113 aaronmk
        try: run_query(db, str_, **opts)
1524
        except InvalidTypeException: # try again as view
1525
            run_query_into(db, mk_select(db, like, limit=0), into=table, **opts)
1526 3383 aaronmk
    if table.is_temp:
1527
        while True:
1528
            try:
1529
                create()
1530
                break
1531
            except DuplicateException:
1532
                table.name = next_version(table.name)
1533
                # try again with next version of name
1534
    else: create()
1535 3085 aaronmk
1536 2760 aaronmk
    # Add indexes
1537 2773 aaronmk
    if has_pkey: has_pkey = already_indexed
1538
    def add_indexes_(): add_indexes(db, table, has_pkey)
1539
    if isinstance(col_indexes, list): col_indexes[0] = add_indexes_ # defer
1540
    elif col_indexes: add_indexes_() # add now
1541 2675 aaronmk
1542 3084 aaronmk
def copy_table_struct(db, src, dest):
1543
    '''Creates a structure-only copy of a table. (Does not copy data.)'''
1544 3085 aaronmk
    create_table(db, dest, has_pkey=False, col_indexes=False, like=src)
1545 3084 aaronmk
1546 5529 aaronmk
def copy_table(db, src, dest):
1547
    '''Creates a copy of a table, including data'''
1548
    copy_table_struct(db, src, dest)
1549
    insert_select(db, dest, None, mk_select(db, src))
1550
1551 3079 aaronmk
### Data
1552 2684 aaronmk
1553 2970 aaronmk
def truncate(db, table, schema='public', **kw_args):
1554
    '''For params, see run_query()'''
1555 2777 aaronmk
    table = sql_gen.as_Table(table, schema)
1556 2970 aaronmk
    return run_query(db, 'TRUNCATE '+table.to_str(db)+' CASCADE', **kw_args)
1557 2732 aaronmk
1558 2965 aaronmk
def empty_temp(db, tables):
1559
    tables = lists.mk_seq(tables)
1560 2971 aaronmk
    for table in tables: truncate(db, table, log_level=3)
1561 2965 aaronmk
1562 1968 aaronmk
def empty_db(db, schema='public', **kw_args):
1563
    '''For kw_args, see tables()'''
1564
    for table in tables(db, schema, **kw_args): truncate(db, table, schema)
1565 3094 aaronmk
1566 7179 aaronmk
def distinct_table(db, table, distinct_on, joins=None):
1567 3094 aaronmk
    '''Creates a copy of a temp table which is distinct on the given columns.
1568 5902 aaronmk
    Adds an index on table's distinct_on columns, to facilitate merge joins.
1569 3097 aaronmk
    @param distinct_on If empty, creates a table with one row. This is useful if
1570
        your distinct_on columns are all literal values.
1571 7179 aaronmk
    @param joins The joins to use when creating the new table
1572 3099 aaronmk
    @return The new table.
1573 3094 aaronmk
    '''
1574 7179 aaronmk
    if joins == None: joins = [table]
1575
1576 3099 aaronmk
    new_table = sql_gen.suffixed_table(table, '_distinct')
1577 3411 aaronmk
    distinct_on = filter(sql_gen.is_table_col, distinct_on)
1578 3094 aaronmk
1579 3099 aaronmk
    copy_table_struct(db, table, new_table)
1580 3097 aaronmk
1581
    limit = None
1582
    if distinct_on == []: limit = 1 # one sample row
1583 5903 aaronmk
    else: add_index(db, distinct_on, table) # for join optimization
1584 3097 aaronmk
1585 7179 aaronmk
    insert_select(db, new_table, None, mk_select(db, joins,
1586
        [sql_gen.Col(sql_gen.all_cols, table)], distinct_on=distinct_on,
1587
        order_by=None, limit=limit))
1588 3099 aaronmk
    analyze(db, new_table)
1589 3094 aaronmk
1590 3099 aaronmk
    return new_table