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

    
835
def select(db, *args, **kw_args):
836
    '''For params, see mk_select() and run_query()'''
837
    recover = kw_args.pop('recover', None)
838
    cacheable = kw_args.pop('cacheable', True)
839
    log_level = kw_args.pop('log_level', 2)
840
    
841
    return run_query(db, mk_select(db, *args, **kw_args), recover, cacheable,
842
        log_level=log_level)
843

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

    
918
def insert_select(db, table, *args, **kw_args):
919
    '''For params, see mk_insert_select() and run_query_into()
920
    @param into sql_gen.Table with suggested name of temp table to put RETURNING
921
        values in
922
    '''
923
    returning = kw_args.get('returning', None)
924
    ignore = kw_args.get('ignore', False)
925
    
926
    into = kw_args.pop('into', None)
927
    if into != None: kw_args['embeddable'] = True
928
    recover = kw_args.pop('recover', None)
929
    if ignore: recover = True
930
    cacheable = kw_args.pop('cacheable', True)
931
    log_level = kw_args.pop('log_level', 2)
932
    
933
    rowcount_only = ignore and returning == None # keep NULL rows on server
934
    if rowcount_only: into = sql_gen.Table('rowcount')
935
    
936
    cur = run_query_into(db, mk_insert_select(db, table, *args, **kw_args),
937
        into, recover=recover, cacheable=cacheable, log_level=log_level)
938
    if rowcount_only: empty_temp(db, into)
939
    autoanalyze(db, table)
940
    return cur
941

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

    
944
def insert(db, table, row, *args, **kw_args):
945
    '''For params, see insert_select()'''
946
    ignore = kw_args.pop('ignore', False)
947
    if ignore: kw_args.setdefault('recover', True)
948
    
949
    if lists.is_seq(row): cols = None
950
    else:
951
        cols = row.keys()
952
        row = row.values()
953
    row = list(row) # ensure that "== []" works
954
    
955
    if row == []: query = None
956
    else: query = sql_gen.Values(row).to_str(db)
957
    
958
    try: return insert_select(db, table, cols, query, *args, **kw_args)
959
    except (DuplicateKeyException, NullValueException):
960
        if not ignore: raise
961
        return None
962

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

    
1002
def update(db, table, *args, **kw_args):
1003
    '''For params, see mk_update() and run_query()'''
1004
    recover = kw_args.pop('recover', None)
1005
    cacheable = kw_args.pop('cacheable', False)
1006
    log_level = kw_args.pop('log_level', 2)
1007
    
1008
    cur = run_query(db, mk_update(db, table, *args, **kw_args), recover,
1009
        cacheable, log_level=log_level)
1010
    autoanalyze(db, table)
1011
    return cur
1012

    
1013
def mk_delete(db, table, cond=None):
1014
    '''
1015
    @param cond sql_gen.Code WHERE condition. e.g. use sql_gen.*Cond objects.
1016
    @return str query
1017
    '''
1018
    query = 'DELETE FROM '+table.to_str(db)
1019
    if cond != None: query += '\nWHERE '+cond.to_str(db)
1020
    
1021
    query = with_explain_comment(db, query)
1022
    
1023
    return query
1024

    
1025
def delete(db, table, *args, **kw_args):
1026
    '''For params, see mk_delete() and run_query()'''
1027
    recover = kw_args.pop('recover', None)
1028
    cacheable = kw_args.pop('cacheable', True)
1029
    log_level = kw_args.pop('log_level', 2)
1030
    
1031
    cur = run_query(db, mk_delete(db, table, *args, **kw_args), recover,
1032
        cacheable, log_level=log_level)
1033
    autoanalyze(db, table)
1034
    return cur
1035

    
1036
def last_insert_id(db):
1037
    module = util.root_module(db.db)
1038
    if module == 'psycopg2': return value(run_query(db, 'SELECT lastval()'))
1039
    elif module == 'MySQLdb': return db.insert_id()
1040
    else: return None
1041

    
1042
def define_func(db, def_):
1043
    func = def_.function
1044
    while True:
1045
        try:
