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
def esc_name_by_module(module, name, preserve_case=False):
230
    if module == 'psycopg2':
231
        if preserve_case: quote = '"'
232
        # Don't enclose in quotes because this disables case-insensitivity
233
        else: return name
234
    elif module == 'MySQLdb': quote = '`'
235
    else: raise NotImplementedError("Can't escape name for "+module+' database')
236
    return quote + name.replace(quote, '') + quote
237
238
def esc_name_by_engine(engine, name, **kw_args):
239
    return esc_name_by_module(db_engines[engine][0], name, **kw_args)
240
241
def esc_name(db, name, **kw_args):
242
    return esc_name_by_module(util.root_module(db.db), name, **kw_args)
243
244 1968 aaronmk
def qual_name(db, schema, table):
245 1981 aaronmk
    def esc_name_(name): return esc_name(db, name, preserve_case=True)
246 2051 aaronmk
    table = esc_name_(table)
247
    if schema != None: return esc_name_(schema)+'.'+table
248
    else: return table
249 1968 aaronmk
250 832 aaronmk
##### Querying
251
252 1894 aaronmk
def run_raw_query(db, *args, **kw_args):
253
    '''For args, see DbConn.run_query()'''
254
    return db.run_query(*args, **kw_args)
255 11 aaronmk
256 832 aaronmk
##### Recoverable querying
257 15 aaronmk
258 11 aaronmk
def with_savepoint(db, func):
259 1872 aaronmk
    savepoint = 'savepoint_'+str(rand.rand_int()) # must be unique
260 830 aaronmk
    run_raw_query(db, 'SAVEPOINT '+savepoint)
261 11 aaronmk
    try: return_val = func()
262
    except:
263 830 aaronmk
        run_raw_query(db, 'ROLLBACK TO SAVEPOINT '+savepoint)
264 11 aaronmk
        raise
265
    else:
266 830 aaronmk
        run_raw_query(db, 'RELEASE SAVEPOINT '+savepoint)
267 11 aaronmk
        return return_val
268
269 1894 aaronmk
def run_query(db, query, params=None, recover=None, cacheable=False):
270 830 aaronmk
    if recover == None: recover = False
271
272 1894 aaronmk
    def run(): return run_raw_query(db, query, params, cacheable)
273 1914 aaronmk
    if recover and not db.is_cached(query, params):
274
        return with_savepoint(db, run)
275
    else: return run() # don't need savepoint if cached
276 830 aaronmk
277 832 aaronmk
##### Basic queries
278
279 1135 aaronmk
def select(db, table, fields=None, conds=None, limit=None, start=None,
280 1981 aaronmk
    recover=None, cacheable=True, table_is_esc=False):
281
    '''
282
    @param fields Use None to select all fields in the table
283
    @param table_is_esc Whether the table name has already been escaped
284
    '''
285 1135 aaronmk
    if conds == None: conds = {}
286 135 aaronmk
    assert limit == None or type(limit) == int
287 865 aaronmk
    assert start == None or type(start) == int
288 1981 aaronmk
    if not table_is_esc: check_name(table)
289 1135 aaronmk
    if fields != None: map(check_name, fields)
290 15 aaronmk
    map(check_name, conds.keys())
291 865 aaronmk
292 11 aaronmk
    def cond(entry):
293 13 aaronmk
        col, value = entry
294 644 aaronmk
        cond_ = esc_name(db, col)+' '
295 11 aaronmk
        if value == None: cond_ += 'IS'
296
        else: cond_ += '='
297
        cond_ += ' %s'
298
        return cond_
299 1135 aaronmk
    query = 'SELECT '
300
    if fields == None: query += '*'
301
    else: query += ', '.join([esc_name(db, field) for field in fields])
302
    query += ' FROM '+esc_name(db, table)
303 865 aaronmk
304
    missing = True
305 89 aaronmk
    if conds != {}:
306
        query += ' WHERE '+' AND '.join(map(cond, conds.iteritems()))
307 865 aaronmk
        missing = False
308
    if limit != None: query += ' LIMIT '+str(limit); missing = False
309
    if start != None:
310
        if start != 0: query += ' OFFSET '+str(start)
311
        missing = False
