Project

General

Profile

1
# Database access
2

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

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

    
17
##### Exceptions
18

    
19
def get_cur_query(cur, input_query=None, input_params=None):
20
    raw_query = None
21
    if hasattr(cur, 'query'): raw_query = cur.query
22
    elif hasattr(cur, '_last_executed'): raw_query = cur._last_executed
23
    
24
    if raw_query != None: return raw_query
25
    else: return repr(input_query)+' % '+repr(input_params)
26

    
27
def _add_cursor_info(e, *args, **kw_args):
28
    '''For params, see get_cur_query()'''
29
    exc.add_msg(e, 'query: '+str(get_cur_query(*args, **kw_args)))
30

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

    
36
class ExceptionWithName(DbException):
37
    def __init__(self, name, cause=None):
38
        DbException.__init__(self, 'for name: '+str(name), cause)
39
        self.name = name
40

    
41
class ExceptionWithColumns(DbException):
42
    def __init__(self, cols, cause=None):
43
        DbException.__init__(self, 'for columns: '+(', '.join(cols)), cause)
44
        self.cols = cols
45

    
46
class NameException(DbException): pass
47

    
48
class DuplicateKeyException(ExceptionWithColumns): pass
49

    
50
class NullValueException(ExceptionWithColumns): pass
51

    
52
class DuplicateTableException(ExceptionWithName): pass
53

    
54
class DuplicateFunctionException(ExceptionWithName): pass
55

    
56
class EmptyRowException(DbException): pass
57

    
58
##### Warnings
59

    
60
class DbWarning(UserWarning): pass
61

    
62
##### Result retrieval
63

    
64
def col_names(cur): return (col[0] for col in cur.description)
65

    
66
def rows(cur): return iter(lambda: cur.fetchone(), None)
67

    
68
def consume_rows(cur):
69
    '''Used to fetch all rows so result will be cached'''
70
    iters.consume_iter(rows(cur))
71

    
72
def next_row(cur): return rows(cur).next()
73

    
74
def row(cur):
75
    row_ = next_row(cur)
76
    consume_rows(cur)
77
    return row_
78

    
79
def next_value(cur): return next_row(cur)[0]
80

    
81
def value(cur): return row(cur)[0]
82

    
83
def values(cur): return iters.func_iter(lambda: next_value(cur))
84

    
85
def value_or_none(cur):
86
    try: return value(cur)
87
    except StopIteration: return None
88

    
89
##### Input validation
90

    
91
def clean_name(name): return re.sub(r'\W', r'', name).lower()
92

    
93
def check_name(name):
94
    if re.search(r'\W', name) != None: raise NameException('Name "'+name
95
        +'" may contain only alphanumeric characters and _')
96

    
97
def esc_name_by_module(module, name, ignore_case=False):
98
    if module == 'psycopg2':
99
        if ignore_case:
100
            # Don't enclose in quotes because this disables case-insensitivity
101
            check_name(name)
102
            return name
103
        else: quote = '"'
104
    elif module == 'MySQLdb': quote = '`'
105
    else: raise NotImplementedError("Can't escape name for "+module+' database')
106
    return quote + name.replace(quote, '') + quote
107

    
108
def esc_name_by_engine(engine, name, **kw_args):
109
    return esc_name_by_module(db_engines[engine][0], name, **kw_args)
110

    
111
def esc_name(db, name, **kw_args):
112
    return esc_name_by_module(util.root_module(db.db), name, **kw_args)
113

    
114
def qual_name(db, schema, table):
115
    def esc_name_(name): return esc_name(db, name)
116
    table = esc_name_(table)
117
    if schema != None: return esc_name_(schema)+'.'+table
118
    else: return table
119

    
120
##### Database connections
121

    
122
db_config_names = ['engine', 'host', 'user', 'password', 'database', 'schemas']
123

    
124
db_engines = {
125
    'MySQL': ('MySQLdb', {'password': 'passwd', 'database': 'db'}),
126
    'PostgreSQL': ('psycopg2', {}),
127
}
128

    
129
DatabaseErrors_set = set([DbException])
130
DatabaseErrors = tuple(DatabaseErrors_set)
131

    
132
def _add_module(module):
133
    DatabaseErrors_set.add(module.DatabaseError)
