Project

General

Profile

1
# Database access
2

    
3
import copy
4
import re
5
import time
6
import warnings
7

    
8
import exc
9
import dicts
10
import iters
11
import lists
12
import profiling
13
from Proxy import Proxy
14
import rand
15
import sql_gen
16
import strings
17
import util
18

    
19
##### Exceptions
20

    
21
def get_cur_query(cur, input_query=None):
22
    raw_query = None
23
    if hasattr(cur, 'query'): raw_query = cur.query
24
    elif hasattr(cur, '_last_executed'): raw_query = cur._last_executed
25
    
26
    if raw_query != None: return raw_query
27
    else: return '[input] '+strings.ustr(input_query)
28

    
29
def _add_cursor_info(e, *args, **kw_args):
30
    '''For params, see get_cur_query()'''
31
    exc.add_msg(e, 'query: '+strings.ustr(get_cur_query(*args, **kw_args)))
32

    
33
class DbException(exc.ExceptionWithCause):
34
    def __init__(self, msg, cause=None, cur=None):
35
        exc.ExceptionWithCause.__init__(self, msg, cause, cause_newline=True)
36
        if cur != None: _add_cursor_info(self, cur)
37

    
38
class ExceptionWithName(DbException):
39
    def __init__(self, name, cause=None):
40
        DbException.__init__(self, 'for name: '
41
            +strings.as_tt(strings.ustr(name)), cause)
42
        self.name = name
43

    
44
class ExceptionWithValue(DbException):
45
    def __init__(self, value, cause=None):
46
        DbException.__init__(self, 'for value: '
47
            +strings.as_tt(strings.urepr(value)), cause)
48
        self.value = value
49

    
50
class ExceptionWithNameType(DbException):
51
    def __init__(self, type_, name, cause=None):
52
        DbException.__init__(self, 'for type: '+strings.as_tt(strings.ustr(
53
            type_))+'; name: '+strings.as_tt(name), cause)
54
        self.type = type_
55
        self.name = name
56

    
57
class ConstraintException(DbException):
58
    def __init__(self, name, cond, cols, cause=None):
59
        msg = 'Violated '+strings.as_tt(name)+' constraint'
60
        if cond != None: msg += ' with condition '+strings.as_tt(cond)
61
        if cols != []: msg += ' on columns: '+strings.as_tt(', '.join(cols))
62
        DbException.__init__(self, msg, cause)
63
        self.name = name
64
        self.cond = cond
65
        self.cols = cols
66

    
67
class MissingCastException(DbException):
68
    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
        self.type = type_
73
        self.col = col
74

    
75
class EncodingException(ExceptionWithName): pass
76

    
77
class DuplicateKeyException(ConstraintException): pass
78

    
79
class NullValueException(ConstraintException): pass
80

    
81
class CheckException(ConstraintException): pass
82

    
83
class InvalidValueException(ExceptionWithValue): pass
84

    
85
class InvalidTypeException(ExceptionWithNameType): pass
86

    
87
class DuplicateException(ExceptionWithNameType): pass
88

    
89
class DoesNotExistException(ExceptionWithNameType): pass
90

    
91
class EmptyRowException(DbException): pass
92

    
93
##### Warnings
94

    
95
class DbWarning(UserWarning): pass
96

    
97
##### 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
##### Escaping
125

    
126
def esc_name_by_module(module, name):
127
    if module == 'psycopg2' or module == None: quote = '"'
128
    elif module == 'MySQLdb': quote = '`'
129
    else: raise NotImplementedError("Can't escape name for "+module+' database')
130
    return sql_gen.esc_name(name, quote)
131

    
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
##### Database connections
145

    
146
db_config_names = ['engine', 'host', 'user', 'password', 'database', 'schemas']
147

    
148
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
log_debug_none = lambda msg, level=2: None
165

    
166
class DbConn:
167
    def __init__(self, db_config, autocommit=True, caching=True,
168
        log_debug=log_debug_none, debug_temp=False, src=None):
169
        '''
170
        @param debug_temp Whether temporary objects should instead be permanent.
171
            This assists in debugging the internal objects used by the program.
172
        @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
        '''
175
        self.db_config = db_config
176
        self.autocommit = autocommit
177
        self.caching = caching
178
        self.log_debug = log_debug
179
        self.debug = log_debug != log_debug_none
180
        self.debug_temp = debug_temp
181
        self.src = src
182
        self.autoanalyze = False
183
        self.autoexplain = False
184
        self.profile_row_ct = None
185
        
186
        self._savepoint = 0
187
        self._reset()
188
    
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
        state['log_debug'] = None # don't pickle the debug callback
197
        state['_DbConn__db'] = None # don't pickle the connection
198
        return state
199
    
200
    def clear_cache(self): self.query_results = {}
201
    
202
    def _reset(self):
203
        self.clear_cache()
204
        assert self._savepoint == 0
205
        self._notices_seen = set()
206
        self.__db = None
207
    
208
    def connected(self): return self.__db != None
209
    
210
    def close(self):
211
        if not self.connected(): return
212
        
213
        # Record that the automatic transaction is now closed
214
        self._savepoint -= 1
215
        
216
        self.db.close()
217
        self._reset()
218
    
219
    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
    def _db(self):
225
        if self.__db == None:
226
            # Process db_config
227
            db_config = self.db_config.copy() # don't modify input!
228
            schemas = db_config.pop('schemas', None)
229
            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
            # Record that a transaction is already open
240
            self._savepoint += 1
241
            
242
            # Configure connection
243
            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
            if schemas != None:
248
                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
        
252
        return self.__db
253
    
254
    class DbCursor(Proxy):
255
        def __init__(self, outer):
256
            Proxy.__init__(self, outer.db.cursor())
257
            self.outer = outer
258
            self.query_results = outer.query_results
259
            self.query_lookup = None
260
            self.result = []
261
        
262
        def execute(self, query):
263
            self._is_insert = query.startswith('INSERT')
264
            self.query_lookup = query
265
            try:
266
                try: cur = self.inner.execute(query)
267
                finally: self.query = get_cur_query(self.inner, query)
268
            except Exception, e:
269
                self.result = e # cache the exception as the result
270
                self._cache_result()
271
                raise
272
            
273
            # Always cache certain queries
274
            query = sql_gen.lstrip(query)
275
            if query.startswith('CREATE') or query.startswith('ALTER'):
276
                # structural changes
277
                # 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
                    self._cache_result()
281
            elif self.rowcount == 0 and query.startswith('SELECT'): # empty
282
                consume_rows(self) # fetch all rows so result will be cached
283
            
284
            return cur
285
        
286
        def fetchone(self):
287
            row = self.inner.fetchone()
288
            if row != None: self.result.append(row)
289
            # otherwise, fetched all rows
290
            else: self._cache_result()
291
            return row
292
        
293
        def _cache_result(self):
294
            # 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
            if self.query_results != None and (not self._is_insert
299
                or isinstance(self.result, Exception)):
300
                
301
                assert self.query_lookup != None
302
                self.query_results[self.query_lookup] = self.CacheCursor(
303
                    util.dict_subset(dicts.AttrsDictView(self),
304
                    ['query', 'result', 'rowcount', 'description']))
305
        
306
        class CacheCursor:
307
            def __init__(self, cached_result): self.__dict__ = cached_result
308
            
309
            def execute(self, *args, **kw_args):
310
                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
    
318
    def esc_value(self, value):
319
        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
        return strings.to_unicode(str_)
327
    
328
    def esc_name(self, name): return esc_name(self, name) # calls global func
329
    
330
    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
    def can_mogrify(self):
338
        module = util.root_module(self.db)
339
        return module == 'psycopg2'
340
    
341
    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
    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
    def print_notices(self):
350
        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
    
356
    def run_query(self, query, cacheable=False, log_level=2,
357
        debug_msg_ref=None):
358
        '''
359
        @param log_ignore_excs The log_level will be increased by 2 if the query
360
            throws one of these exceptions.
361
        @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
        '''
365
        assert query != None
366
        
367
        if self.autocommit and self.src != None:
368
            query = sql_gen.esc_comment(self.src)+'\t'+query
369
        
370
        if not self.caching: cacheable = False
371
        used_cache = False
372
        
373
        if self.debug:
374
            profiler = profiling.ItersProfiler(start_now=True, iter_text='row')
375
        try:
376
            # Get cursor
377
            if cacheable:
378
                try: cur = self.query_results[query]
379
                except KeyError: cur = self.DbCursor(self)
380
                else: used_cache = True
381
            else: cur = self.db.cursor()
382
            
383
            # Run query
384
            try: cur.execute(query)
385
            except Exception, e:
386
                _add_cursor_info(e, self, query)
387
                raise
388
            else: self.do_autocommit()
389
        finally:
390
            if self.debug:
391
                profiler.stop(self.profile_row_ct)
392
                
393
                ## Log or return query
394
                
395
                query = strings.ustr(get_cur_query(cur, query))
396
                # Put the src comment on a separate line in the log file
397
                query = query.replace('\t', '\n', 1)
398
                
399
                msg = 'DB query: '
400
                
401
                if used_cache: msg += 'cache hit'
402
                elif cacheable: msg += 'cache miss'
403
                else: msg += 'non-cacheable'
404
                
405
                msg += ':\n'+profiler.msg()+'\n'+strings.as_code(query, 'SQL')
406
                
407
                if debug_msg_ref != None: debug_msg_ref[0] = msg
408
                else: self.log_debug(msg, log_level)
409
                
410
                self.print_notices()
411
        
412
        return cur
413
    
414
    def is_cached(self, query): return query in self.query_results
415
    
416
    def with_autocommit(self, func):
417
        import psycopg2.extensions
418
        
419
        prev_isolation_level = self.db.isolation_level
