Project

General

Profile

1 11 aaronmk
# Database access
2
3 1869 aaronmk
import copy
4 11 aaronmk
import re
5 865 aaronmk
import warnings
6 11 aaronmk
7 300 aaronmk
import exc
8 1909 aaronmk
import dicts
9 1893 aaronmk
import iters
10 1960 aaronmk
import lists
11 1889 aaronmk
from Proxy import Proxy
12 1872 aaronmk
import rand
13 862 aaronmk
import strings
14 131 aaronmk
import util
15 11 aaronmk
16 832 aaronmk
##### Exceptions
17
18 135 aaronmk
def get_cur_query(cur):
19
    if hasattr(cur, 'query'): return cur.query
20
    elif hasattr(cur, '_last_executed'): return cur._last_executed
21
    else: return None
22 14 aaronmk
23 300 aaronmk
def _add_cursor_info(e, cur): exc.add_msg(e, 'query: '+get_cur_query(cur))
24 135 aaronmk
25 300 aaronmk
class DbException(exc.ExceptionWithCause):
26 14 aaronmk
    def __init__(self, msg, cause=None, cur=None):
27 300 aaronmk
        exc.ExceptionWithCause.__init__(self, msg, cause)
28 14 aaronmk
        if cur != None: _add_cursor_info(self, cur)
29
30 360 aaronmk
class NameException(DbException): pass
31
32 468 aaronmk
class ExceptionWithColumns(DbException):
33
    def __init__(self, cols, cause=None):
34
        DbException.__init__(self, 'columns: ' + ', '.join(cols), cause)
35
        self.cols = cols
36 11 aaronmk
37 468 aaronmk
class DuplicateKeyException(ExceptionWithColumns): pass
38 13 aaronmk
39 468 aaronmk
class NullValueException(ExceptionWithColumns): pass
40 13 aaronmk
41 89 aaronmk
class EmptyRowException(DbException): pass
42
43 865 aaronmk
##### Warnings
44
45
class DbWarning(UserWarning): pass
46
47 1930 aaronmk
##### Result retrieval
48
49
def col_names(cur): return (col[0] for col in cur.description)
50
51
def rows(cur): return iter(lambda: cur.fetchone(), None)
52
53
def consume_rows(cur):
54
    '''Used to fetch all rows so result will be cached'''
55
    iters.consume_iter(rows(cur))
56
57
def next_row(cur): return rows(cur).next()
58
59
def row(cur):
60
    row_ = next_row(cur)
61
    consume_rows(cur)
62
    return row_
63
64
def next_value(cur): return next_row(cur)[0]
65
66
def value(cur): return row(cur)[0]
67
68
def values(cur): return iters.func_iter(lambda: next_value(cur))
69
70
def value_or_none(cur):
71
    try: return value(cur)
72
    except StopIteration: return None
73
74 1869 aaronmk
##### Database connections
75 1849 aaronmk
76 1926 aaronmk
db_config_names = ['engine', 'host', 'user', 'password', 'database']
77
78 1869 aaronmk
db_engines = {
79
    'MySQL': ('MySQLdb', {'password': 'passwd', 'database': 'db'}),
80
    'PostgreSQL': ('psycopg2', {}),
81
}
82
83
DatabaseErrors_set = set([DbException])
84
DatabaseErrors = tuple(DatabaseErrors_set)
85
86
def _add_module(module):
87
    DatabaseErrors_set.add(module.DatabaseError)
88
    global DatabaseErrors
89
    DatabaseErrors = tuple(DatabaseErrors_set)
90
91
def db_config_str(db_config):
92
    return db_config['engine']+' database '+db_config['database']
93
94 1909 aaronmk
def _query_lookup(query, params): return (query, dicts.make_hashable(params))
95 1894 aaronmk
96 1901 aaronmk
log_debug_none = lambda msg: None
97
98 1849 aaronmk
class DbConn:
99 2047 aaronmk
    def __init__(self, db_config, serializable=True, log_debug=log_debug_none,
100 2050 aaronmk
        caching=True):