134
    global DatabaseErrors
135
    DatabaseErrors = tuple(DatabaseErrors_set)
136

    
137
def db_config_str(db_config):
138
    return db_config['engine']+' database '+db_config['database']
139

    
140
def _query_lookup(query, params): return (query, dicts.make_hashable(params))
141

    
142
log_debug_none = lambda msg: None
143

    
144
class DbConn:
145
    def __init__(self, db_config, serializable=True, autocommit=False,
146
        caching=True, log_debug=log_debug_none):
147
        self.db_config = db_config
148
        self.serializable = serializable
149
        self.autocommit = autocommit
150
        self.caching = caching
151
        self.log_debug = log_debug
152
        self.debug = log_debug != log_debug_none
153
        
154
        self.__db = None
155
        self.query_results = {}
156
        self._savepoint = 0
157
    
158
    def __getattr__(self, name):
159
        if name == '__dict__': raise Exception('getting __dict__')
160
        if name == 'db': return self._db()
161
        else: raise AttributeError()
162
    
163
    def __getstate__(self):
164
        state = copy.copy(self.__dict__) # shallow copy
165
        state['log_debug'] = None # don't pickle the debug callback
166
        state['_DbConn__db'] = None # don't pickle the connection
167
        return state
168
    
169
    def connected(self): return self.__db != None
170
    
171
    def _db(self):
172
        if self.__db == None:
173
            # Process db_config
174
            db_config = self.db_config.copy() # don't modify input!
175
            schemas = db_config.pop('schemas', None)
176
            module_name, mappings = db_engines[db_config.pop('engine')]
177
            module = __import__(module_name)
178
            _add_module(module)
179
            for orig, new in mappings.iteritems():
180
                try: util.rename_key(db_config, orig, new)
181
                except KeyError: pass
182
            
183
            # Connect
184
            self.__db = module.connect(**db_config)
185
            
186
            # Configure connection
187
            if self.serializable and not self.autocommit:
188
                self.db.set_session(isolation_level='SERIALIZABLE')
189
            if schemas != None:
190
                schemas_ = ''.join((esc_name(self, s)+', '
191
                    for s in schemas.split(',')))
192
                run_raw_query(self, "SELECT set_config('search_path', \
193
%s || current_setting('search_path'), false)", [schemas_])
194
        
195
        return self.__db
196
    
197
    class DbCursor(Proxy):
198
        def __init__(self, outer):
199
            Proxy.__init__(self, outer.db.cursor())
200
            self.outer = outer
201
            self.query_results = outer.query_results
202
            self.query_lookup = None
203
            self.result = []
204
        
205
        def execute(self, query, params=None):
206
            self._is_insert = query.upper().find('INSERT') >= 0
207
            self.query_lookup = _query_lookup(query, params)
208
            try:
209
                try:
210
                    return_value = self.inner.execute(query, params)
211
                    self.outer.do_autocommit()
212
                finally: self.query = get_cur_query(self.inner)
213
            except Exception, e:
214
                _add_cursor_info(e, self, query, params)
215
                self.result = e # cache the exception as the result
216
                self._cache_result()
217
                raise
218
            # Fetch all rows so result will be cached
219
            if self.rowcount == 0 and not self._is_insert: consume_rows(self)
220
            return return_value
221
        
222
        def fetchone(self):
223
            row = self.inner.fetchone()
224
            if row != None: self.result.append(row)
225
            # otherwise, fetched all rows
226
            else: self._cache_result()
227
            return row
228
        
229
        def _cache_result(self):
230
            # For inserts, only cache exceptions since inserts are not
231
            # idempotent, but an invalid insert will always be invalid
232
            if self.query_results != None and (not self._is_insert
233
                or isinstance(self.result, Exception)):
234
                
235
                assert self.query_lookup != None
236
                self.query_results[self.query_lookup] = self.CacheCursor(
237
                    util.dict_subset(dicts.AttrsDictView(self),
238
                    ['query', 'result', 'rowcount', 'description']))
239
        
240
        class CacheCursor:
241
            def __init__(self, cached_result): self.__dict__ = cached_result
242
            
243
            def execute(self, *args, **kw_args):
244
                if isinstance(self.result, Exception): raise self.result
245
                # otherwise, result is a rows list