1046
            run_query(db, def_.to_str(db), recover=True, cacheable=True,
1047
                log_ignore_excs=(DuplicateException,))
1048
            break # successful
1049
        except DuplicateException:
1050
            func.name = next_version(func.name)
1051
            # try again with next version of name
1052

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

    
1085
def flatten(db, into, joins, cols, limit=None, start=None, **kw_args):
1086
    '''For params, see mk_flatten_mapping()
1087
    @return See return value of mk_flatten_mapping()
1088
    '''
1089
    items = mk_flatten_mapping(db, into, cols, as_items=True, **kw_args)
1090
    cols = [sql_gen.NamedCol(new.name, old) for old, new in items]
1091
    run_query_into(db, mk_select(db, joins, cols, limit=limit, start=start),
1092
        into=into, add_pkey_=True)
1093
        # don't cache because the temp table will usually be truncated after use
1094
    return dict(items)
1095

    
1096
##### Database structure introspection
1097

    
1098
#### Tables
1099

    
1100
def tables(db, schema_like='public', table_like='%', exact=False,
1101
    cacheable=True):
1102
    if exact: compare = '='
1103
    else: compare = 'LIKE'
1104
    
1105
    module = util.root_module(db.db)
1106
    if module == 'psycopg2':
1107
        conds = [('schemaname', sql_gen.CompareCond(schema_like, compare)),
1108
            ('tablename', sql_gen.CompareCond(table_like, compare))]
1109
        return values(select(db, 'pg_tables', ['tablename'], conds,
1110
            order_by='tablename', cacheable=cacheable, log_level=4))
1111
    elif module == 'MySQLdb':
1112
        return values(run_query(db, 'SHOW TABLES LIKE '+db.esc_value(table_like)
1113
            , cacheable=True, log_level=4))
1114
    else: raise NotImplementedError("Can't list tables for "+module+' database')
1115

    
1116
def table_exists(db, table, cacheable=True):
1117
    table = sql_gen.as_Table(table)
1118
    return list(tables(db, table.schema, table.name, True, cacheable)) != []
1119

    
1120
def table_row_count(db, table, recover=None):
1121
    return value(run_query(db, mk_select(db, table, [sql_gen.row_count],
1122
        order_by=None), recover=recover, log_level=3))
1123

    
1124
def table_col_names(db, table, recover=None):
1125
    return list(col_names(select(db, table, limit=0, recover=recover,
1126
        log_level=4)))
1127

    
1128
def table_cols(db, table, *args, **kw_args):
1129
    return [sql_gen.as_Col(strings.ustr(c), table)
1130
        for c in table_col_names(db, table, *args, **kw_args)]
1131

    
1132
def table_pkey_index(db, table, recover=None):
1133
    table_str = sql_gen.Literal(table.to_str(db))
1134
    try:
1135
        return sql_gen.Table(value(run_query(db, '''\
1136
SELECT relname
1137
FROM pg_index
1138
JOIN pg_class index ON index.oid = indexrelid
1139
WHERE
1140
indrelid = '''+table_str.to_str(db)+'''::regclass
1141
AND indisprimary
1142
'''
1143
            , recover, cacheable=True, log_level=4)), table.schema)
1144
    except StopIteration: raise DoesNotExistException('primary key', '')
1145

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

    
1171
def table_has_pkey(db, table, recover=None):
1172
    try: table_pkey_col(db, table, recover)
1173
    except DoesNotExistException: return False
1174
    else: return True
1175

    
1176
def pkey_name(db, table, recover=None):
1177
    '''If no pkey, returns the first column in the table.'''
1178
    return pkey_col(db, table, recover).name
1179

    
1180
def pkey_col(db, table, recover=None):
1181
    '''If no pkey, returns the first column in the table.'''
1182
    try: return table_pkey_col(db, table, recover)
1183
    except DoesNotExistException: return table_cols(db, table, recover)[0]
1184

    
1185
not_null_col = 'not_null_col'
1186

    
1187
def table_not_null_col(db, table, recover=None):
1188
    '''Name assumed to be the value of not_null_col. If not found, uses pkey.'''