101 1869 aaronmk
        self.db_config = db_config
102
        self.serializable = serializable
103 1901 aaronmk
        self.log_debug = log_debug
104 2047 aaronmk
        self.caching = caching
105 1869 aaronmk
106
        self.__db = None
107 1889 aaronmk
        self.query_results = {}
108 1869 aaronmk
109
    def __getattr__(self, name):
110
        if name == '__dict__': raise Exception('getting __dict__')
111
        if name == 'db': return self._db()
112
        else: raise AttributeError()
113
114
    def __getstate__(self):
115
        state = copy.copy(self.__dict__) # shallow copy
116 1915 aaronmk
        state['log_debug'] = None # don't pickle the debug callback
117 1869 aaronmk
        state['_DbConn__db'] = None # don't pickle the connection
118
        return state
119
120
    def _db(self):
121
        if self.__db == None:
122
            # Process db_config
123
            db_config = self.db_config.copy() # don't modify input!
124
            module_name, mappings = db_engines[db_config.pop('engine')]
125
            module = __import__(module_name)
126
            _add_module(module)
127
            for orig, new in mappings.iteritems():
128
                try: util.rename_key(db_config, orig, new)
129
                except KeyError: pass
130
131
            # Connect
132
            self.__db = module.connect(**db_config)
133
134
            # Configure connection
135
            if self.serializable: run_raw_query(self,
136
                'SET TRANSACTION ISOLATION LEVEL SERIALIZABLE')
137
138
        return self.__db
139 1889 aaronmk
140 1891 aaronmk
    class DbCursor(Proxy):
141 1927 aaronmk
        def __init__(self, outer):
142 1891 aaronmk
            Proxy.__init__(self, outer.db.cursor())
143 1927 aaronmk
            self.query_results = outer.query_results
144 1894 aaronmk
            self.query_lookup = None
145 1891 aaronmk
            self.result = []
146 1889 aaronmk
147 1894 aaronmk
        def execute(self, query, params=None):
148 1930 aaronmk
            self._is_insert = query.upper().find('INSERT') >= 0
149 1894 aaronmk
            self.query_lookup = _query_lookup(query, params)
150 1904 aaronmk
            try: return_value = self.inner.execute(query, params)
151
            except Exception, e:
152
                self.result = e # cache the exception as the result
153
                self._cache_result()
154
                raise
155
            finally: self.query = get_cur_query(self.inner)
156 1930 aaronmk
            # Fetch all rows so result will be cached
157
            if self.rowcount == 0 and not self._is_insert: consume_rows(self)
158 1894 aaronmk
            return return_value
159
160 1891 aaronmk
        def fetchone(self):
161
            row = self.inner.fetchone()
162 1899 aaronmk
            if row != None: self.result.append(row)
163
            # otherwise, fetched all rows
164 1904 aaronmk
            else: self._cache_result()
165
            return row
166
167
        def _cache_result(self):
168 1906 aaronmk
            # For inserts, only cache exceptions since inserts are not
169
            # idempotent, but an invalid insert will always be invalid
170 1930 aaronmk
            if self.query_results != None and (not self._is_insert
171 1906 aaronmk
                or isinstance(self.result, Exception)):
172
173 1894 aaronmk
                assert self.query_lookup != None
174 1916 aaronmk
                self.query_results[self.query_lookup] = self.CacheCursor(
175
                    util.dict_subset(dicts.AttrsDictView(self),
176
                    ['query', 'result', 'rowcount', 'description']))
177 1906 aaronmk
178 1916 aaronmk
        class CacheCursor:
179
            def __init__(self, cached_result): self.__dict__ = cached_result
180
181 1927 aaronmk
            def execute(self, *args, **kw_args):
182 1916 aaronmk
                if isinstance(self.result, Exception): raise self.result
183
                # otherwise, result is a rows list
184
                self.iter = iter(self.result)
185
186
            def fetchone(self):