246
                self.iter = iter(self.result)
247
            
248
            def fetchone(self):
249
                try: return self.iter.next()
250
                except StopIteration: return None
251
    
252
    def esc_value(self, value):
253
        module = util.root_module(db.db)
254
        if module == 'psycopg2': return db.db.cursor().mogrify('%s', [value])
255
        elif module == 'MySQLdb':
256
            import _mysql
257
            return _mysql.escape_string(value)
258
        else: raise NotImplementedError("Can't escape value for "+module
259
            +' database')
260
    
261
    def run_query(self, query, params=None, cacheable=False):
262
        '''Translates known DB errors to typed exceptions:
263
        See self.DbCursor.execute().'''
264
        assert query != None
265
        
266
        if not self.caching: cacheable = False
267
        used_cache = False
268
        try:
269
            # Get cursor
270
            if cacheable:
271
                query_lookup = _query_lookup(query, params)
272
                try:
273
                    cur = self.query_results[query_lookup]
274
                    used_cache = True
275
                except KeyError: cur = self.DbCursor(self)
276
            else: cur = self.db.cursor()
277
            
278
            # Run query
279
            cur.execute(query, params)
280
        finally:
281
            if self.debug: # only compute msg if needed
282
                if used_cache: cache_status = 'Cache hit'
283
                elif cacheable: cache_status = 'Cache miss'
284
                else: cache_status = 'Non-cacheable'
285
                self.log_debug(cache_status+': '
286
                    +strings.one_line(str(get_cur_query(cur, query, params))))
287
        
288
        return cur
289
    
290
    def is_cached(self, query, params=None):
291
        return _query_lookup(query, params) in self.query_results
292
    
293
    def with_savepoint(self, func):
294
        savepoint = 'level_'+str(self._savepoint)
295
        self.run_query('SAVEPOINT '+savepoint)
296
        self._savepoint += 1
297
        try: 
298
            try: return_val = func()
299
            finally:
300
                self._savepoint -= 1
301
                assert self._savepoint >= 0
302
        except:
303
            self.run_query('ROLLBACK TO SAVEPOINT '+savepoint)
304
            raise
305
        else:
306
            self.run_query('RELEASE SAVEPOINT '+savepoint)
307
            self.do_autocommit()
308
            return return_val
309
    
310
    def do_autocommit(self):
311
        '''Autocommits if outside savepoint'''
312
        assert self._savepoint >= 0
313
        if self.autocommit and self._savepoint == 0:
314
            self.log_debug('Autocommiting')
315
            self.db.commit()
316

    
317
connect = DbConn
318

    
319
##### Querying
320

    
321
def run_raw_query(db, *args, **kw_args):
322
    '''For params, see DbConn.run_query()'''
323
    return db.run_query(*args, **kw_args)
324

    
325
def mogrify(db, query, params):
326
    module = util.root_module(db.db)
327
    if module == 'psycopg2': return db.db.cursor().mogrify(query, params)
328
    else: raise NotImplementedError("Can't mogrify query for "+module+
329
        ' database')
330

    
331
##### Recoverable querying
332

    
333
def with_savepoint(db, func): return db.with_savepoint(func)
334

    
335
def run_query(db, query, params=None, recover=None, cacheable=False):
336
    if recover == None: recover = False
337
    
338
    try:
339
        def run(): return run_raw_query(db, query, params, cacheable)
340
        if recover and not db.is_cached(query, params):
341
            return with_savepoint(db, run)
342
        else: return run() # don't need savepoint if cached
343
    except Exception, e:
344
        if not recover: raise # need savepoint to run index_cols()
345
        msg = str(e)
346
        match = re.search(r'duplicate key value violates unique constraint '
347
            r'"((_?[^\W_]+)_[^"]+)"', msg)
348
        if match:
349
            constraint, table = match.groups()
350
            try: cols = index_cols(db, table, constraint)
351
            except NotImplementedError: raise e
352
            else: raise DuplicateKeyException(cols, e)
353
        match = re.search(r'null value in column "(\w+)" violates not-null '
354
            'constraint', msg)
355
        if match: raise NullValueException([match.group(1)], e)
356
        match = re.search(r'relation "(\w+)" already exists', msg)