1189
    if not_null_col in table_col_names(db, table, recover): return not_null_col
1190
    else: return pkey_name(db, table, recover)
1191

    
1192
def constraint_cond(db, constraint):
1193
    module = util.root_module(db.db)
1194
    if module == 'psycopg2':
1195
        table_str = sql_gen.Literal(constraint.table.to_str(db))
1196
        name_str = sql_gen.Literal(constraint.name)
1197
        return value(run_query(db, '''\
1198
SELECT consrc
1199
FROM pg_constraint
1200
WHERE
1201
conrelid = '''+table_str.to_str(db)+'''::regclass
1202
AND conname = '''+name_str.to_str(db)+'''
1203
'''
1204
            , cacheable=True, log_level=4))
1205
    else: raise NotImplementedError("Can't get constraint condition for "
1206
        +module+' database')
1207

    
1208
def index_exprs(db, index):
1209
    index = sql_gen.as_Table(index)
1210
    module = util.root_module(db.db)
1211
    if module == 'psycopg2':
1212
        qual_index = sql_gen.Literal(index.to_str(db))
1213
        return list(values(run_query(db, '''\
1214
SELECT pg_get_indexdef(indexrelid, generate_series(1, indnatts), true)
1215
FROM pg_index
1216
WHERE indexrelid = '''+qual_index.to_str(db)+'''::regclass
1217
'''
1218
            , cacheable=True, log_level=4)))
1219
    else: raise NotImplementedError()
1220

    
1221
def index_cols(db, index):
1222
    '''Can also use this for UNIQUE constraints, because a UNIQUE index is
1223
    automatically created. When you don't know whether something is a UNIQUE
1224
    constraint or a UNIQUE index, use this function.'''
1225
    return map(sql_gen.parse_expr_col, index_exprs(db, index))
1226

    
1227
def index_cond(db, index):
1228
    index = sql_gen.as_Table(index)
1229
    module = util.root_module(db.db)
1230
    if module == 'psycopg2':
1231
        qual_index = sql_gen.Literal(index.to_str(db))
1232
        return value(run_query(db, '''\
1233
SELECT pg_get_expr(indpred, indrelid, true)
1234
FROM pg_index
1235
WHERE indexrelid = '''+qual_index.to_str(db)+'''::regclass
1236
'''
1237
            , cacheable=True, log_level=4))
1238
    else: raise NotImplementedError()
1239

    
1240
def index_order_by(db, index):
1241
    return sql_gen.CustomCode(', '.join(index_exprs(db, index)))
1242

    
1243
def table_cluster_on(db, table, recover=None):
1244
    '''
1245
    @return The table's cluster index, or its pkey if none is set
1246
    '''
1247
    table_str = sql_gen.Literal(table.to_str(db))
1248
    try:
1249
        return sql_gen.Table(value(run_query(db, '''\
1250
SELECT relname
1251
FROM pg_index
1252
JOIN pg_class index ON index.oid = indexrelid
1253
WHERE
1254
indrelid = '''+table_str.to_str(db)+'''::regclass
1255
AND indisclustered
1256
'''
1257
            , recover, cacheable=True, log_level=4)), table.schema)
1258
    except StopIteration: return table_pkey_index(db, table, recover)
1259

    
1260
def table_order_by(db, table, recover=None):
1261
    if table.order_by == None:
1262
        try: table.order_by = index_order_by(db, table_cluster_on(db, table,
1263
            recover))
1264
        except DoesNotExistException: pass
1265
    return table.order_by
1266

    
1267
#### Functions
1268

    
1269
def function_exists(db, function):
1270
    qual_function = sql_gen.Literal(function.to_str(db))
1271
    try:
1272
        select(db, fields=[sql_gen.Cast('regproc', qual_function)],
1273
            recover=True, cacheable=True, log_level=4)
1274
    except DoesNotExistException: return False
1275
    except DuplicateException: return True # overloaded function
1276
    else: return True
1277

    
1278
def function_param0_type(db, function):
1279
    qual_function = sql_gen.Literal(function.to_str(db))