187
                try: return self.iter.next()
188
                except StopIteration: return None
189 1891 aaronmk
190 1894 aaronmk
    def run_query(self, query, params=None, cacheable=False):
191 2047 aaronmk
        if not self.caching: cacheable = False
192 1903 aaronmk
        used_cache = False
193
        try:
194 1927 aaronmk
            # Get cursor
195
            if cacheable:
196
                query_lookup = _query_lookup(query, params)
197
                try:
198
                    cur = self.query_results[query_lookup]
199
                    used_cache = True
200
                except KeyError: cur = self.DbCursor(self)
201
            else: cur = self.db.cursor()
202
203
            # Run query
204
            try: cur.execute(query, params)
205
            except Exception, e:
206
                _add_cursor_info(e, cur)
207
                raise
208 1903 aaronmk
        finally:
209
            if self.log_debug != log_debug_none: # only compute msg if needed
210
                if used_cache: cache_status = 'Cache hit'
211
                elif cacheable: cache_status = 'Cache miss'
212
                else: cache_status = 'Non-cacheable'
213 1927 aaronmk
                self.log_debug(cache_status+': '
214
                    +strings.one_line(get_cur_query(cur)))
215 1903 aaronmk
216
        return cur
217 1914 aaronmk
218
    def is_cached(self, query, params=None):
219
        return _query_lookup(query, params) in self.query_results
220 1849 aaronmk
221 1869 aaronmk
connect = DbConn
222
223 1919 aaronmk
##### Input validation
224
225
def check_name(name):
226
    if re.search(r'\W', name) != None: raise NameException('Name "'+name
227
        +'" may contain only alphanumeric characters and _')
228
229 2061 aaronmk
def esc_name_by_module(module, name, ignore_case=False):
230 1919 aaronmk
    if module == 'psycopg2':
231 2061 aaronmk
        if ignore_case:
232
            # Don't enclose in quotes because this disables case-insensitivity
233 2057 aaronmk
            check_name(name)
234
            return name
235 2061 aaronmk
        else: quote = '"'
236 1919 aaronmk
    elif module == 'MySQLdb': quote = '`'
237
    else: raise NotImplementedError("Can't escape name for "+module+' database')
238
    return quote + name.replace(quote, '') + quote
239
240
def esc_name_by_engine(engine, name, **kw_args):
241
    return esc_name_by_module(db_engines[engine][0], name, **kw_args)
242
243
def esc_name(db, name, **kw_args):
244
    return esc_name_by_module(util.root_module(db.db), name, **kw_args)
245
246 1968 aaronmk
def qual_name(db, schema, table):
247 2060 aaronmk
    def esc_name_(name): return esc_name(db, name)
248 2051 aaronmk
    table = esc_name_(table)
249
    if schema != None: return esc_name_(schema)+'.'+table
250
    else: return table
251 1968 aaronmk
252 832 aaronmk
##### Querying
253
254 1894 aaronmk
def run_raw_query(db, *args, **kw_args):
255
    '''For args, see DbConn.run_query()'''
256
    return db.run_query(*args, **kw_args)
257 11 aaronmk
258 832 aaronmk
##### Recoverable querying
259 15 aaronmk
260 11 aaronmk
def with_savepoint(db, func):
261 1872 aaronmk
    savepoint = 'savepoint_'+str(rand.rand_int()) # must be unique
262 830 aaronmk
    run_raw_query(db, 'SAVEPOINT '+savepoint)
263 11 aaronmk
    try: return_val = func()
264
    except:
265 830 aaronmk
        run_raw_query(db, 'ROLLBACK TO SAVEPOINT '+savepoint)
266 11 aaronmk
        raise
267
    else:
268 830 aaronmk
        run_raw_query(db, 'RELEASE SAVEPOINT '+savepoint)
269 11 aaronmk
        return return_val
270
271 1894 aaronmk
def run_query(db, query, params=None, recover=None, cacheable=False):
272 830 aaronmk
    if recover == None: recover = False