357
        if match: raise DuplicateTableException(match.group(1), e)
358
        match = re.search(r'function "(\w+)" already exists', msg)
359
        if match: raise DuplicateFunctionException(match.group(1), e)
360
        raise # no specific exception raised
361

    
362
##### Basic queries
363

    
364
def next_version(name):
365
    '''Prepends the version # so it won't be removed if the name is truncated'''
366
    version = 1 # first existing name was version 0
367
    match = re.match(r'^v(\d+)_(.*)$', name)
368
    if match:
369
        version = int(match.group(1))+1
370
        name = match.group(2)
371
    return 'v'+str(version)+'_'+name
372

    
373
def run_query_into(db, query, params, into_ref=None, *args, **kw_args):
374
    '''Outputs a query to a temp table.
375
    For params, see run_query().
376
    '''
377
    if into_ref == None: return run_query(db, query, params, *args, **kw_args)
378
    else: # place rows in temp table
379
        check_name(into_ref[0])
380
        kw_args['recover'] = True
381
        while True:
382
            try:
383
                create_query = 'CREATE'
384
                if not db.debug: create_query += ' TEMP'
385
                create_query += ' TABLE '+into_ref[0]+' AS '+query
386
                
387
                return run_query(db, create_query, params, *args, **kw_args)
388
                    # CREATE TABLE AS sets rowcount to # rows in query
389
            except DuplicateTableException, e:
390
                into_ref[0] = next_version(into_ref[0])
391
                # try again with next version of name
392

    
393
order_by_pkey = object() # tells mk_select() to order by the pkey
394

    
395
join_using = object() # tells mk_select() to join the column with USING
396

    
397
filter_out = object() # tells mk_select() to filter out rows that match the join
398

    
399
distinct_on_all = object() # tells mk_select() to SELECT DISTINCT ON all columns
400

    
401
def mk_select(db, tables, fields=None, conds=None, distinct_on=None, limit=None,
402
    start=None, order_by=order_by_pkey, table_is_esc=False):
403
    '''
404
    @param tables The single table to select from, or a list of tables to join
405
        together: [table0, (table1, joins), ...]
406
        
407
        joins has the format: dict(right_col=left_col, ...)
408
        * if left_col is join_using, left_col is set to right_col
409
        * if left_col is filter_out, the tables are LEFT JOINed together and the
410
          query is filtered by `right_col IS NULL` (indicating no match)
411
    @param fields Use None to select all fields in the table
412
    @param distinct_on The columns to SELECT DISTINCT ON, or distinct_on_all to
413
        use all columns
414
    @param table_is_esc Whether the table name has already been escaped
415
    @return tuple(query, params)
416
    '''
417
    def esc_name_(name): return esc_name(db, name)
418
    
419
    if not lists.is_seq(tables): tables = [tables]
420
    tables = list(tables) # don't modify input! (list() copies input)
421
    table0 = tables.pop(0) # first table is separate
422
    
423
    if conds == None: conds = {}
424
    assert limit == None or type(limit) == int
425
    assert start == None or type(start) == int
426
    if order_by == order_by_pkey:
427
        order_by = pkey(db, table0, recover=True, table_is_esc=table_is_esc)
428
    if not table_is_esc: table0 = esc_name_(table0)
429
    
430
    params = []
431
    
432
    def parse_col(field, default_table=None):
433
        '''Parses fields'''
434
        if field == None: field = (field,) # for None values, tuple is optional
435
        is_tuple = isinstance(field, tuple)
436
        if is_tuple and len(field) == 1: # field is literal value
437
            value, = field
438
            sql_ = '%s'
439
            params.append(value)
440
        elif is_tuple and len(field) == 2: # field is col with table
441
            table, col = field
442
            if not table_is_esc: table = esc_name_(table)
443
            sql_ = table+'.'+esc_name_(col)
444
        else:
445
            sql_ = esc_name_(field) # field is col name
446
            if default_table != None: sql_ = default_table+'.'+sql_
447
        return sql_
448
    def cond(entry):
449
        '''Parses conditions'''
450
        col, value = entry
451
        cond_ = parse_col(col)+' '
452
        if value == None: cond_ += 'IS'
453
        else: cond_ += '='
454
        cond_ += ' %s'
455
        return cond_
456
    