420
        self.db.set_isolation_level(
421
            psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
422
        try: return func()
423
        finally: self.db.set_isolation_level(prev_isolation_level)
424
    
425
    def with_savepoint(self, func):
426
        top = self._savepoint == 0
427
        savepoint = 'level_'+str(self._savepoint)
428
        
429
        if self.debug:
430
            self.log_debug('Begin transaction', level=4)
431
            profiler = profiling.ItersProfiler(start_now=True, iter_text='row')
432
        
433
        # Must happen before running queries so they don't get autocommitted
434
        self._savepoint += 1
435
        
436
        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
        except:
443
            if top: query = 'ROLLBACK'
444
            else: query = 'ROLLBACK TO SAVEPOINT '+savepoint
445
            self.run_query(query, log_level=4)
446
            
447
            raise
448
        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
            if not top:
453
                self.run_query('RELEASE SAVEPOINT '+savepoint, log_level=4)
454
            
455
            self._savepoint -= 1
456
            assert self._savepoint >= 0
457
            
458
            if self.debug:
459
                profiler.stop(self.profile_row_ct)
460
                self.log_debug('End transaction\n'+profiler.msg(), level=4)
461
            
462
            self.do_autocommit() # OK to do this after ROLLBACK TO SAVEPOINT
463
    
464
    def do_autocommit(self):
465
        '''Autocommits if outside savepoint'''
466
        assert self._savepoint >= 1
467
        if self.autocommit and self._savepoint == 1:
468
            self.log_debug('Autocommitting', level=4)
469
            self.db.commit()
470
    
471
    def col_info(self, col, cacheable=True):
472
        module = util.root_module(self.db)
473
        if module == 'psycopg2':
474
            qual_table = sql_gen.Literal(col.table.to_str(self))
475
            col_name_str = sql_gen.Literal(col.name)
476
            try:
477
                type_, is_array, default, nullable = row(run_query(self, '''\
478
SELECT
479
format_type(COALESCE(NULLIF(typelem, 0), pg_type.oid), -1) AS type
480
, typcategory = 'A' AS type_is_array
481
, pg_get_expr(pg_attrdef.adbin, attrelid, true) AS default
482
, 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
attrelid = '''+qual_table.to_str(self)+'''::regclass
488
AND attname = '''+col_name_str.to_str(self)+'''
489
'''
490
                    , recover=True, cacheable=cacheable, log_level=4))
491
            except (DoesNotExistException, StopIteration):
492
                raise sql_gen.NoUnderlyingTableException(col)
493
            if is_array: type_ = sql_gen.ArrayType(type_)
494
        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
        
514
        if default != None: default = sql_gen.as_Code(default, self)
515
        return sql_gen.TypedCol(col.name, type_, default, nullable)
516
    
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

    
522
connect = DbConn
523

    
524
##### Recoverable querying
525

    
526
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
    def make_DuplicateKeyException(constraint, e):
536
        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
        return DuplicateKeyException(constraint, cond, cols, e)
544
    
545
    match = re.match(r'^duplicate key value violates unique constraint "(.+?)"',
546
        msg)
547
    if match:
548
        constraint, = match.groups()
549
        raise make_DuplicateKeyException(constraint, e)
550
    
551
    match = re.match(r'^could not create unique index "(.+?)"\n'
552
        r'DETAIL:  Key .+? is duplicated', msg)
553
    if match:
554
        constraint, = match.groups()
555
        raise DuplicateKeyException(constraint, None, [], e)
556
    
557
    match = re.match(r'^null value in column "(.+?)" violates not-null'
558
        r' constraint', msg)
559
    if match:
560
        col, = match.groups()
561
        raise NullValueException('NOT NULL', None, [col], e)
562
    
563
    match = re.match(r'^new row for relation "(.+?)" violates check '
564
        r'constraint "(.+?)"', msg)
565
    if match:
566
        table, constraint = match.groups()
567
        constraint = sql_gen.Col(constraint, table)
568
        cond = None
569
        if recover: # need auto-rollback to run constraint_cond()
570
            try: cond = constraint_cond(db, constraint)
571
            except NotImplementedError: pass
572
        raise CheckException(constraint.to_str(db), cond, [], e)
573
    
574
    match = re.match(r'^(?:invalid input (?:syntax|value)\b[^:]*'
575
        r'|.+? out of range)(?:: "(.+?)")?', msg)
576
    if match:
577
        value, = match.groups()
578
        value = util.do_ignore_none(strings.to_unicode, value)
579
        raise InvalidValueException(value, e)
580
    
581
    match = re.match(r'^column "(.+?)" is of type (.+?) but expression '
582
        r'is of type', msg)
583
    if match:
584
        col, type_ = match.groups()
585
        raise MissingCastException(type_, col, e)
586
    
587
    match = re.match(r'^could not determine polymorphic type because '
588
        r'input has type "unknown"', msg)
589
    if match: raise MissingCastException('text', None, e)
590
    
591
    match = re.match(r'^.+? types (.+?) and (.+?) cannot be matched', msg)
592
    if match:
593
        type0, type1 = match.groups()
594
        raise MissingCastException(type0, None, e)
595
    
596
    match = re.match(r'^.*?\brelation "(.+?)" is not a (table)', msg)
597
    if match:
598
        name, type_ = match.groups()
599
        raise InvalidTypeException(type_, name, e)
600
    
601
    typed_name_re = r'^(\S+) "(.+?)"(?: of relation ".+?")?'
602
    
603
    match = re.match(typed_name_re+r'.*? already exists', msg)
604
    if match:
605
        type_, name = match.groups()
606
        raise DuplicateException(type_, name, e)
607
    
608
    match = re.match(r'more than one (\S+) named ""(.+?)""', msg)
609
    if match:
610
        type_, name = match.groups()
611
        raise DuplicateException(type_, name, e)
612
    
613
    match = re.match(typed_name_re+r' does not exist', msg)
614
    if match:
615
        type_, name = match.groups()
616
        if type_ == 'function':
617
            match = re.match(r'^(.+?)\(.*\)$', name)
618
            if match: # includes params, so is call rather than cast to regproc
619
                function_name, = match.groups()
620
                func = sql_gen.Function(function_name)
621
                if function_exists(db, func) and msg.find('CAST') < 0:
622
                    # not found only because of a missing cast
623
                    type_ = function_param0_type(db, func)
624
                    col = None
625
                    if type_ == 'anyelement': type_ = 'text'
626
                    elif type_ == 'hstore': # cast just the value param
627
                        type_ = 'text'
628
                        col = 'value'
629
                    raise MissingCastException(type_, col, e)
630
        raise DoesNotExistException(type_, name, e)
631
    
632
    raise # no specific exception raised
633

    
634
def with_savepoint(db, func): return db.with_savepoint(func)
635

    
636
def run_query(db, query, recover=None, cacheable=False, log_level=2,
637
    log_ignore_excs=None, **kw_args):
638
    '''For params, see DbConn.run_query()'''
639
    if recover == None: recover = False
640
    if log_ignore_excs == None: log_ignore_excs = ()
641
    log_ignore_excs = tuple(log_ignore_excs)
642
    debug_msg_ref = [None]
643
    
644
    query = with_explain_comment(db, query)
645
    
646
    try:
647
        try:
648
            def run(): return db.run_query(query, cacheable, log_level,
649
                debug_msg_ref, **kw_args)
650
            if recover and not db.is_cached(query):
651
                return with_savepoint(db, run)
652
            else: return run() # don't need savepoint if cached
653
        except Exception, e:
654
            # Give failed EXPLAIN approximately the log_level of its query
655
            if query.startswith('EXPLAIN'): log_level -= 1
656
            
657
            parse_exception(db, e, recover)
658
    except log_ignore_excs:
659
        log_level += 2
660
        raise
661
    finally:
662
        if debug_msg_ref[0] != None: db.log_debug(debug_msg_ref[0], log_level)
663

    
664
##### Basic queries
665

    
666
def is_explainable(query):
667
    # See <http://www.postgresql.org/docs/8.3/static/sql-explain.html#AEN57749>
668
    return re.match(r'^(?:SELECT|INSERT|UPDATE|DELETE|VALUES|EXECUTE|DECLARE)\b'
669
        , query)
670

    
671
def explain(db, query, **kw_args):
672
    '''
673
    For params, see run_query().
674
    '''
675
    kw_args.setdefault('log_level', 4)
676
    
677
    return strings.ustr(strings.join_lines(values(run_query(db,
678
        'EXPLAIN '+query, recover=True, cacheable=True, **kw_args))))
679
        # not a higher log_level because it's useful to see what query is being
680
        # run before it's executed, which EXPLAIN effectively provides
681

    
682
def has_comment(query): return query.endswith('*/')
683

    
684
def with_explain_comment(db, query, **kw_args):
685
    if db.autoexplain and not has_comment(query) and is_explainable(query):
686
        query += '\n'+sql_gen.esc_comment(' EXPLAIN:\n'
687
            +explain(db, query, **kw_args))
688
    return query
689

    
690
def next_version(name):
691
    version = 1 # first existing name was version 0
692
    match = re.match(r'^(.*)#(\d+)$', name)
693
    if match:
694
        name, version = match.groups()
695
        version = int(version)+1
696
    return sql_gen.concat(name, '#'+str(version))
697

    
698
def lock_table(db, table, mode):
699
    table = sql_gen.as_Table(table)
700
    run_query(db, 'LOCK TABLE '+table.to_str(db)+' IN '+mode+' MODE')
701

    
702
def run_query_into(db, query, into=None, add_pkey_=False, **kw_args):
703
    '''Outputs a query to a temp table.
704
    For params, see run_query().
705
    '''
706
    if into == None: return run_query(db, query, **kw_args)
707
    
708
    assert isinstance(into, sql_gen.Table)
709
    
710
    into.is_temp = True
711
    # "temporary tables cannot specify a schema name", so remove schema
712
    into.schema = None
713
    
714
    kw_args['recover'] = True
715
    kw_args.setdefault('log_ignore_excs', (DuplicateException,))
716
    
717
    temp = not db.debug_temp # tables are permanent in debug_temp mode
718
    
719
    # Create table
720
    while True:
721
        create_query = 'CREATE'
722
        if temp: create_query += ' TEMP'
723
        create_query += ' TABLE '+into.to_str(db)+' AS\n'+query
724
        
725
        try:
726
            cur = run_query(db, create_query, **kw_args)
727
                # CREATE TABLE AS sets rowcount to # rows in query
728
            break
729
        except DuplicateException, e:
730
            into.name = next_version(into.name)
731
            # try again with next version of name
732
    
733
    if add_pkey_: add_pkey_or_index(db, into, warn=True)
734
    
735
    # According to the PostgreSQL doc, "The autovacuum daemon cannot access and
736
    # therefore cannot vacuum or analyze temporary tables. [...] if a temporary
737
    # table is going to be used in complex queries, it is wise to run ANALYZE on
738
    # the temporary table after it is populated."
739
    # (http://www.postgresql.org/docs/9.1/static/sql-createtable.html)
740
    # If into is not a temp table, ANALYZE is useful but not required.
741
    analyze(db, into)
742
    
743
    return cur
744

    
745
order_by_pkey = object() # tells mk_select() to order by the pkey
746

    
747
distinct_on_all = object() # tells mk_select() to SELECT DISTINCT ON all columns
748

    
749
def has_subset_func(db, table):
750
    return sql_gen.is_underlying_table(table) and function_exists(db, table)
751

    
752
def mk_select(db, tables=None, fields=None, conds=None, distinct_on=[],
753
    limit=None, start=None, order_by=order_by_pkey, default_table=None,
754
    explain=True):
755
    '''
756
    @param tables The single table to select from, or a list of tables to join
757
        together, with tables after the first being sql_gen.Join objects
758
    @param fields Use None to select all fields in the table
759
    @param conds WHERE conditions: [(compare_left_side, compare_right_side),...]
760
        * container can be any iterable type
761
        * compare_left_side: sql_gen.Code|str (for col name)
762
        * compare_right_side: sql_gen.ValueCond|literal value
763
    @param distinct_on The columns to SELECT DISTINCT ON, or distinct_on_all to
764
        use all columns
765
    @return query
766
    '''
767
    # Parse tables param
768
    tables = lists.mk_seq(tables)
769
    tables = list(tables) # don't modify input! (list() copies input)
770
    table0 = sql_gen.as_Table(tables.pop(0)) # first table is separate
771
    
772
    # Parse other params
773
    if conds == None: conds = []
774
    elif dicts.is_dict(conds): conds = conds.items()
775
    conds = list(conds) # don't modify input! (list() copies input)
776
    assert limit == None or isinstance(limit, (int, long))
777
    assert start == None or isinstance(start, (int, long))
778
    if limit == 0: order_by = None
779
    if order_by in (None, order_by_pkey) and has_subset_func(db, table0):
780
        # can use subset function for fast querying at large OFFSET values
781
        run_query(db, 'SET LOCAL enable_sort TO off')
782
        table0 = sql_gen.FunctionCall(table0, limit_=limit, offset_=start)
783
        if limit != 0: limit = None # done by function
784
        start = None # done by function
785
        order_by = None # done by function
786
    elif order_by is order_by_pkey:
787
        if lists.is_seq(distinct_on) and distinct_on: order_by = distinct_on[0]
788
        elif table0 != None: order_by = table_order_by(db, table0, recover=True)
789
        else: order_by = None
790
    
791
    query = 'SELECT'
792
    
793
    def parse_col(col): return sql_gen.as_Col(col, default_table).to_str(db)
794
    
795
    # DISTINCT ON columns
796
    if distinct_on != []:
797
        query += '\nDISTINCT'
798
        if distinct_on is not distinct_on_all:
799
            query += ' ON ('+(', '.join(map(parse_col, distinct_on)))+')'
800
    
801
    # Columns
802
    if query.find('\n') >= 0: whitespace = '\n'
803
    else: whitespace = ' '
804
    if fields == None: query += whitespace+'*'
805
    else:
806
        assert fields != []
807
        if len(fields) > 1: whitespace = '\n'
808
        query += whitespace+('\n, '.join(map(parse_col, fields)))
809
    
810
    # Main table
811
    if query.find('\n') >= 0 or len(tables) > 0: whitespace = '\n'
812
    else: whitespace = ' '
813
    if table0 != None: query += whitespace+'FROM '+table0.to_str(db)
814
    
815
    # Add joins
816
    left_table = table0
817
    for join_ in tables:
818
        table = join_.table
819
        
820
        # Parse special values
821
        if join_.type_ is sql_gen.filter_out: # filter no match
822
            conds.append((sql_gen.Col(table_not_null_col(db, table), table),
823
                sql_gen.CompareCond(None, '~=')))
824
        
825
        query += '\n'+join_.to_str(db, left_table)
826
        
827
        left_table = table
828
    
829
    missing = True
830
    if conds != []:
831
        if len(conds) == 1: whitespace = ' '
832
        else: whitespace = '\n'
833
        query += '\n'+sql_gen.combine_conds([sql_gen.ColValueCond(l, r)
834
            .to_str(db) for l, r in conds], 'WHERE')
835
    if order_by != None:
836
        query += '\nORDER BY '+sql_gen.as_Col(order_by).to_str(db)
837
    if limit != None: query += '\nLIMIT '+str(limit)
838
    if start != None:
839
        if start != 0: query += '\nOFFSET '+str(start)
840
    
841
    if explain: query = with_explain_comment(db, query)
842
    
843
    return query
844

    
845
def select(db, *args, **kw_args):
846
    '''For params, see mk_select() and run_query()'''
847
    recover = kw_args.pop('recover', None)
848
    cacheable = kw_args.pop('cacheable', True)
849
    log_level = kw_args.pop('log_level', 2)
850
    
851
    return run_query(db, mk_select(db, *args, **kw_args), recover, cacheable,
852
        log_level=log_level)
853

    
854
def mk_insert_select(db, table, cols=None, select_query=None, returning=None,
855
    embeddable=False, ignore=False, src=None):
856
    '''
857
    @param returning str|None An inserted column (such as pkey) to return
858
    @param embeddable Whether the query should be embeddable as a nested SELECT.
859
        Warning: If you set this and cacheable=True when the query is run, the
860
        query will be fully cached, not just if it raises an exception.
861
    @param ignore Whether to ignore duplicate keys.
862
    @param src Will be included in the name of any created function, to help
863
        identify the data source in pg_stat_activity.
864
    '''
865
    table = sql_gen.remove_table_rename(sql_gen.as_Table(table))
866
    if cols == []: cols = None # no cols (all defaults) = unknown col names
867
    if cols != None: cols = [sql_gen.to_name_only_col(c, table) for c in cols]
868
    if select_query == None: select_query = 'DEFAULT VALUES'
869
    if returning != None: returning = sql_gen.as_Col(returning, table)
870
    
871
    first_line = 'INSERT INTO '+table.to_str(db)
872
    
873
    def mk_insert(select_query):
874
        query = first_line
875
        if cols != None:
876
            query += '\n('+(', '.join((c.to_str(db) for c in cols)))+')'
877
        query += '\n'+select_query
878
        
879
        if returning != None:
880
            returning_name_col = sql_gen.to_name_only_col(returning)
881
            query += '\nRETURNING '+returning_name_col.to_str(db)
882
        
883
        return query
884
    
885
    return_type = sql_gen.CustomCode('unknown')
886
    if returning != None: return_type = sql_gen.ColType(returning)
887
    
888
    if ignore:
889
        # Always return something to set the correct rowcount
890
        if returning == None: returning = sql_gen.NamedCol('NULL', None)
891
        
892
        embeddable = True # must use function
893
        
894
        if cols == None: row = [sql_gen.Col(sql_gen.all_cols, 'row')]
895
        else: row = [sql_gen.Col(c.name, 'row') for c in cols]
896
        
897
        query = sql_gen.RowExcIgnore(sql_gen.RowType(table), select_query,
898
            sql_gen.ReturnQuery(mk_insert(sql_gen.Values(row).to_str(db))),
899
            cols)
900
    else: query = mk_insert(select_query)
901
    
902
    if embeddable:
903
        # Create function
904
        function_name = sql_gen.clean_name(first_line)
905
        if src != None: function_name = src+': '+function_name
906
        while True:
907
            try:
908
                func = db.TempFunction(function_name)
909
                def_ = sql_gen.FunctionDef(func, sql_gen.SetOf(return_type),
910
                    query)
911
                
912
                run_query(db, def_.to_str(db), recover=True, cacheable=True,
913
                    log_ignore_excs=(DuplicateException,))
914
                break # this version was successful
915
            except DuplicateException, e:
916
                function_name = next_version(function_name)
917
                # try again with next version of name
918
        
919
        # Return query that uses function
920
        cols = None
921
        if returning != None: cols = [returning]
922
        func_table = sql_gen.NamedTable('f', sql_gen.FunctionCall(func), cols)
923
            # AS clause requires function alias
924
        return mk_select(db, func_table, order_by=None)
925
    
926
    return query
927

    
928
def insert_select(db, table, *args, **kw_args):
929
    '''For params, see mk_insert_select() and run_query_into()
930
    @param into sql_gen.Table with suggested name of temp table to put RETURNING
931
        values in
932
    '''
933
    returning = kw_args.get('returning', None)
934
    ignore = kw_args.get('ignore', False)
935
    
936
    into = kw_args.pop('into', None)
937
    if into != None: kw_args['embeddable'] = True
938
    recover = kw_args.pop('recover', None)
939
    if ignore: recover = True
940
    cacheable = kw_args.pop('cacheable', True)
941
    log_level = kw_args.pop('log_level', 2)
942
    
943
    rowcount_only = ignore and returning == None # keep NULL rows on server
944
    if rowcount_only: into = sql_gen.Table('rowcount')
945
    
946
    cur = run_query_into(db, mk_insert_select(db, table, *args, **kw_args),
947
        into, recover=recover, cacheable=cacheable, log_level=log_level)
948
    if rowcount_only: empty_temp(db, into)
949
    autoanalyze(db, table)
950
    return cur
951

    
952
default = sql_gen.default # tells insert() to use the default value for a column
953

    
954
def insert(db, table, row, *args, **kw_args):
955
    '''For params, see insert_select()'''
956
    ignore = kw_args.pop('ignore', False)
957
    if ignore: kw_args.setdefault('recover', True)
958
    
959
    if lists.is_seq(row): cols = None
960
    else:
961
        cols = row.keys()
962
        row = row.values()
963
    row = list(row) # ensure that "== []" works
964
    
965
    if row == []: query = None
966
    else: query = sql_gen.Values(row).to_str(db)
967
    
968
    try: return insert_select(db, table, cols, query, *args, **kw_args)
969
    except (DuplicateKeyException, NullValueException):
970
        if not ignore: raise
971
        return None
972

    
973
def mk_update(db, table, changes=None, cond=None, in_place=False,
974
    cacheable_=True):
975
    '''
976
    @param changes [(col, new_value),...]
977
        * container can be any iterable type
978
        * col: sql_gen.Code|str (for col name)
979
        * new_value: sql_gen.Code|literal value
980
    @param cond sql_gen.Code WHERE condition. e.g. use sql_gen.*Cond objects.
981
    @param in_place If set, locks the table and updates rows in place.
982
        This avoids creating dead rows in PostgreSQL.
983
        * cond must be None
984
    @param cacheable_ Whether column structure information used to generate the
985
        query can be cached
986
    @return str query
987
    '''
988
    table = sql_gen.as_Table(table)
989
    changes = [(sql_gen.to_name_only_col(c, table), sql_gen.as_Value(v))
990
        for c, v in changes]
991
    
992
    if in_place:
993
        assert cond == None
994
        
995
        def col_type(col):
996
            return sql_gen.canon_type(db.col_info(
997
                sql_gen.with_default_table(c, table), cacheable_).type)
998
        changes = [(c, v, col_type(c)) for c, v in changes]
999
        query = 'ALTER TABLE '+table.to_str(db)+'\n'
1000
        query += ',\n'.join(('ALTER COLUMN '+c.to_str(db)+' TYPE '+t+'\nUSING '
1001
            +v.to_str(db) for c, v, t in changes))
1002
    else:
1003
        query = 'UPDATE '+table.to_str(db)+'\nSET\n'
1004
        query += ',\n'.join((c.to_str(db)+' = '+v.to_str(db)
1005
            for c, v in changes))
1006
        if cond != None: query += '\nWHERE\n'+cond.to_str(db)
1007
    
1008
    query = with_explain_comment(db, query)
1009
    
1010
    return query
1011

    
1012
def update(db, table, *args, **kw_args):
1013
    '''For params, see mk_update() and run_query()'''
1014
    recover = kw_args.pop('recover', None)
1015
    cacheable = kw_args.pop('cacheable', False)
1016
    log_level = kw_args.pop('log_level', 2)
1017
    
1018
    cur = run_query(db, mk_update(db, table, *args, **kw_args), recover,
1019
        cacheable, log_level=log_level)
1020
    autoanalyze(db, table)
1021
    return cur
1022

    
1023
def mk_delete(db, table, cond=None):
1024
    '''
1025
    @param cond sql_gen.Code WHERE condition. e.g. use sql_gen.*Cond objects.
1026
    @return str query
1027
    '''
1028
    query = 'DELETE FROM '+table.to_str(db)
1029
    if cond != None: query += '\nWHERE '+cond.to_str(db)
1030
    
1031
    query = with_explain_comment(db, query)
1032
    
1033
    return query
1034

    
1035
def delete(db, table, *args, **kw_args):
1036
    '''For params, see mk_delete() and run_query()'''
1037
    recover = kw_args.pop('recover', None)
1038
    cacheable = kw_args.pop('cacheable', True)
1039
    log_level = kw_args.pop('log_level', 2)
1040
    
1041
    cur = run_query(db, mk_delete(db, table, *args, **kw_args), recover,
1042
        cacheable, log_level=log_level)
1043
    autoanalyze(db, table)
1044
    return cur
1045

    
1046
def last_insert_id(db):
1047
    module = util.root_module(db.db)
1048
    if module == 'psycopg2': return value(run_query(db, 'SELECT lastval()'))
1049
    elif module == 'MySQLdb': return db.insert_id()
1050
    else: return None
1051

    
1052
def define_func(db, def_):
1053
    func = def_.function
1054
    while True:
1055
        try:
1056
            run_query(db, def_.to_str(db), recover=True, cacheable=True,
1057
                log_ignore_excs=(DuplicateException,))
1058
            break # successful
1059
        except DuplicateException:
1060
            func.name = next_version(func.name)
1061
            # try again with next version of name
1062

    
1063
def mk_flatten_mapping(db, into, cols, preserve=[], as_items=False):
1064
    '''Creates a mapping from original column names (which may have collisions)
1065
    to names that will be distinct among the columns' tables.
1066
    This is meant to be used for several tables that are being joined together.
1067
    @param cols The columns to combine. Duplicates will be removed.
1068
    @param into The table for the new columns.
1069
    @param preserve [sql_gen.Col...] Columns not to rename. Note that these
1070
        columns will be included in the mapping even if they are not in cols.
1071
        The tables of the provided Col objects will be changed to into, so make
1072
        copies of them if you want to keep the original tables.
1073
    @param as_items Whether to return a list of dict items instead of a dict
1074
    @return dict(orig_col=new_col, ...)
1075
        * orig_col: sql_gen.Col(orig_col_name, orig_table)
1076
        * new_col: sql_gen.Col(orig_col_name, into)
1077
        * All mappings use the into table so its name can easily be
1078
          changed for all columns at once
1079
    '''
1080
    cols = lists.uniqify(cols)
1081
    
1082
    items = []
1083
    for col in preserve:
1084
        orig_col = copy.copy(col)
1085
        col.table = into
1086
        items.append((orig_col, col))
1087
    preserve = set(preserve)
1088
    for col in cols:
1089
        if col not in preserve:
1090
            items.append((col, sql_gen.Col(strings.ustr(col), into, col.srcs)))
1091
    
1092
    if not as_items: items = dict(items)
1093
    return items
1094

    
1095
def flatten(db, into, joins, cols, limit=None, start=None, **kw_args):
1096
    '''For params, see mk_flatten_mapping()
1097
    @return See return value of mk_flatten_mapping()
1098
    '''
1099
    items = mk_flatten_mapping(db, into, cols, as_items=True, **kw_args)
1100
    cols = [sql_gen.NamedCol(new.name, old) for old, new in items]
1101
    run_query_into(db, mk_select(db, joins, cols, limit=limit, start=start),
1102
        into=into, add_pkey_=True)
1103
        # don't cache because the temp table will usually be truncated after use
1104
    return dict(items)
1105

    
1106
##### Database structure introspection
1107

    
1108
#### Tables
1109

    
1110
def tables(db, schema_like='public', table_like='%', exact=False,
1111
    cacheable=True):
1112
    if exact: compare = '='
1113
    else: compare = 'LIKE'
1114
    
1115
    module = util.root_module(db.db)
1116
    if module == 'psycopg2':
1117
        conds = [('schemaname', sql_gen.CompareCond(schema_like, compare)),
1118
            ('tablename', sql_gen.CompareCond(table_like, compare))]
1119
        return values(select(db, 'pg_tables', ['tablename'], conds,
1120
            order_by='tablename', cacheable=cacheable, log_level=4))
1121
    elif module == 'MySQLdb':
1122
        return values(run_query(db, 'SHOW TABLES LIKE '+db.esc_value(table_like)
1123
            , cacheable=True, log_level=4))
1124
    else: raise NotImplementedError("Can't list tables for "+module+' database')
1125

    
1126
def table_exists(db, table, cacheable=True):
1127
    table = sql_gen.as_Table(table)
1128
    return list(tables(db, table.schema, table.name, True, cacheable)) != []
1129

    
1130
def table_row_count(db, table, recover=None):
1131
    return value(run_query(db, mk_select(db, table, [sql_gen.row_count],
1132
        order_by=None), recover=recover, log_level=3))
1133

    
1134
def table_col_names(db, table, recover=None):
1135
    return list(col_names(select(db, table, limit=0, recover=recover,
1136
        log_level=4)))
1137

    
1138
def table_cols(db, table, *args, **kw_args):
1139
    return [sql_gen.as_Col(strings.ustr(c), table)
1140
        for c in table_col_names(db, table, *args, **kw_args)]
1141

    
1142
def table_pkey_index(db, table, recover=None):
1143
    table_str = sql_gen.Literal(table.to_str(db))
1144
    try:
1145
        return sql_gen.Table(value(run_query(db, '''\
1146
SELECT relname
1147
FROM pg_index
1148
JOIN pg_class index ON index.oid = indexrelid
1149
WHERE
1150
indrelid = '''+table_str.to_str(db)+'''::regclass
1151
AND indisprimary
1152
'''
1153
            , recover, cacheable=True, log_level=4)), table.schema)
1154
    except StopIteration: raise DoesNotExistException('primary key', '')
1155

    
1156
def table_pkey_col(db, table, recover=None):
1157
    table = sql_gen.as_Table(table)
1158
    
1159
    module = util.root_module(db.db)
1160
    if module == 'psycopg2':
1161
        return sql_gen.Col(index_cols(db, table_pkey_index(db, table,
1162
            recover))[0], table)
1163
    else:
1164
        join_cols = ['table_schema', 'table_name', 'constraint_schema',
1165
            'constraint_name']
1166
        tables = [sql_gen.Table('key_column_usage', 'information_schema'),
1167
            sql_gen.Join(
1168
                sql_gen.Table('table_constraints', 'information_schema'),
1169
                dict(((c, sql_gen.join_same_not_null) for c in join_cols)))]
1170
        cols = [sql_gen.Col('column_name')]
1171
        
1172
        conds = [('constraint_type', 'PRIMARY KEY'), ('table_name', table.name)]
1173
        schema = table.schema
1174
        if schema != None: conds.append(('table_schema', schema))
1175
        order_by = 'position_in_unique_constraint'
1176
        
1177
        try: return sql_gen.Col(value(select(db, tables, cols, conds,
1178
            order_by=order_by, limit=1, log_level=4)), table)
1179
        except StopIteration: raise DoesNotExistException('primary key', '')
1180

    
1181
def table_has_pkey(db, table, recover=None):
1182
    try: table_pkey_col(db, table, recover)
1183
    except DoesNotExistException: return False
1184
    else: return True
1185

    
1186
def pkey_name(db, table, recover=None):
1187
    '''If no pkey, returns the first column in the table.'''
1188
    return pkey_col(db, table, recover).name
1189

    
1190
def pkey_col(db, table, recover=None):
1191
    '''If no pkey, returns the first column in the table.'''
1192
    try: return table_pkey_col(db, table, recover)
1193
    except DoesNotExistException: return table_cols(db, table, recover)[0]
1194

    
1195
not_null_col = 'not_null_col'
1196

    
1197
def table_not_null_col(db, table, recover=None):
1198
    '''Name assumed to be the value of not_null_col. If not found, uses pkey.'''
1199
    if not_null_col in table_col_names(db, table, recover): return not_null_col
1200
    else: return pkey_name(db, table, recover)
1201

    
1202
def constraint_cond(db, constraint):
1203
    module = util.root_module(db.db)
1204
    if module == 'psycopg2':
1205
        table_str = sql_gen.Literal(constraint.table.to_str(db))
1206
        name_str = sql_gen.Literal(constraint.name)
1207
        return value(run_query(db, '''\
1208
SELECT consrc
1209
FROM pg_constraint
1210
WHERE
1211
conrelid = '''+table_str.to_str(db)+'''::regclass
1212
AND conname = '''+name_str.to_str(db)+'''
1213
'''
1214
            , cacheable=True, log_level=4))
1215
    else: raise NotImplementedError("Can't get constraint condition for "
1216
        +module+' database')
1217

    
1218
def index_exprs(db, index):
1219
    index = sql_gen.as_Table(index)
1220
    module = util.root_module(db.db)
1221
    if module == 'psycopg2':
1222
        qual_index = sql_gen.Literal(index.to_str(db))
1223
        return list(values(run_query(db, '''\
1224
SELECT pg_get_indexdef(indexrelid, generate_series(1, indnatts), true)
1225
FROM pg_index
1226
WHERE indexrelid = '''+qual_index.to_str(db)+'''::regclass
1227
'''
1228
            , cacheable=True, log_level=4)))
1229
    else: raise NotImplementedError()
1230

    
1231
def index_cols(db, index):
1232
    '''Can also use this for UNIQUE constraints, because a UNIQUE index is
1233
    automatically created. When you don't know whether something is a UNIQUE
1234
    constraint or a UNIQUE index, use this function.'''
1235
    return map(sql_gen.parse_expr_col, index_exprs(db, index))
1236

    
1237
def index_cond(db, index):
1238
    index = sql_gen.as_Table(index)
1239
    module = util.root_module(db.db)
1240
    if module == 'psycopg2':
1241
        qual_index = sql_gen.Literal(index.to_str(db))
1242
        return value(run_query(db, '''\
1243
SELECT pg_get_expr(indpred, indrelid, true)
1244
FROM pg_index
1245
WHERE indexrelid = '''+qual_index.to_str(db)+'''::regclass
1246
'''
1247
            , cacheable=True, log_level=4))
1248
    else: raise NotImplementedError()
1249

    
1250
def index_order_by(db, index):
1251
    return sql_gen.CustomCode(', '.join(index_exprs(db, index)))
1252

    
1253
def table_cluster_on(db, table, recover=None):
1254
    '''
1255
    @return The table's cluster index, or its pkey if none is set
1256
    '''
1257
    table_str = sql_gen.Literal(table.to_str(db))
1258
    try:
1259
        return sql_gen.Table(value(run_query(db, '''\
1260
SELECT relname
1261
FROM pg_index
1262
JOIN pg_class index ON index.oid = indexrelid
1263
WHERE
1264
indrelid = '''+table_str.to_str(db)+'''::regclass
1265
AND indisclustered
1266
'''
1267
            , recover, cacheable=True, log_level=4)), table.schema)
1268
    except StopIteration: return table_pkey_index(db, table, recover)
1269

    
1270
def table_order_by(db, table, recover=None):
1271
    '''
1272
    @return None if table is view, because table_cluster_on() would return None
1273
    '''
1274
    if table.order_by == None:
1275
        try: table.order_by = index_order_by(db, table_cluster_on(db, table,
1276
            recover))
1277
        except DoesNotExistException: pass
1278
    return table.order_by
1279

    
1280
#### Views
1281

    
1282
def view_exists(db, view):
1283
    view_str = sql_gen.Literal(view.to_str(db))
1284
    try:
1285
        return value(run_query(db, '''\
1286
SELECT relkind = 'v'
1287
FROM pg_class
1288
WHERE oid = '''+view_str.to_str(db)+'''::regclass
1289
'''
1290
            , cacheable=True, log_level=4))
1291
    except DoesNotExistException: return False
1292

    
1293
#### Functions
1294

    
1295
def function_exists(db, function):
1296
    qual_function = sql_gen.Literal(function.to_str(db))
1297
    try:
1298
        select(db, fields=[sql_gen.Cast('regproc', qual_function)],
1299
            recover=True, cacheable=True, log_level=4)
1300
    except DoesNotExistException: return False
1301
    except DuplicateException: return True # overloaded function
1302
    else: return True
1303

    
1304
def function_param0_type(db, function):
1305
    qual_function = sql_gen.Literal(function.to_str(db))
1306
    return value(run_query(db, '''\
1307
SELECT proargtypes[0]::regtype
1308
FROM pg_proc
1309
WHERE oid = '''+qual_function.to_str(db)+'''::regproc
1310
'''
1311
        , cacheable=True, log_level=4))
1312

    
1313
##### Structural changes
1314

    
1315
#### Columns
1316

    
1317
def add_col(db, table, col, comment=None, if_not_exists=False, **kw_args):
1318
    '''
1319
    @param col TypedCol Name may be versioned, so be sure to propagate any
1320
        renaming back to any source column for the TypedCol.
1321
    @param comment None|str SQL comment used to distinguish columns of the same
1322
        name from each other when they contain different data, to allow the
1323
        ADD COLUMN query to be cached. If not set, query will not be cached.
1324
    '''
1325
    assert isinstance(col, sql_gen.TypedCol)
1326
    
1327
    while True:
1328
        str_ = 'ALTER TABLE '+table.to_str(db)+' ADD COLUMN '+col.to_str(db)
1329
        if comment != None: str_ += ' '+sql_gen.esc_comment(comment)
1330
        
1331
        try:
1332
            run_query(db, str_, recover=True, cacheable=True, **kw_args)
1333
            break
1334
        except DuplicateException:
1335
            if if_not_exists: raise
1336
            col.name = next_version(col.name)
1337
            # try again with next version of name
1338

    
1339
def add_not_null(db, col):
1340
    table = col.table
1341
    col = sql_gen.to_name_only_col(col)
1342
    run_query(db, 'ALTER TABLE '+table.to_str(db)+' ALTER COLUMN '
1343
        +col.to_str(db)+' SET NOT NULL', cacheable=True, log_level=3)
1344

    
1345
def drop_not_null(db, col):
1346
    table = col.table
1347
    col = sql_gen.to_name_only_col(col)
1348
    run_query(db, 'ALTER TABLE '+table.to_str(db)+' ALTER COLUMN '
1349
        +col.to_str(db)+' DROP NOT NULL', cacheable=True, log_level=3)
1350

    
1351
row_num_col = '_row_num'
1352

    
1353
row_num_col_def = sql_gen.TypedCol('', 'serial', nullable=False,
1354
    constraints='PRIMARY KEY')
1355

    
1356
def add_row_num(db, table, name=row_num_col):
1357
    '''Adds a row number column to a table. Its definition is in
1358
    row_num_col_def. It will be the primary key.'''
1359
    col_def = copy.copy(row_num_col_def)
1360
    col_def.name = name
1361
    add_col(db, table, col_def, comment='', if_not_exists=True, log_level=3)
1362

    
1363
#### Indexes
1364

    
1365
def add_pkey(db, table, cols=None, recover=None):
1366
    '''Adds a primary key.
1367
    @param cols [sql_gen.Col,...] The columns in the primary key.
1368
        Defaults to the first column in the table.
1369
    @pre The table must not already have a primary key.
1370
    '''
1371
    table = sql_gen.as_Table(table)
1372
    if cols == None: cols = [pkey_name(db, table, recover)]
1373
    col_strs = [sql_gen.to_name_only_col(v).to_str(db) for v in cols]
1374
    
1375
    run_query(db, 'ALTER TABLE '+table.to_str(db)+' ADD PRIMARY KEY ('
1376
        +(', '.join(col_strs))+')', recover=True, cacheable=True, log_level=3,
1377
        log_ignore_excs=(DuplicateException,))
1378

    
1379
def add_index(db, exprs, table=None, unique=False, ensure_not_null_=True):
1380
    '''Adds an index on column(s) or expression(s) if it doesn't already exist.
1381
    Currently, only function calls and literal values are supported expressions.
1382
    @param ensure_not_null_ If set, translates NULL values to sentinel values.
1383
        This allows indexes to be used for comparisons where NULLs are equal.
1384
    '''
1385
    exprs = lists.mk_seq(exprs)
1386
    
1387
    # Parse exprs
1388
    old_exprs = exprs[:]
1389
    exprs = []
1390
    cols = []
1391
    for i, expr in enumerate(old_exprs):
1392
        expr = sql_gen.as_Col(expr, table)
1393
        
1394
        # Handle nullable columns
1395
        if ensure_not_null_:
1396
            try: expr = sql_gen.ensure_not_null(db, expr)
1397
            except KeyError: pass # unknown type, so just create plain index
1398
        
1399
        # Extract col
1400
        expr = copy.deepcopy(expr) # don't modify input!
1401
        col = expr
1402
        if isinstance(expr, sql_gen.FunctionCall): col = expr.args[0]
1403
        expr = sql_gen.cast_literal(expr)
1404
        if not isinstance(expr, (sql_gen.Expr, sql_gen.Col)):
1405
            expr = sql_gen.Expr(expr)
1406
            
1407
        
1408
        # Extract table
1409
        if table == None:
1410
            assert sql_gen.is_table_col(col)
1411
            table = col.table
1412
        
1413
        if isinstance(col, sql_gen.Col): col.table = None
1414
        
1415
        exprs.append(expr)
1416
        cols.append(col)
1417
    
1418
    table = sql_gen.as_Table(table)
1419
    
1420
    # Add index
1421
    str_ = 'CREATE'
1422
    if unique: str_ += ' UNIQUE'
1423
    str_ += ' INDEX ON '+table.to_str(db)+' ('+(
1424
        ', '.join((v.to_str(db) for v in exprs)))+')'
1425
    run_query(db, str_, recover=True, cacheable=True, log_level=3)
1426

    
1427
def add_pkey_index(db, table): add_index(db, pkey_col(db, table), table)
1428

    
1429
def add_pkey_or_index(db, table, cols=None, recover=None, warn=False):
1430
    try: add_pkey(db, table, cols, recover)
1431
    except DuplicateKeyException, e:
1432
        if warn: warnings.warn(UserWarning(exc.str_(e)))
1433
        add_pkey_index(db, table)
1434

    
1435
already_indexed = object() # tells add_indexes() the pkey has already been added
1436

    
1437
def add_indexes(db, table, has_pkey=True):
1438
    '''Adds an index on all columns in a table.
1439
    @param has_pkey bool|already_indexed Whether a pkey instead of a regular
1440
        index should be added on the first column.
1441
        * If already_indexed, the pkey is assumed to have already been added
1442
    '''
1443
    cols = table_col_names(db, table)
1444
    if has_pkey:
1445
        if has_pkey is not already_indexed: add_pkey(db, table)
1446
        cols = cols[1:]
1447
    for col in cols: add_index(db, col, table)
1448

    
1449
#### Tables
1450

    
1451
### Maintenance
1452

    
1453
def analyze(db, table):
1454
    table = sql_gen.as_Table(table)
1455
    run_query(db, 'ANALYZE '+table.to_str(db), log_level=3)
1456

    
1457
def autoanalyze(db, table):
1458
    if db.autoanalyze: analyze(db, table)
1459

    
1460
def vacuum(db, table):
1461
    table = sql_gen.as_Table(table)
1462
    db.with_autocommit(lambda: run_query(db, 'VACUUM ANALYZE '+table.to_str(db),
1463
        log_level=3))
1464

    
1465
### Lifecycle
1466

    
1467
def drop(db, type_, name):
1468
    name = sql_gen.as_Name(name)
1469
    run_query(db, 'DROP '+type_+' IF EXISTS '+name.to_str(db)+' CASCADE')
1470

    
1471
def drop_table(db, table): drop(db, 'TABLE', table)
1472

    
1473
def create_table(db, table, cols=[], has_pkey=True, col_indexes=True,
1474
    like=None):
1475
    '''Creates a table.
1476
    @param cols [sql_gen.TypedCol,...] The column names and types
1477
    @param has_pkey If set, the first column becomes the primary key.
1478
    @param col_indexes bool|[ref]
1479
        * If True, indexes will be added on all non-pkey columns.
1480
        * If a list reference, [0] will be set to a function to do this.
1481
          This can be used to delay index creation until the table is populated.
1482
    '''
1483
    table = sql_gen.as_Table(table)
1484
    
1485
    if like != None:
1486
        cols = [sql_gen.CustomCode('LIKE '+like.to_str(db)+' INCLUDING ALL')
1487
            ]+cols
1488
        table.order_by = like.order_by
1489
    if has_pkey:
1490
        cols[0] = pkey = copy.copy(cols[0]) # don't modify input!
1491
        pkey.constraints = 'PRIMARY KEY'
1492
    
1493
    temp = table.is_temp and not db.debug_temp
1494
        # temp tables permanent in debug_temp mode
1495
    
1496
    # Create table
1497
    def create():
1498
        str_ = 'CREATE'
1499
        if temp: str_ += ' TEMP'
1500
        str_ += ' TABLE '+table.to_str(db)+' (\n'
1501
        str_ += '\n, '.join(c.to_str(db) for c in cols)
1502
        str_ += '\n);'
1503
        
1504
        opts = dict(recover=True, cacheable=True, log_level=2,
1505
            log_ignore_excs=(DuplicateException,))
1506
        try: run_query(db, str_, **opts)
1507
        except InvalidTypeException: # try again as view
1508
            run_query_into(db, mk_select(db, like, limit=0), into=table, **opts)
1509
    if table.is_temp:
1510
        while True:
1511
            try:
1512
                create()
1513
                break
1514
            except DuplicateException:
1515
                table.name = next_version(table.name)
1516
                # try again with next version of name
1517
    else: create()
1518
    
1519
    # Add indexes
1520
    if has_pkey: has_pkey = already_indexed
1521
    def add_indexes_(): add_indexes(db, table, has_pkey)
1522
    if isinstance(col_indexes, list): col_indexes[0] = add_indexes_ # defer
1523
    elif col_indexes: add_indexes_() # add now
1524

    
1525
def copy_table_struct(db, src, dest):
1526
    '''Creates a structure-only copy of a table. (Does not copy data.)'''
1527
    create_table(db, dest, has_pkey=False, col_indexes=False, like=src)
1528

    
1529
def copy_table(db, src, dest):
1530
    '''Creates a copy of a table, including data'''
1531
    copy_table_struct(db, src, dest)
1532
    insert_select(db, dest, None, mk_select(db, src))
1533

    
1534
### Data
1535

    
1536
def truncate(db, table, schema='public', **kw_args):
1537
    '''For params, see run_query()'''
1538
    table = sql_gen.as_Table(table, schema)
1539
    return run_query(db, 'TRUNCATE '+table.to_str(db)+' CASCADE', **kw_args)
1540

    
1541
def empty_temp(db, tables):
1542
    tables = lists.mk_seq(tables)
1543
    for table in tables: truncate(db, table, log_level=3)
1544

    
1545
def empty_db(db, schema='public', **kw_args):
1546
    '''For kw_args, see tables()'''
1547
    for table in tables(db, schema, **kw_args): truncate(db, table, schema)
1548

    
1549
def distinct_table(db, table, distinct_on, joins=None):
1550
    '''Creates a copy of a temp table which is distinct on the given columns.
1551
    Adds an index on table's distinct_on columns, to facilitate merge joins.
1552
    @param distinct_on If empty, creates a table with one row. This is useful if
1553
        your distinct_on columns are all literal values.
1554
    @param joins The joins to use when creating the new table
1555
    @return The new table.
1556
    '''
1557
    if joins == None: joins = [table]
1558
    
1559
    new_table = sql_gen.suffixed_table(table, '_distinct')
1560
    distinct_on = filter(sql_gen.is_table_col, distinct_on)
1561
    
1562
    copy_table_struct(db, table, new_table)
1563
    
1564
    limit = None
1565
    if distinct_on == []: limit = 1 # one sample row
1566
    else: add_index(db, distinct_on, table) # for join optimization
1567
    
1568
    insert_select(db, new_table, None, mk_select(db, joins,
1569
        [sql_gen.Col(sql_gen.all_cols, table)], distinct_on=distinct_on,
1570
        order_by=None, limit=limit))
1571
    analyze(db, new_table)
1572
    
1573
    return new_table
(35-35/51)