273
274 1894 aaronmk
    def run(): return run_raw_query(db, query, params, cacheable)
275 1914 aaronmk
    if recover and not db.is_cached(query, params):
276
        return with_savepoint(db, run)
277
    else: return run() # don't need savepoint if cached
278 830 aaronmk
279 832 aaronmk
##### Basic queries
280
281 2054 aaronmk
def mk_select(db, table, fields=None, conds=None, limit=None, start=None,
282
    table_is_esc=False):
283 1981 aaronmk
    '''
284
    @param fields Use None to select all fields in the table
285
    @param table_is_esc Whether the table name has already been escaped
286 2054 aaronmk
    @return tuple(query, params)
287 1981 aaronmk
    '''
288 2060 aaronmk
    def esc_name_(name): return esc_name(db, name)
289 2058 aaronmk
290 1135 aaronmk
    if conds == None: conds = {}
291 135 aaronmk
    assert limit == None or type(limit) == int
292 865 aaronmk
    assert start == None or type(start) == int
293 2058 aaronmk
    if not table_is_esc: table = esc_name_(table)
294 865 aaronmk
295 2056 aaronmk
    params = []
296
297
    def parse_col(field):
298
        '''Parses fields'''
299
        if isinstance(field, tuple): # field is literal values
300
            value, col = field
301
            sql_ = '%s'
302
            params.append(value)
303 2058 aaronmk
            if col != None: sql_ += ' AS '+esc_name_(col)
304
        else: sql_ = esc_name_(field) # field is col name
305 2056 aaronmk
        return sql_
306 11 aaronmk
    def cond(entry):
307 2056 aaronmk
        '''Parses conditions'''
308 13 aaronmk
        col, value = entry
309 2058 aaronmk
        cond_ = esc_name_(col)+' '
310 11 aaronmk
        if value == None: cond_ += 'IS'
311
        else: cond_ += '='
312
        cond_ += ' %s'
313
        return cond_
314 2056 aaronmk
315 1135 aaronmk
    query = 'SELECT '
316
    if fields == None: query += '*'
317 2056 aaronmk
    else: query += ', '.join(map(parse_col, fields))
318 2055 aaronmk
    query += ' FROM '+table
319 865 aaronmk
320
    missing = True
321 89 aaronmk
    if conds != {}:
322
        query += ' WHERE '+' AND '.join(map(cond, conds.iteritems()))
323 2056 aaronmk
        params += conds.values()
324 865 aaronmk
        missing = False
325
    if limit != None: query += ' LIMIT '+str(limit); missing = False
326
    if start != None:
327
        if start != 0: query += ' OFFSET '+str(start)
328
        missing = False
329
    if missing: warnings.warn(DbWarning(
330
        'SELECT statement missing a WHERE, LIMIT, or OFFSET clause: '+query))
331
332 2056 aaronmk
    return (query, params)
333 11 aaronmk
334 2054 aaronmk
def select(db, *args, **kw_args):
335
    '''For params, see mk_select() and run_query()'''
336
    recover = kw_args.pop('recover', None)
337
    cacheable = kw_args.pop('cacheable', True)
338
339
    query, params = mk_select(db, *args, **kw_args)
340
    return run_query(db, query, params, recover, cacheable)
341
342 1961 aaronmk
default = object() # tells insert() to use the default value for a column
343
344 2063 aaronmk
def insert_select(db, table, select_query=None, cols=None, params=None,
345
    returning=None, recover=None, cacheable=True, table_is_esc=False):
346 1960 aaronmk
    '''
347
    @param returning str|None An inserted column (such as pkey) to return
348
    @param table_is_esc Whether the table name has already been escaped
349
    '''
350 2063 aaronmk
    if select_query == None: select_query = 'DEFAULT VALUES'
351
    if cols == []: cols = None # no cols (all defaults) = unknown col names
352 1960 aaronmk
    if not table_is_esc: check_name(table)
353 2063 aaronmk
354
    # Build query
355
    query = 'INSERT INTO '+table