457
    query = 'SELECT'
458
    
459
    # DISTINCT ON columns
460
    if distinct_on != None:
461
        query += ' DISTINCT'
462
        if distinct_on != distinct_on_all:
463
            query += ' ON ('+(', '.join(map(parse_col, distinct_on)))+')'
464
    
465
    # Columns
466
    query += ' '
467
    if fields == None: query += '*'
468
    else: query += ', '.join(map(parse_col, fields))
469
    
470
    # Main table
471
    query += ' FROM '+table0
472
    
473
    # Add joins
474
    left_table = table0
475
    for table, joins in tables:
476
        if not table_is_esc: table = esc_name_(table)
477
        
478
        left_join_ref = [False]
479
        
480
        def join(entry):
481
            '''Parses non-USING joins'''
482
            right_col, left_col = entry
483
            
484
            # Parse special values
485
            if left_col == None: left_col = (left_col,)
486
                # for None values, tuple is optional
487
            elif left_col == join_using: left_col = right_col
488
            elif left_col == filter_out:
489
                left_col = right_col
490
                left_join_ref[0] = True
491
                conds[(table, right_col)] = None # filter query by no match
492
            
493
            # Create SQL
494
            right_col = table+'.'+esc_name_(right_col)
495
            sql_ = right_col+' '
496
            if isinstance(left_col, tuple) and len(left_col) == 1:
497
                # col is literal value
498
                value, = left_col
499
                if value == None: sql_ += 'IS'
500
                else: sql_ += '='
501
                sql_ += ' %s'
502
                params.append(value)
503
            else: # col is name
504
                left_col = parse_col(left_col, left_table)
505
                sql_ += ('= '+left_col+' OR ('+right_col+' IS NULL AND '
506
                    +left_col+' IS NULL)')
507
            
508
            return sql_
509
        
510
        # Create join condition and determine join type
511
        if reduce(operator.and_, (v == join_using for v in joins.itervalues())):
512
            # all cols w/ USING, so can use simpler USING syntax
513
            join_cond = 'USING ('+(', '.join(joins.iterkeys()))+')'
514
        else: join_cond = 'ON '+(' AND '.join(map(join, joins.iteritems())))
515
        
516
        # Create join
517
        if left_join_ref[0]: query += ' LEFT'
518
        query += ' JOIN '+table+' '+join_cond
519
        
520
        left_table = table
521
    
522
    missing = True
523
    if conds != {}:
524
        query += ' WHERE '+(' AND '.join(map(cond, conds.iteritems())))
525
        params += conds.values()
526
        missing = False
527
    if order_by != None: query += ' ORDER BY '+parse_col(order_by, table0)
528
    if limit != None: query += ' LIMIT '+str(limit); missing = False
529
    if start != None:
530
        if start != 0: query += ' OFFSET '+str(start)
531
        missing = False
532
    if missing: warnings.warn(DbWarning(
533
        'SELECT statement missing a WHERE, LIMIT, or OFFSET clause: '+query))
534
    
535
    return (query, params)
536

    
537
def select(db, *args, **kw_args):
538
    '''For params, see mk_select() and run_query()'''
539
    recover = kw_args.pop('recover', None)
540
    cacheable = kw_args.pop('cacheable', True)
541
    
542
    query, params = mk_select(db, *args, **kw_args)
543
    return run_query(db, query, params, recover, cacheable)
544

    
545
def mk_insert_select(db, table, cols=None, select_query=None, params=None,
546
    returning=None, embeddable=False, table_is_esc=False):
547
    '''
548
    @param returning str|None An inserted column (such as pkey) to return
549
    @param embeddable Whether the query should be embeddable as a nested SELECT.
550
        Warning: If you set this and cacheable=True when the query is run, the
551
        query will be fully cached, not just if it raises an exception.
552
    @param table_is_esc Whether the table name has already been escaped
553
    '''
554
    if select_query == None: select_query = 'DEFAULT VALUES'
555
    if cols == []: cols = None # no cols (all defaults) = unknown col names
556
    if not table_is_esc: check_name(table)
557
    
558
    # Build query
559
    query = 'INSERT INTO '+table
560
    if cols != None:
561
        map(check_name, cols)
562
        query += ' ('+', '.join(cols)+')'