1280
    return value(run_query(db, '''\
1281
SELECT proargtypes[0]::regtype
1282
FROM pg_proc
1283
WHERE oid = '''+qual_function.to_str(db)+'''::regproc
1284
'''
1285
        , cacheable=True, log_level=4))
1286

    
1287
##### Structural changes
1288

    
1289
#### Columns
1290

    
1291
def add_col(db, table, col, comment=None, if_not_exists=False, **kw_args):
1292
    '''
1293
    @param col TypedCol Name may be versioned, so be sure to propagate any
1294
        renaming back to any source column for the TypedCol.
1295
    @param comment None|str SQL comment used to distinguish columns of the same
1296
        name from each other when they contain different data, to allow the
1297
        ADD COLUMN query to be cached. If not set, query will not be cached.
1298
    '''
1299
    assert isinstance(col, sql_gen.TypedCol)
1300
    
1301
    while True:
1302
        str_ = 'ALTER TABLE '+table.to_str(db)+' ADD COLUMN '+col.to_str(db)
1303
        if comment != None: str_ += ' '+sql_gen.esc_comment(comment)
1304
        
1305
        try:
1306
            run_query(db, str_, recover=True, cacheable=True, **kw_args)
1307
            break
1308
        except DuplicateException:
1309
            if if_not_exists: raise
1310
            col.name = next_version(col.name)
1311
            # try again with next version of name
1312

    
1313
def add_not_null(db, col):
1314
    table = col.table
1315
    col = sql_gen.to_name_only_col(col)
1316
    run_query(db, 'ALTER TABLE '+table.to_str(db)+' ALTER COLUMN '
1317
        +col.to_str(db)+' SET NOT NULL', cacheable=True, log_level=3)
1318

    
1319
def drop_not_null(db, col):
1320
    table = col.table
1321
    col = sql_gen.to_name_only_col(col)
1322
    run_query(db, 'ALTER TABLE '+table.to_str(db)+' ALTER COLUMN '
1323
        +col.to_str(db)+' DROP NOT NULL', cacheable=True, log_level=3)
1324

    
1325
row_num_col = '_row_num'
1326

    
1327
row_num_col_def = sql_gen.TypedCol('', 'serial', nullable=False,
1328
    constraints='PRIMARY KEY')
1329

    
1330
def add_row_num(db, table, name=row_num_col):
1331
    '''Adds a row number column to a table. Its definition is in
1332
    row_num_col_def. It will be the primary key.'''
1333
    col_def = copy.copy(row_num_col_def)
1334
    col_def.name = name
1335
    add_col(db, table, col_def, comment='', if_not_exists=True, log_level=3)
1336

    
1337
#### Indexes
1338

    
1339
def add_pkey(db, table, cols=None, recover=None):
1340
    '''Adds a primary key.
1341
    @param cols [sql_gen.Col,...] The columns in the primary key.
1342
        Defaults to the first column in the table.
1343
    @pre The table must not already have a primary key.
1344
    '''
1345
    table = sql_gen.as_Table(table)
1346
    if cols == None: cols = [pkey_name(db, table, recover)]
1347
    col_strs = [sql_gen.to_name_only_col(v).to_str(db) for v in cols]
1348
    
1349
    run_query(db, 'ALTER TABLE '+table.to_str(db)+' ADD PRIMARY KEY ('
1350
        +(', '.join(col_strs))+')', recover=True, cacheable=True, log_level=3,
1351
        log_ignore_excs=(DuplicateException,))
1352

    
1353
def add_index(db, exprs, table=None, unique=False, ensure_not_null_=True):
1354
    '''Adds an index on column(s) or expression(s) if it doesn't already exist.
1355
    Currently, only function calls and literal values are supported expressions.
1356
    @param ensure_not_null_ If set, translates NULL values to sentinel values.
1357
        This allows indexes to be used for comparisons where NULLs are equal.
1358
    '''
1359
    exprs = lists.mk_seq(exprs)
1360
    
1361
    # Parse exprs
1362
    old_exprs = exprs[:]
1363
    exprs = []
1364
    cols = []
1365
    for i, expr in enumerate(old_exprs):