356
    if cols != None:
357
        map(check_name, cols)
358
        query += ' ('+', '.join(cols)+')'
359
    query += ' '+select_query
360
361
    if returning != None:
362
        check_name(returning)
363
        query += ' RETURNING '+returning
364
365
    return run_query(db, query, params, recover, cacheable)
366
367
def insert(db, table, row, *args, **kw_args):
368
    '''For args, see insert_select()'''
369 1960 aaronmk
    if lists.is_seq(row): cols = None
370
    else:
371
        cols = row.keys()
372
        row = row.values()
373
    row = list(row) # ensure that "!= []" works
374
375 1961 aaronmk
    # Check for special values
376
    labels = []
377
    values = []
378
    for value in row:
379
        if value == default: labels.append('DEFAULT')
380
        else:
381
            labels.append('%s')
382
            values.append(value)
383
384
    # Build query
385 2063 aaronmk
    if values != []: query = ' VALUES ('+(', '.join(labels))+')'
386
    else: query = None
387 1554 aaronmk
388 2063 aaronmk
    return insert_select(db, table, query, cols, values, *args, **kw_args)
389 11 aaronmk
390 135 aaronmk
def last_insert_id(db):
391 1849 aaronmk
    module = util.root_module(db.db)
392 135 aaronmk
    if module == 'psycopg2': return value(run_query(db, 'SELECT lastval()'))
393
    elif module == 'MySQLdb': return db.insert_id()
394
    else: return None
395 13 aaronmk
396 1968 aaronmk
def truncate(db, table, schema='public'):
397
    return run_query(db, 'TRUNCATE '+qual_name(db, schema, table)+' CASCADE')
398 832 aaronmk
399
##### Database structure queries
400
401 1850 aaronmk
def pkey(db, table, recover=None):
402 832 aaronmk
    '''Assumed to be first column in table'''
403
    check_name(table)
404 1929 aaronmk
    return col_names(select(db, table, limit=0, recover=recover)).next()
405 832 aaronmk
406 853 aaronmk
def index_cols(db, table, index):
407
    '''Can also use this for UNIQUE constraints, because a UNIQUE index is
408
    automatically created. When you don't know whether something is a UNIQUE
409
    constraint or a UNIQUE index, use this function.'''
410
    check_name(table)
411
    check_name(index)
412 1909 aaronmk
    module = util.root_module(db.db)
413
    if module == 'psycopg2':
414
        return list(values(run_query(db, '''\
415 853 aaronmk
SELECT attname
416 866 aaronmk
FROM
417
(
418
        SELECT attnum, attname
419
        FROM pg_index
420
        JOIN pg_class index ON index.oid = indexrelid
421
        JOIN pg_class table_ ON table_.oid = indrelid
422
        JOIN pg_attribute ON attrelid = indrelid AND attnum = ANY (indkey)
423
        WHERE
424
            table_.relname = %(table)s
425
            AND index.relname = %(index)s
426
    UNION
427
        SELECT attnum, attname
428
        FROM
429
        (
430
            SELECT
431
                indrelid
432
                , (regexp_matches(indexprs, E':varattno (\\\\d+)', 'g'))[1]::int
433
                    AS indkey
434
            FROM pg_index
435
            JOIN pg_class index ON index.oid = indexrelid
436
            JOIN pg_class table_ ON table_.oid = indrelid
437
            WHERE
438
                table_.relname = %(table)s
439
                AND index.relname = %(index)s
440
        ) s
441
        JOIN pg_attribute ON attrelid = indrelid AND attnum = indkey
442
) s
443 853 aaronmk
ORDER BY attnum
444
''',
445 1909 aaronmk
            {'table': table, 'index': index}, cacheable=True)))
446
    else: raise NotImplementedError("Can't list index columns for "+module+
447
        ' database')
448 853 aaronmk
449 464 aaronmk
def constraint_cols(db, table, constraint):
450
    check_name(table)
451
    check_name(constraint)