563
    query += ' '+select_query
564
    
565
    if returning != None:
566
        check_name(returning)
567
        query += ' RETURNING '+returning
568
    
569
    if embeddable:
570
        # Create function
571
        function_name = '_'.join(map(clean_name, ['insert', table] + cols))
572
        return_type = 'SETOF '+table+'.'+returning+'%TYPE'
573
        while True:
574
            try:
575
                function = function_name
576
                if not db.debug: function = 'pg_temp.'+function
577
                
578
                function_query = '''\
579
CREATE FUNCTION '''+function+'''() RETURNS '''+return_type+'''
580
    LANGUAGE sql
581
    AS $$'''+mogrify(db, query, params)+''';$$;
582
'''
583
                run_query(db, function_query, recover=True, cacheable=True)
584
                break # this version was successful
585
            except DuplicateFunctionException, e:
586
                function_name = next_version(function_name)
587
                # try again with next version of name
588
        
589
        # Return query that uses function
590
        return mk_select(db, function+'() AS f ('+returning+')', start=0,
591
            order_by=None, table_is_esc=True)# AS clause requires function alias
592
    
593
    return (query, params)
594

    
595
def insert_select(db, *args, **kw_args):
596
    '''For params, see mk_insert_select() and run_query_into()
597
    @param into_ref List with name of temp table to place RETURNING values in
598
    '''
599
    into_ref = kw_args.pop('into_ref', None)
600
    if into_ref != None: kw_args['embeddable'] = True
601
    recover = kw_args.pop('recover', None)
602
    cacheable = kw_args.pop('cacheable', True)
603
    
604
    query, params = mk_insert_select(db, *args, **kw_args)
605
    return run_query_into(db, query, params, into_ref, recover=recover,
606
        cacheable=cacheable)
607

    
608
default = object() # tells insert() to use the default value for a column
609

    
610
def insert(db, table, row, *args, **kw_args):
611
    '''For params, see insert_select()'''
612
    if lists.is_seq(row): cols = None
613
    else:
614
        cols = row.keys()
615
        row = row.values()
616
    row = list(row) # ensure that "!= []" works
617
    
618
    # Check for special values
619
    labels = []
620
    values = []
621
    for value in row:
622
        if value == default: labels.append('DEFAULT')
623
        else:
624
            labels.append('%s')
625
            values.append(value)
626
    
627
    # Build query
628
    if values != []: query = ' VALUES ('+(', '.join(labels))+')'
629
    else: query = None
630
    
631
    return insert_select(db, table, cols, query, values, *args, **kw_args)
632

    
633
def last_insert_id(db):
634
    module = util.root_module(db.db)
635
    if module == 'psycopg2': return value(run_query(db, 'SELECT lastval()'))
636
    elif module == 'MySQLdb': return db.insert_id()
637
    else: return None
638

    
639
def truncate(db, table, schema='public'):
640
    return run_query(db, 'TRUNCATE '+qual_name(db, schema, table)+' CASCADE')
641

    
642
##### Database structure queries
643

    
644
def pkey(db, table, recover=None, table_is_esc=False):
645
    '''Assumed to be first column in table'''
646
    return col_names(select(db, table, limit=0, order_by=None, recover=recover,
647
        table_is_esc=table_is_esc)).next()
648

    
649
def index_cols(db, table, index):
650
    '''Can also use this for UNIQUE constraints, because a UNIQUE index is
651
    automatically created. When you don't know whether something is a UNIQUE
652
    constraint or a UNIQUE index, use this function.'''
653
    check_name(table)
654
    check_name(index)
655
    module = util.root_module(db.db)
656
    if module == 'psycopg2':