312
    if missing: warnings.warn(DbWarning(
313
        'SELECT statement missing a WHERE, LIMIT, or OFFSET clause: '+query))
314
315 1905 aaronmk
    return run_query(db, query, conds.values(), recover, cacheable)
316 11 aaronmk
317 1961 aaronmk
default = object() # tells insert() to use the default value for a column
318
319 1960 aaronmk
def insert(db, table, row, returning=None, recover=None, cacheable=True,
320
    table_is_esc=False):
321
    '''
322
    @param returning str|None An inserted column (such as pkey) to return
323
    @param table_is_esc Whether the table name has already been escaped
324
    '''
325
    if not table_is_esc: check_name(table)
326
    if lists.is_seq(row): cols = None
327
    else:
328
        cols = row.keys()
329
        row = row.values()
330
        map(check_name, cols)
331
    row = list(row) # ensure that "!= []" works
332
333 1961 aaronmk
    # Check for special values
334
    labels = []
335
    values = []
336
    for value in row:
337
        if value == default: labels.append('DEFAULT')
338
        else:
339
            labels.append('%s')
340
            values.append(value)
341
342
    # Build query
343 89 aaronmk
    query = 'INSERT INTO '+table
344 1961 aaronmk
    if values != []:
345 1960 aaronmk
        if cols != None: query += ' ('+', '.join(cols)+')'
346 1961 aaronmk
        query += ' VALUES ('+(', '.join(labels))+')'
347 89 aaronmk
    else: query += ' DEFAULT VALUES'
348 1554 aaronmk
349
    if returning != None:
350
        check_name(returning)
351
        query += ' RETURNING '+returning
352
353 1961 aaronmk
    return run_query(db, query, values, recover, cacheable)
354 11 aaronmk
355 135 aaronmk
def last_insert_id(db):
356 1849 aaronmk
    module = util.root_module(db.db)
357 135 aaronmk
    if module == 'psycopg2': return value(run_query(db, 'SELECT lastval()'))
358
    elif module == 'MySQLdb': return db.insert_id()
359
    else: return None
360 13 aaronmk
361 1968 aaronmk
def truncate(db, table, schema='public'):
362
    return run_query(db, 'TRUNCATE '+qual_name(db, schema, table)+' CASCADE')
363 832 aaronmk
364
##### Database structure queries
365
366 1850 aaronmk
def pkey(db, table, recover=None):
367 832 aaronmk
    '''Assumed to be first column in table'''
368
    check_name(table)
369 1929 aaronmk
    return col_names(select(db, table, limit=0, recover=recover)).next()
370 832 aaronmk
371 853 aaronmk
def index_cols(db, table, index):
372
    '''Can also use this for UNIQUE constraints, because a UNIQUE index is
373
    automatically created. When you don't know whether something is a UNIQUE
374
    constraint or a UNIQUE index, use this function.'''
375
    check_name(table)
376
    check_name(index)
377 1909 aaronmk
    module = util.root_module(db.db)
378
    if module == 'psycopg2':
379
        return list(values(run_query(db, '''\
380 853 aaronmk
SELECT attname
381 866 aaronmk
FROM
382
(
383
        SELECT attnum, attname
384
        FROM pg_index
385
        JOIN pg_class index ON index.oid = indexrelid
386
        JOIN pg_class table_ ON table_.oid = indrelid
387
        JOIN pg_attribute ON attrelid = indrelid AND attnum = ANY (indkey)
388
        WHERE
389
            table_.relname = %(table)s
390
            AND index.relname = %(index)s
391
    UNION
392
        SELECT attnum, attname
393
        FROM
394
        (
395
            SELECT
396
                indrelid
397
                , (regexp_matches(indexprs, E':varattno (\\\\d+)', 'g'))[1]::int
398
                    AS indkey
399
            FROM pg_index
400
            JOIN pg_class index ON index.oid = indexrelid
401
            JOIN pg_class table_ ON table_.oid = indrelid
402
            WHERE
403
                table_.relname = %(table)s
404
                AND index.relname = %(index)s
405
        ) s
406
        JOIN pg_attribute ON attrelid = indrelid AND attnum = indkey
407
) s
408 853 aaronmk
ORDER BY attnum
409
''',
410 1909 aaronmk
            {'table': table, 'index': index}, cacheable=True)))