1366
        expr = sql_gen.as_Col(expr, table)
1367
        
1368
        # Handle nullable columns
1369
        if ensure_not_null_:
1370
            try: expr = sql_gen.ensure_not_null(db, expr)
1371
            except KeyError: pass # unknown type, so just create plain index
1372
        
1373
        # Extract col
1374
        expr = copy.deepcopy(expr) # don't modify input!
1375
        col = expr
1376
        if isinstance(expr, sql_gen.FunctionCall): col = expr.args[0]
1377
        expr = sql_gen.cast_literal(expr)
1378
        if not isinstance(expr, (sql_gen.Expr, sql_gen.Col)):
1379
            expr = sql_gen.Expr(expr)
1380
            
1381
        
1382
        # Extract table
1383
        if table == None:
1384
            assert sql_gen.is_table_col(col)
1385
            table = col.table
1386
        
1387
        if isinstance(col, sql_gen.Col): col.table = None
1388
        
1389
        exprs.append(expr)
1390
        cols.append(col)
1391
    
1392
    table = sql_gen.as_Table(table)
1393
    
1394
    # Add index
1395
    str_ = 'CREATE'
1396
    if unique: str_ += ' UNIQUE'
1397
    str_ += ' INDEX ON '+table.to_str(db)+' ('+(
1398
        ', '.join((v.to_str(db) for v in exprs)))+')'
1399
    run_query(db, str_, recover=True, cacheable=True, log_level=3)
1400

    
1401
def add_pkey_index(db, table): add_index(db, pkey_col(db, table), table)
1402

    
1403
def add_pkey_or_index(db, table, cols=None, recover=None, warn=False):
1404
    try: add_pkey(db, table, cols, recover)
1405
    except DuplicateKeyException, e:
1406
        if warn: warnings.warn(UserWarning(exc.str_(e)))
1407
        add_pkey_index(db, table)
1408

    
1409
already_indexed = object() # tells add_indexes() the pkey has already been added
1410

    
1411
def add_indexes(db, table, has_pkey=True):
1412
    '''Adds an index on all columns in a table.
1413
    @param has_pkey bool|already_indexed Whether a pkey instead of a regular
1414
        index should be added on the first column.
1415
        * If already_indexed, the pkey is assumed to have already been added
1416
    '''
1417
    cols = table_col_names(db, table)
1418
    if has_pkey:
1419
        if has_pkey is not already_indexed: add_pkey(db, table)
1420
        cols = cols[1:]
1421
    for col in cols: add_index(db, col, table)
1422

    
1423
#### Tables
1424

    
1425
### Maintenance
1426

    
1427
def analyze(db, table):
1428
    table = sql_gen.as_Table(table)
1429
    run_query(db, 'ANALYZE '+table.to_str(db), log_level=3)
1430

    
1431
def autoanalyze(db, table):
1432
    if db.autoanalyze: analyze(db, table)
1433

    
1434
def vacuum(db, table):
1435
    table = sql_gen.as_Table(table)
1436
    db.with_autocommit(lambda: run_query(db, 'VACUUM ANALYZE '+table.to_str(db),
1437
        log_level=3))
1438

    
1439
### Lifecycle
1440

    
1441
def drop(db, type_, name):
1442
    name = sql_gen.as_Name(name)
1443
    run_query(db, 'DROP '+type_+' IF EXISTS '+name.to_str(db)+' CASCADE')
1444

    
1445
def drop_table(db, table): drop(db, 'TABLE', table)
1446

    
1447
def create_table(db, table, cols=[], has_pkey=True, col_indexes=True,
1448
    like=None):
1449
    '''Creates a table.
1450
    @param cols [sql_gen.TypedCol,...] The column names and types
1451
    @param has_pkey If set, the first column becomes the primary key.
1452
    @param col_indexes bool|[ref]
1453
        * If True, indexes will be added on all non-pkey columns.
1454
        * If a list reference, [0] will be set to a function to do this.
1455
          This can be used to delay index creation until the table is populated.
1456
    '''
1457
    table = sql_gen.as_Table(table)
1458
    