657
        return list(values(run_query(db, '''\
658
SELECT attname
659
FROM
660
(
661
        SELECT attnum, attname
662
        FROM pg_index
663
        JOIN pg_class index ON index.oid = indexrelid
664
        JOIN pg_class table_ ON table_.oid = indrelid
665
        JOIN pg_attribute ON attrelid = indrelid AND attnum = ANY (indkey)
666
        WHERE
667
            table_.relname = %(table)s
668
            AND index.relname = %(index)s
669
    UNION
670
        SELECT attnum, attname
671
        FROM
672
        (
673
            SELECT
674
                indrelid
675
                , (regexp_matches(indexprs, E':varattno (\\\\d+)', 'g'))[1]::int
676
                    AS indkey
677
            FROM pg_index
678
            JOIN pg_class index ON index.oid = indexrelid
679
            JOIN pg_class table_ ON table_.oid = indrelid
680
            WHERE
681
                table_.relname = %(table)s
682
                AND index.relname = %(index)s
683
        ) s
684
        JOIN pg_attribute ON attrelid = indrelid AND attnum = indkey
685
) s
686
ORDER BY attnum
687
''',
688
            {'table': table, 'index': index}, cacheable=True)))
689
    else: raise NotImplementedError("Can't list index columns for "+module+
690
        ' database')
691

    
692
def constraint_cols(db, table, constraint):
693
    check_name(table)
694
    check_name(constraint)
695
    module = util.root_module(db.db)
696
    if module == 'psycopg2':
697
        return list(values(run_query(db, '''\
698
SELECT attname
699
FROM pg_constraint
700
JOIN pg_class ON pg_class.oid = conrelid
701
JOIN pg_attribute ON attrelid = conrelid AND attnum = ANY (conkey)
702
WHERE
703
    relname = %(table)s
704
    AND conname = %(constraint)s
705
ORDER BY attnum
706
''',
707
            {'table': table, 'constraint': constraint})))
708
    else: raise NotImplementedError("Can't list constraint columns for "+module+
709
        ' database')
710

    
711
row_num_col = '_row_num'
712

    
713
def add_row_num(db, table):
714
    '''Adds a row number column to a table. Its name is in row_num_col. It will
715
    be the primary key.'''
716
    check_name(table)
717
    run_query(db, 'ALTER TABLE '+table+' ADD COLUMN '+row_num_col
718
        +' serial NOT NULL PRIMARY KEY')
719

    
720
def tables(db, schema='public', table_like='%'):
721
    module = util.root_module(db.db)
722
    params = {'schema': schema, 'table_like': table_like}
723
    if module == 'psycopg2':
724
        return values(run_query(db, '''\
725
SELECT tablename
726
FROM pg_tables
727
WHERE
728
    schemaname = %(schema)s
729
    AND tablename LIKE %(table_like)s
730
ORDER BY tablename
731
''',
732
            params, cacheable=True))
733
    elif module == 'MySQLdb':
734
        return values(run_query(db, 'SHOW TABLES LIKE %(table_like)s', params,
735
            cacheable=True))
736
    else: raise NotImplementedError("Can't list tables for "+module+' database')
737

    
738
##### Database management
739

    
740
def empty_db(db, schema='public', **kw_args):
741
    '''For kw_args, see tables()'''
742
    for table in tables(db, schema, **kw_args): truncate(db, table, schema)
743

    
744
##### Heuristic queries
745

    
746
def put(db, table, row, pkey_=None, row_ct_ref=None):
747
    '''Recovers from errors.
748
    Only works under PostgreSQL (uses INSERT RETURNING).
749
    '''
750
    if pkey_ == None: pkey_ = pkey(db, table, recover=True)
751
    
752
    try:
753
        cur = insert(db, table, row, pkey_, recover=True)
754
        if row_ct_ref != None and cur.rowcount >= 0:
755
            row_ct_ref[0] += cur.rowcount
756
        return value(cur)
757
    except DuplicateKeyException, e:
758
        return value(select(db, table, [pkey_],
759
            util.dict_subset_right_join(row, e.cols), recover=True))
760

    
761
def get(db, table, row, pkey, row_ct_ref=None, create=False):
762
    '''Recovers from errors'''
763
    try: return value(select(db, table, [pkey], row, limit=1, recover=True))
764
    except StopIteration:
765
        if not create: raise
766
        return put(db, table, row, pkey, row_ct_ref) # insert new row
767

    
768
def put_table(db, out_table, in_tables, mapping, limit=None, start=0,
769
    row_ct_ref=None, table_is_esc=False):
770
    '''Recovers from errors.
771
    Only works under PostgreSQL (uses INSERT RETURNING).
772
    @param in_tables The main input table to select from, followed by a list of
773
        tables to join with it using the main input table's pkey
774
    @return (table, col) Where the pkeys (from INSERT RETURNING) are made
775
        available
776
    '''