452 1849 aaronmk
    module = util.root_module(db.db)
453 464 aaronmk
    if module == 'psycopg2':
454
        return list(values(run_query(db, '''\
455
SELECT attname
456
FROM pg_constraint
457
JOIN pg_class ON pg_class.oid = conrelid
458
JOIN pg_attribute ON attrelid = conrelid AND attnum = ANY (conkey)
459
WHERE
460
    relname = %(table)s
461
    AND conname = %(constraint)s
462
ORDER BY attnum
463
''',
464
            {'table': table, 'constraint': constraint})))
465
    else: raise NotImplementedError("Can't list constraint columns for "+module+
466
        ' database')
467
468 1968 aaronmk
def tables(db, schema='public', table_like='%'):
469 1849 aaronmk
    module = util.root_module(db.db)
470 1968 aaronmk
    params = {'schema': schema, 'table_like': table_like}
471 832 aaronmk
    if module == 'psycopg2':
472 1968 aaronmk
        return values(run_query(db, '''\
473
SELECT tablename
474
FROM pg_tables
475
WHERE
476
    schemaname = %(schema)s
477
    AND tablename LIKE %(table_like)s
478
ORDER BY tablename
479
''',
480
            params, cacheable=True))
481
    elif module == 'MySQLdb':
482
        return values(run_query(db, 'SHOW TABLES LIKE %(table_like)s', params,
483
            cacheable=True))
484 832 aaronmk
    else: raise NotImplementedError("Can't list tables for "+module+' database')
485 830 aaronmk
486 833 aaronmk
##### Database management
487
488 1968 aaronmk
def empty_db(db, schema='public', **kw_args):
489
    '''For kw_args, see tables()'''
490
    for table in tables(db, schema, **kw_args): truncate(db, table, schema)
491 833 aaronmk
492 832 aaronmk
##### Heuristic queries
493
494 1554 aaronmk
def try_insert(db, table, row, returning=None):
495 830 aaronmk
    '''Recovers from errors'''
496 1554 aaronmk
    try: return insert(db, table, row, returning, recover=True)
497 46 aaronmk
    except Exception, e:
498
        msg = str(e)
499 465 aaronmk
        match = re.search(r'duplicate key value violates unique constraint '
500
            r'"(([^\W_]+)_[^"]+)"', msg)
501
        if match:
502
            constraint, table = match.groups()
503 854 aaronmk
            try: cols = index_cols(db, table, constraint)
504 465 aaronmk
            except NotImplementedError: raise e
505 851 aaronmk
            else: raise DuplicateKeyException(cols, e)
506 13 aaronmk
        match = re.search(r'null value in column "(\w+)" violates not-null '
507
            'constraint', msg)
508 470 aaronmk
        if match: raise NullValueException([match.group(1)], e)
509 13 aaronmk
        raise # no specific exception raised
510 11 aaronmk
511 471 aaronmk
def put(db, table, row, pkey, row_ct_ref=None):
512 1554 aaronmk
    '''Recovers from errors.
513
    Only works under PostgreSQL (uses `INSERT ... RETURNING`)'''
514 471 aaronmk
    try:
515 1554 aaronmk
        cur = try_insert(db, table, row, pkey)
516
        if row_ct_ref != None and cur.rowcount >= 0:
517
            row_ct_ref[0] += cur.rowcount
518
        return value(cur)
519 471 aaronmk
    except DuplicateKeyException, e:
520 1069 aaronmk
        return value(select(db, table, [pkey],
521
            util.dict_subset_right_join(row, e.cols), recover=True))
522 471 aaronmk
523 473 aaronmk
def get(db, table, row, pkey, row_ct_ref=None, create=False):
524 830 aaronmk
    '''Recovers from errors'''
525
    try: return value(select(db, table, [pkey], row, 1, recover=True))
526 14 aaronmk
    except StopIteration:
527 40 aaronmk
        if not create: raise
528 471 aaronmk
        return put(db, table, row, pkey, row_ct_ref) # insert new row