1459
    if like != None:
1460
        cols = [sql_gen.CustomCode('LIKE '+like.to_str(db)+' INCLUDING ALL')
1461
            ]+cols
1462
        table.order_by = like.order_by
1463
    if has_pkey:
1464
        cols[0] = pkey = copy.copy(cols[0]) # don't modify input!
1465
        pkey.constraints = 'PRIMARY KEY'
1466
    
1467
    temp = table.is_temp and not db.debug_temp
1468
        # temp tables permanent in debug_temp mode
1469
    
1470
    # Create table
1471
    def create():
1472
        str_ = 'CREATE'
1473
        if temp: str_ += ' TEMP'
1474
        str_ += ' TABLE '+table.to_str(db)+' (\n'
1475
        str_ += '\n, '.join(c.to_str(db) for c in cols)
1476
        str_ += '\n);'
1477
        
1478
        opts = dict(recover=True, cacheable=True, log_level=2,
1479
            log_ignore_excs=(DuplicateException,))
1480
        try: run_query(db, str_, **opts)
1481
        except InvalidTypeException: # try again as view
1482
            run_query_into(db, mk_select(db, like, limit=0), into=table, **opts)
1483
    if table.is_temp:
1484
        while True:
1485
            try:
1486
                create()
1487
                break
1488
            except DuplicateException:
1489
                table.name = next_version(table.name)
1490
                # try again with next version of name
1491
    else: create()
1492
    
1493
    # Add indexes
1494
    if has_pkey: has_pkey = already_indexed
1495
    def add_indexes_(): add_indexes(db, table, has_pkey)
1496
    if isinstance(col_indexes, list): col_indexes[0] = add_indexes_ # defer
1497
    elif col_indexes: add_indexes_() # add now
1498

    
1499
def copy_table_struct(db, src, dest):
1500
    '''Creates a structure-only copy of a table. (Does not copy data.)'''
1501
    create_table(db, dest, has_pkey=False, col_indexes=False, like=src)
1502

    
1503
def copy_table(db, src, dest):
1504
    '''Creates a copy of a table, including data'''
1505
    copy_table_struct(db, src, dest)
1506
    insert_select(db, dest, None, mk_select(db, src))
1507

    
1508
### Data
1509

    
1510
def truncate(db, table, schema='public', **kw_args):
1511
    '''For params, see run_query()'''
1512
    table = sql_gen.as_Table(table, schema)
1513
    return run_query(db, 'TRUNCATE '+table.to_str(db)+' CASCADE', **kw_args)
1514

    
1515
def empty_temp(db, tables):
1516
    tables = lists.mk_seq(tables)
1517
    for table in tables: truncate(db, table, log_level=3)
1518

    
1519
def empty_db(db, schema='public', **kw_args):
1520
    '''For kw_args, see tables()'''
1521
    for table in tables(db, schema, **kw_args): truncate(db, table, schema)
1522

    
1523
def distinct_table(db, table, distinct_on, joins=None):
1524
    '''Creates a copy of a temp table which is distinct on the given columns.
1525
    Adds an index on table's distinct_on columns, to facilitate merge joins.
1526
    @param distinct_on If empty, creates a table with one row. This is useful if
1527
        your distinct_on columns are all literal values.
1528
    @param joins The joins to use when creating the new table
1529
    @return The new table.
1530
    '''
1531
    if joins == None: joins = [table]
1532
    
1533
    new_table = sql_gen.suffixed_table(table, '_distinct')
1534
    distinct_on = filter(sql_gen.is_table_col, distinct_on)
1535
    
1536
    copy_table_struct(db, table, new_table)
1537
    
1538
    limit = None
1539
    if distinct_on == []: limit = 1 # one sample row
1540
    else: add_index(db, distinct_on, table) # for join optimization
1541
    
1542
    insert_select(db, new_table, None, mk_select(db, joins,
1543
        [sql_gen.Col(sql_gen.all_cols, table)], distinct_on=distinct_on,
1544
        order_by=None, limit=limit))
1545
    analyze(db, new_table)
1546
    
1547
    return new_table
(33-33/47)