777
    temp_suffix = clean_name(out_table)
778
        # suffix, not prefix, so main name won't be removed if name is truncated
779
    pkeys_ref = ['pkeys_'+temp_suffix]
780
    
781
    # Join together input tables
782
    in_tables = in_tables[:] # don't modify input!
783
    in_tables0 = in_tables.pop(0) # first table is separate
784
    in_pkey = pkey(db, in_tables0, recover=True, table_is_esc=table_is_esc)
785
    insert_joins = [in_tables0]+[(t, {in_pkey: join_using}) for t in in_tables]
786
    
787
    out_pkey = pkey(db, out_table, recover=True, table_is_esc=table_is_esc)
788
    pkeys_cols = [in_pkey, out_pkey]
789
    
790
    pkeys_table_exists_ref = [False]
791
    def run_query_into_pkeys(query, params):
792
        if pkeys_table_exists_ref[0]:
793
            insert_select(db, pkeys_ref[0], pkeys_cols, query, params)
794
        else:
795
            run_query_into(db, query, params, into_ref=pkeys_ref)
796
            pkeys_table_exists_ref[0] = True
797
    
798
    conds = {}
799
    distinct_on = None
800
    def mk_main_select(cols):
801
        return mk_select(db, insert_joins, cols, conds, distinct_on,
802
            order_by=None, limit=limit, start=start, table_is_esc=table_is_esc)
803
    
804
    # Do inserts and selects
805
    out_pkeys_ref = ['out_pkeys_'+temp_suffix]
806
    while True:
807
        try:
808
            cur = insert_select(db, out_table, mapping.keys(),
809
                *mk_main_select(mapping.values()), returning=out_pkey,
810
                into_ref=out_pkeys_ref, recover=True, table_is_esc=table_is_esc)
811
            if row_ct_ref != None and cur.rowcount >= 0:
812
                row_ct_ref[0] += cur.rowcount
813
                add_row_num(db, out_pkeys_ref[0]) # for joining with input pkeys
814
            
815
            # Get input pkeys corresponding to rows in insert
816
            in_pkeys_ref = ['in_pkeys_'+temp_suffix]
817
            run_query_into(db, *mk_main_select([in_pkey]),
818
                into_ref=in_pkeys_ref)
819
            add_row_num(db, in_pkeys_ref[0]) # for joining with output pkeys
820
            
821
            # Join together output and input pkeys
822
            run_query_into_pkeys(*mk_select(db, [in_pkeys_ref[0],
823
                (out_pkeys_ref[0], {row_num_col: join_using})], pkeys_cols,
824
                start=0))
825
            
826
            break # insert successful
827
        except DuplicateKeyException, e:
828
            join_cols = util.dict_subset_right_join(mapping, e.cols)
829
            select_joins = insert_joins + [(out_table, join_cols)]
830
            
831
            # Get pkeys of already existing rows
832
            run_query_into_pkeys(*mk_select(db, select_joins, pkeys_cols,
833
                order_by=None, start=0, table_is_esc=table_is_esc))
834
            
835
            # Save existing pkeys in temp table for joining on
836
            existing_pkeys_ref = ['existing_pkeys_'+temp_suffix]
837
            run_query_into(db, *mk_select(db, pkeys_ref[0], [in_pkey],
838
                order_by=None, start=0, table_is_esc=True),
839
                into_ref=existing_pkeys_ref)
840
                # need table_is_esc=True to make table name case-insensitive
841
            
842
            # rerun loop with additional constraints
843
            break # but until NullValueExceptions are handled, end loop here
844
    
845
    return (pkeys_ref[0], out_pkey)
846

    
847
##### Data cleanup
848

    
849
def cleanup_table(db, table, cols, table_is_esc=False):
850
    def esc_name_(name): return esc_name(db, name)
851
    
852
    if not table_is_esc: check_name(table)
853
    cols = map(esc_name_, cols)
854
    
855
    run_query(db, 'UPDATE '+table+' SET\n'+(',\n'.join(('\n'+col
856
        +' = nullif(nullif(trim(both from '+col+"), %(null0)s), %(null1)s)"
857
            for col in cols))),
858
        dict(null0='', null1=r'\N'))
(22-22/34)