411
    else: raise NotImplementedError("Can't list index columns for "+module+
412
        ' database')
413 853 aaronmk
414 464 aaronmk
def constraint_cols(db, table, constraint):
415
    check_name(table)
416
    check_name(constraint)
417 1849 aaronmk
    module = util.root_module(db.db)
418 464 aaronmk
    if module == 'psycopg2':
419
        return list(values(run_query(db, '''\
420
SELECT attname
421
FROM pg_constraint
422
JOIN pg_class ON pg_class.oid = conrelid
423
JOIN pg_attribute ON attrelid = conrelid AND attnum = ANY (conkey)
424
WHERE
425
    relname = %(table)s
426
    AND conname = %(constraint)s
427
ORDER BY attnum
428
''',
429
            {'table': table, 'constraint': constraint})))
430
    else: raise NotImplementedError("Can't list constraint columns for "+module+
431
        ' database')
432
433 1968 aaronmk
def tables(db, schema='public', table_like='%'):
434 1849 aaronmk
    module = util.root_module(db.db)
435 1968 aaronmk
    params = {'schema': schema, 'table_like': table_like}
436 832 aaronmk
    if module == 'psycopg2':
437 1968 aaronmk
        return values(run_query(db, '''\
438
SELECT tablename
439
FROM pg_tables
440
WHERE
441
    schemaname = %(schema)s
442
    AND tablename LIKE %(table_like)s
443
ORDER BY tablename
444
''',
445
            params, cacheable=True))
446
    elif module == 'MySQLdb':
447
        return values(run_query(db, 'SHOW TABLES LIKE %(table_like)s', params,
448
            cacheable=True))
449 832 aaronmk
    else: raise NotImplementedError("Can't list tables for "+module+' database')
450 830 aaronmk
451 833 aaronmk
##### Database management
452
453 1968 aaronmk
def empty_db(db, schema='public', **kw_args):
454
    '''For kw_args, see tables()'''
455
    for table in tables(db, schema, **kw_args): truncate(db, table, schema)
456 833 aaronmk
457 832 aaronmk
##### Heuristic queries
458
459 1554 aaronmk
def try_insert(db, table, row, returning=None):
460 830 aaronmk
    '''Recovers from errors'''
461 1554 aaronmk
    try: return insert(db, table, row, returning, recover=True)
462 46 aaronmk
    except Exception, e:
463
        msg = str(e)
464 465 aaronmk
        match = re.search(r'duplicate key value violates unique constraint '
465
            r'"(([^\W_]+)_[^"]+)"', msg)
466
        if match:
467
            constraint, table = match.groups()
468 854 aaronmk
            try: cols = index_cols(db, table, constraint)
469 465 aaronmk
            except NotImplementedError: raise e
470 851 aaronmk
            else: raise DuplicateKeyException(cols, e)
471 13 aaronmk
        match = re.search(r'null value in column "(\w+)" violates not-null '
472
            'constraint', msg)
473 470 aaronmk
        if match: raise NullValueException([match.group(1)], e)
474 13 aaronmk
        raise # no specific exception raised
475 11 aaronmk
476 471 aaronmk
def put(db, table, row, pkey, row_ct_ref=None):
477 1554 aaronmk
    '''Recovers from errors.
478
    Only works under PostgreSQL (uses `INSERT ... RETURNING`)'''
479 471 aaronmk
    try:
480 1554 aaronmk
        cur = try_insert(db, table, row, pkey)
481
        if row_ct_ref != None and cur.rowcount >= 0:
482
            row_ct_ref[0] += cur.rowcount
483
        return value(cur)
484 471 aaronmk
    except DuplicateKeyException, e:
485 1069 aaronmk
        return value(select(db, table, [pkey],
486
            util.dict_subset_right_join(row, e.cols), recover=True))
487 471 aaronmk
488 473 aaronmk
def get(db, table, row, pkey, row_ct_ref=None, create=False):
489 830 aaronmk
    '''Recovers from errors'''
490
    try: return value(select(db, table, [pkey], row, 1, recover=True))
491 14 aaronmk
    except StopIteration:
492 40 aaronmk
        if not create: raise
493 471 aaronmk
        return put(db, table, row, pkey, row_ct_ref) # insert new row