Project

General

Profile

1 11 aaronmk
# Database access
2
3 1869 aaronmk
import copy
4 2127 aaronmk
import operator
5 11 aaronmk
import re
6 865 aaronmk
import warnings
7 11 aaronmk
8 300 aaronmk
import exc
9 1909 aaronmk
import dicts
10 1893 aaronmk
import iters
11 1960 aaronmk
import lists
12 1889 aaronmk
from Proxy import Proxy
13 1872 aaronmk
import rand
14 2217 aaronmk
import sql_gen
15 862 aaronmk
import strings
16 131 aaronmk
import util
17 11 aaronmk
18 832 aaronmk
##### Exceptions
19
20 2170 aaronmk
def get_cur_query(cur, input_query=None, input_params=None):
21 2168 aaronmk
    raw_query = None
22
    if hasattr(cur, 'query'): raw_query = cur.query
23
    elif hasattr(cur, '_last_executed'): raw_query = cur._last_executed
24 2170 aaronmk
25
    if raw_query != None: return raw_query
26
    else: return repr(input_query)+' % '+repr(input_params)
27 14 aaronmk
28 2170 aaronmk
def _add_cursor_info(e, *args, **kw_args):
29
    '''For params, see get_cur_query()'''
30
    exc.add_msg(e, 'query: '+str(get_cur_query(*args, **kw_args)))
31 135 aaronmk
32 300 aaronmk
class DbException(exc.ExceptionWithCause):
33 14 aaronmk
    def __init__(self, msg, cause=None, cur=None):
34 2145 aaronmk
        exc.ExceptionWithCause.__init__(self, msg, cause, cause_newline=True)
35 14 aaronmk
        if cur != None: _add_cursor_info(self, cur)
36
37 2143 aaronmk
class ExceptionWithName(DbException):
38
    def __init__(self, name, cause=None):
39 2145 aaronmk
        DbException.__init__(self, 'for name: '+str(name), cause)
40 2143 aaronmk
        self.name = name
41 360 aaronmk
42 2240 aaronmk
class ExceptionWithNameValue(DbException):
43
    def __init__(self, name, value, cause=None):
44
        DbException.__init__(self,
45 2247 aaronmk
            'for name: '+str(name)+'; value: '+strings.ustr(value), cause)
46 2240 aaronmk
        self.name = name
47
        self.value = value
48
49 468 aaronmk
class ExceptionWithColumns(DbException):
50
    def __init__(self, cols, cause=None):
51 2145 aaronmk
        DbException.__init__(self, 'for columns: '+(', '.join(cols)), cause)
52 468 aaronmk
        self.cols = cols
53 11 aaronmk
54 2143 aaronmk
class NameException(DbException): pass
55
56 468 aaronmk
class DuplicateKeyException(ExceptionWithColumns): pass
57 13 aaronmk
58 468 aaronmk
class NullValueException(ExceptionWithColumns): pass
59 13 aaronmk
60 2240 aaronmk
class FunctionValueException(ExceptionWithNameValue): pass
61 2239 aaronmk
62 2143 aaronmk
class DuplicateTableException(ExceptionWithName): pass
63
64 2188 aaronmk
class DuplicateFunctionException(ExceptionWithName): pass
65
66 89 aaronmk
class EmptyRowException(DbException): pass
67
68 865 aaronmk
##### Warnings
69
70
class DbWarning(UserWarning): pass
71
72 1930 aaronmk
##### Result retrieval
73
74
def col_names(cur): return (col[0] for col in cur.description)
75
76
def rows(cur): return iter(lambda: cur.fetchone(), None)
77
78
def consume_rows(cur):
79
    '''Used to fetch all rows so result will be cached'''
80
    iters.consume_iter(rows(cur))
81
82
def next_row(cur): return rows(cur).next()
83
84
def row(cur):
85
    row_ = next_row(cur)
86
    consume_rows(cur)
87
    return row_
88
89
def next_value(cur): return next_row(cur)[0]
90
91
def value(cur): return row(cur)[0]
92
93
def values(cur): return iters.func_iter(lambda: next_value(cur))
94
95
def value_or_none(cur):
96
    try: return value(cur)
97
    except StopIteration: return None
98
99 2101 aaronmk
##### Input validation
100
101 2198 aaronmk
def clean_name(name): return re.sub(r'\W', r'', name).lower()
102 2101 aaronmk
103
def check_name(name):
104
    if re.search(r'\W', name) != None: raise NameException('Name "'+name
105
        +'" may contain only alphanumeric characters and _')
106
107
def esc_name_by_module(module, name, ignore_case=False):
108
    if module == 'psycopg2':
109
        if ignore_case:
110
            # Don't enclose in quotes because this disables case-insensitivity
111
            check_name(name)
112
            return name
113
        else: quote = '"'
114
    elif module == 'MySQLdb': quote = '`'
115
    else: raise NotImplementedError("Can't escape name for "+module+' database')
116
    return quote + name.replace(quote, '') + quote
117
118
def esc_name_by_engine(engine, name, **kw_args):
119
    return esc_name_by_module(db_engines[engine][0], name, **kw_args)
120
121
def esc_name(db, name, **kw_args):
122
    return esc_name_by_module(util.root_module(db.db), name, **kw_args)
123
124
def qual_name(db, schema, table):
125
    def esc_name_(name): return esc_name(db, name)
126
    table = esc_name_(table)
127
    if schema != None: return esc_name_(schema)+'.'+table
128
    else: return table
129
130 1869 aaronmk
##### Database connections
131 1849 aaronmk
132 2097 aaronmk
db_config_names = ['engine', 'host', 'user', 'password', 'database', 'schemas']
133 1926 aaronmk
134 1869 aaronmk
db_engines = {
135
    'MySQL': ('MySQLdb', {'password': 'passwd', 'database': 'db'}),
136
    'PostgreSQL': ('psycopg2', {}),
137
}
138
139
DatabaseErrors_set = set([DbException])
140
DatabaseErrors = tuple(DatabaseErrors_set)
141
142
def _add_module(module):
143
    DatabaseErrors_set.add(module.DatabaseError)
144
    global DatabaseErrors
145
    DatabaseErrors = tuple(DatabaseErrors_set)
146
147
def db_config_str(db_config):
148
    return db_config['engine']+' database '+db_config['database']
149
150 1909 aaronmk
def _query_lookup(query, params): return (query, dicts.make_hashable(params))
151 1894 aaronmk
152 1901 aaronmk
log_debug_none = lambda msg: None
153
154 1849 aaronmk
class DbConn:
155 2190 aaronmk
    def __init__(self, db_config, serializable=True, autocommit=False,
156
        caching=True, log_debug=log_debug_none):
157 1869 aaronmk
        self.db_config = db_config
158
        self.serializable = serializable
159 2190 aaronmk
        self.autocommit = autocommit
160
        self.caching = caching
161 1901 aaronmk
        self.log_debug = log_debug
162 2193 aaronmk
        self.debug = log_debug != log_debug_none
163 1869 aaronmk
164
        self.__db = None
165 1889 aaronmk
        self.query_results = {}
166 2139 aaronmk
        self._savepoint = 0
167 1869 aaronmk
168
    def __getattr__(self, name):
169
        if name == '__dict__': raise Exception('getting __dict__')
170
        if name == 'db': return self._db()
171
        else: raise AttributeError()
172
173
    def __getstate__(self):
174
        state = copy.copy(self.__dict__) # shallow copy
175 1915 aaronmk
        state['log_debug'] = None # don't pickle the debug callback
176 1869 aaronmk
        state['_DbConn__db'] = None # don't pickle the connection
177
        return state
178
179 2165 aaronmk
    def connected(self): return self.__db != None
180
181 1869 aaronmk
    def _db(self):
182
        if self.__db == None:
183
            # Process db_config
184
            db_config = self.db_config.copy() # don't modify input!
185 2097 aaronmk
            schemas = db_config.pop('schemas', None)
186 1869 aaronmk
            module_name, mappings = db_engines[db_config.pop('engine')]
187
            module = __import__(module_name)
188
            _add_module(module)
189
            for orig, new in mappings.iteritems():
190
                try: util.rename_key(db_config, orig, new)
191
                except KeyError: pass
192
193
            # Connect
194
            self.__db = module.connect(**db_config)
195
196
            # Configure connection
197 2234 aaronmk
            if self.serializable and not self.autocommit: run_raw_query(self,
198
                'SET TRANSACTION ISOLATION LEVEL SERIALIZABLE')
199 2101 aaronmk
            if schemas != None:
200
                schemas_ = ''.join((esc_name(self, s)+', '
201
                    for s in schemas.split(',')))
202
                run_raw_query(self, "SELECT set_config('search_path', \
203
%s || current_setting('search_path'), false)", [schemas_])
204 1869 aaronmk
205
        return self.__db
206 1889 aaronmk
207 1891 aaronmk
    class DbCursor(Proxy):
208 1927 aaronmk
        def __init__(self, outer):
209 1891 aaronmk
            Proxy.__init__(self, outer.db.cursor())
210 2191 aaronmk
            self.outer = outer
211 1927 aaronmk
            self.query_results = outer.query_results
212 1894 aaronmk
            self.query_lookup = None
213 1891 aaronmk
            self.result = []
214 1889 aaronmk
215 1894 aaronmk
        def execute(self, query, params=None):
216 1930 aaronmk
            self._is_insert = query.upper().find('INSERT') >= 0
217 1894 aaronmk
            self.query_lookup = _query_lookup(query, params)
218 2148 aaronmk
            try:
219 2191 aaronmk
                try:
220
                    return_value = self.inner.execute(query, params)
221
                    self.outer.do_autocommit()
222 2148 aaronmk
                finally: self.query = get_cur_query(self.inner)
223 1904 aaronmk
            except Exception, e:
224 2170 aaronmk
                _add_cursor_info(e, self, query, params)
225 1904 aaronmk
                self.result = e # cache the exception as the result
226
                self._cache_result()
227
                raise
228 1930 aaronmk
            # Fetch all rows so result will be cached
229
            if self.rowcount == 0 and not self._is_insert: consume_rows(self)
230 1894 aaronmk
            return return_value
231
232 1891 aaronmk
        def fetchone(self):
233
            row = self.inner.fetchone()
234 1899 aaronmk
            if row != None: self.result.append(row)
235
            # otherwise, fetched all rows
236 1904 aaronmk
            else: self._cache_result()
237
            return row
238
239
        def _cache_result(self):
240 1906 aaronmk
            # For inserts, only cache exceptions since inserts are not
241
            # idempotent, but an invalid insert will always be invalid
242 1930 aaronmk
            if self.query_results != None and (not self._is_insert
243 1906 aaronmk
                or isinstance(self.result, Exception)):
244
245 1894 aaronmk
                assert self.query_lookup != None
246 1916 aaronmk
                self.query_results[self.query_lookup] = self.CacheCursor(
247
                    util.dict_subset(dicts.AttrsDictView(self),
248
                    ['query', 'result', 'rowcount', 'description']))
249 1906 aaronmk
250 1916 aaronmk
        class CacheCursor:
251
            def __init__(self, cached_result): self.__dict__ = cached_result
252
253 1927 aaronmk
            def execute(self, *args, **kw_args):
254 1916 aaronmk
                if isinstance(self.result, Exception): raise self.result
255
                # otherwise, result is a rows list
256
                self.iter = iter(self.result)
257
258
            def fetchone(self):
259
                try: return self.iter.next()
260
                except StopIteration: return None
261 1891 aaronmk
262 2212 aaronmk
    def esc_value(self, value):
263 2215 aaronmk
        module = util.root_module(self.db)
264
        if module == 'psycopg2': return self.db.cursor().mogrify('%s', [value])
265 2212 aaronmk
        elif module == 'MySQLdb':
266
            import _mysql
267
            return _mysql.escape_string(value)
268
        else: raise NotImplementedError("Can't escape value for "+module
269
            +' database')
270
271 1894 aaronmk
    def run_query(self, query, params=None, cacheable=False):
272 2148 aaronmk
        '''Translates known DB errors to typed exceptions:
273
        See self.DbCursor.execute().'''
274 2167 aaronmk
        assert query != None
275
276 2047 aaronmk
        if not self.caching: cacheable = False
277 1903 aaronmk
        used_cache = False
278
        try:
279 1927 aaronmk
            # Get cursor
280
            if cacheable:
281
                query_lookup = _query_lookup(query, params)
282
                try:
283
                    cur = self.query_results[query_lookup]
284
                    used_cache = True
285
                except KeyError: cur = self.DbCursor(self)
286
            else: cur = self.db.cursor()
287
288
            # Run query
289 2148 aaronmk
            cur.execute(query, params)
290 1903 aaronmk
        finally:
291 2193 aaronmk
            if self.debug: # only compute msg if needed
292 1903 aaronmk
                if used_cache: cache_status = 'Cache hit'
293
                elif cacheable: cache_status = 'Cache miss'
294
                else: cache_status = 'Non-cacheable'
295 1927 aaronmk
                self.log_debug(cache_status+': '
296 2170 aaronmk
                    +strings.one_line(str(get_cur_query(cur, query, params))))
297 1903 aaronmk
298
        return cur
299 1914 aaronmk
300
    def is_cached(self, query, params=None):
301
        return _query_lookup(query, params) in self.query_results
302 2139 aaronmk
303
    def with_savepoint(self, func):
304 2171 aaronmk
        savepoint = 'level_'+str(self._savepoint)
305 2139 aaronmk
        self.run_query('SAVEPOINT '+savepoint)
306
        self._savepoint += 1
307
        try:
308
            try: return_val = func()
309
            finally:
310
                self._savepoint -= 1
311
                assert self._savepoint >= 0
312
        except:
313
            self.run_query('ROLLBACK TO SAVEPOINT '+savepoint)
314
            raise
315
        else:
316
            self.run_query('RELEASE SAVEPOINT '+savepoint)
317 2191 aaronmk
            self.do_autocommit()
318 2139 aaronmk
            return return_val
319 2191 aaronmk
320
    def do_autocommit(self):
321
        '''Autocommits if outside savepoint'''
322
        assert self._savepoint >= 0
323
        if self.autocommit and self._savepoint == 0:
324
            self.log_debug('Autocommiting')
325
            self.db.commit()
326 1849 aaronmk
327 1869 aaronmk
connect = DbConn
328
329 832 aaronmk
##### Querying
330
331 1894 aaronmk
def run_raw_query(db, *args, **kw_args):
332 2085 aaronmk
    '''For params, see DbConn.run_query()'''
333 1894 aaronmk
    return db.run_query(*args, **kw_args)
334 11 aaronmk
335 2068 aaronmk
def mogrify(db, query, params):
336
    module = util.root_module(db.db)
337
    if module == 'psycopg2': return db.db.cursor().mogrify(query, params)
338
    else: raise NotImplementedError("Can't mogrify query for "+module+
339
        ' database')
340
341 832 aaronmk
##### Recoverable querying
342 15 aaronmk
343 2139 aaronmk
def with_savepoint(db, func): return db.with_savepoint(func)
344 11 aaronmk
345 1894 aaronmk
def run_query(db, query, params=None, recover=None, cacheable=False):
346 830 aaronmk
    if recover == None: recover = False
347
348 2148 aaronmk
    try:
349
        def run(): return run_raw_query(db, query, params, cacheable)
350
        if recover and not db.is_cached(query, params):
351
            return with_savepoint(db, run)
352
        else: return run() # don't need savepoint if cached
353
    except Exception, e:
354
        if not recover: raise # need savepoint to run index_cols()
355 2247 aaronmk
        msg = exc.str_(e)
356 2240 aaronmk
357 2148 aaronmk
        match = re.search(r'duplicate key value violates unique constraint '
358
            r'"((_?[^\W_]+)_[^"]+)"', msg)
359
        if match:
360
            constraint, table = match.groups()
361
            try: cols = index_cols(db, table, constraint)
362
            except NotImplementedError: raise e
363
            else: raise DuplicateKeyException(cols, e)
364 2240 aaronmk
365 2148 aaronmk
        match = re.search(r'null value in column "(\w+)" violates not-null '
366 2239 aaronmk
            r'constraint', msg)
367 2148 aaronmk
        if match: raise NullValueException([match.group(1)], e)
368 2240 aaronmk
369 2247 aaronmk
        match = re.search(r'invalid input (?:syntax|value)\b.*: "(.+)"\n'
370 2242 aaronmk
            r'(?:(?s).*)\bfunction "(\w+)".*\bat assignment', msg)
371 2240 aaronmk
        if match:
372 2247 aaronmk
            raise FunctionValueException(match.group(2),
373
                strings.to_unicode(match.group(1)), e)
374 2240 aaronmk
375 2148 aaronmk
        match = re.search(r'relation "(\w+)" already exists', msg)
376
        if match: raise DuplicateTableException(match.group(1), e)
377 2240 aaronmk
378 2188 aaronmk
        match = re.search(r'function "(\w+)" already exists', msg)
379
        if match: raise DuplicateFunctionException(match.group(1), e)
380 2240 aaronmk
381 2148 aaronmk
        raise # no specific exception raised
382 830 aaronmk
383 832 aaronmk
##### Basic queries
384
385 2153 aaronmk
def next_version(name):
386
    '''Prepends the version # so it won't be removed if the name is truncated'''
387 2163 aaronmk
    version = 1 # first existing name was version 0
388 2153 aaronmk
    match = re.match(r'^v(\d+)_(.*)$', name)
389
    if match:
390
        version = int(match.group(1))+1
391
        name = match.group(2)
392
    return 'v'+str(version)+'_'+name
393
394 2151 aaronmk
def run_query_into(db, query, params, into_ref=None, *args, **kw_args):
395 2085 aaronmk
    '''Outputs a query to a temp table.
396
    For params, see run_query().
397
    '''
398 2151 aaronmk
    if into_ref == None: return run_query(db, query, params, *args, **kw_args)
399 2085 aaronmk
    else: # place rows in temp table
400 2151 aaronmk
        check_name(into_ref[0])
401 2153 aaronmk
        kw_args['recover'] = True
402
        while True:
403
            try:
404 2194 aaronmk
                create_query = 'CREATE'
405
                if not db.debug: create_query += ' TEMP'
406
                create_query += ' TABLE '+into_ref[0]+' AS '+query
407
408
                return run_query(db, create_query, params, *args, **kw_args)
409 2153 aaronmk
                    # CREATE TABLE AS sets rowcount to # rows in query
410
            except DuplicateTableException, e:
411
                into_ref[0] = next_version(into_ref[0])
412
                # try again with next version of name
413 2085 aaronmk
414 2120 aaronmk
order_by_pkey = object() # tells mk_select() to order by the pkey
415
416 2127 aaronmk
join_using = object() # tells mk_select() to join the column with USING
417
418 2187 aaronmk
filter_out = object() # tells mk_select() to filter out rows that match the join
419 2180 aaronmk
420 2199 aaronmk
distinct_on_all = object() # tells mk_select() to SELECT DISTINCT ON all columns
421
422 2233 aaronmk
def mk_select(db, tables, fields=None, conds=None, distinct_on=[], limit=None,
423 2227 aaronmk
    start=None, order_by=order_by_pkey, table_is_esc=False, default_table=None):
424 1981 aaronmk
    '''
425 2121 aaronmk
    @param tables The single table to select from, or a list of tables to join
426 2187 aaronmk
        together: [table0, (table1, joins), ...]
427
428
        joins has the format: dict(right_col=left_col, ...)
429
        * if left_col is join_using, left_col is set to right_col
430
        * if left_col is filter_out, the tables are LEFT JOINed together and the
431
          query is filtered by `right_col IS NULL` (indicating no match)
432 1981 aaronmk
    @param fields Use None to select all fields in the table
433 2218 aaronmk
    @param conds WHERE conditions: dict(compare_left_side=compare_right_side):
434
        * compare_left_side: Code|str (for col name)
435
        * compare_right_side: ValueCond|literal value
436 2199 aaronmk
    @param distinct_on The columns to SELECT DISTINCT ON, or distinct_on_all to
437
        use all columns
438 1981 aaronmk
    @param table_is_esc Whether the table name has already been escaped
439 2054 aaronmk
    @return tuple(query, params)
440 1981 aaronmk
    '''
441 2060 aaronmk
    def esc_name_(name): return esc_name(db, name)
442 2058 aaronmk
443 2121 aaronmk
    if not lists.is_seq(tables): tables = [tables]
444 2141 aaronmk
    tables = list(tables) # don't modify input! (list() copies input)
445 2121 aaronmk
    table0 = tables.pop(0) # first table is separate
446
447 1135 aaronmk
    if conds == None: conds = {}
448 135 aaronmk
    assert limit == None or type(limit) == int
449 865 aaronmk
    assert start == None or type(start) == int
450 2254 aaronmk
    if order_by is order_by_pkey:
451 2121 aaronmk
        order_by = pkey(db, table0, recover=True, table_is_esc=table_is_esc)
452
    if not table_is_esc: table0 = esc_name_(table0)
453 865 aaronmk
454 2056 aaronmk
    params = []
455
456 2227 aaronmk
    def parse_col(field):
457 2056 aaronmk
        '''Parses fields'''
458 2227 aaronmk
        return sql_gen.value2sql_gen(field, default_table, table_is_esc,
459
            assume_col=True).to_str(db)
460 11 aaronmk
    def cond(entry):
461 2056 aaronmk
        '''Parses conditions'''
462 2217 aaronmk
        left, right = entry
463 2226 aaronmk
        return sql_gen.cond2sql_gen(right).to_str(db, sql_gen.col2sql_gen(left))
464 2056 aaronmk
465 2200 aaronmk
    query = 'SELECT'
466
467
    # DISTINCT ON columns
468 2233 aaronmk
    if distinct_on != []:
469 2200 aaronmk
        query += ' DISTINCT'
470 2254 aaronmk
        if distinct_on is not distinct_on_all:
471 2200 aaronmk
            query += ' ON ('+(', '.join(map(parse_col, distinct_on)))+')'
472
473
    # Columns
474
    query += ' '
475 1135 aaronmk
    if fields == None: query += '*'
476 2056 aaronmk
    else: query += ', '.join(map(parse_col, fields))
477 2200 aaronmk
478
    # Main table
479 2121 aaronmk
    query += ' FROM '+table0
480 865 aaronmk
481 2122 aaronmk
    # Add joins
482
    left_table = table0
483
    for table, joins in tables:
484
        if not table_is_esc: table = esc_name_(table)
485
486 2238 aaronmk
        left_join = False
487 2187 aaronmk
488 2238 aaronmk
        # Parse special values
489 2255 aaronmk
        joins = joins.copy() # don't modify input!
490 2238 aaronmk
        for right, left in joins.items():
491 2254 aaronmk
            if left is filter_out:
492 2238 aaronmk
                left_join = True
493
                # Can't use join_using because the columns being joined can have
494
                # different values in a LEFT JOIN.
495
                # Use '~=' because it's OK to pass NULLs through.
496
                joins[right] = sql_gen.CompareCond(sql_gen.col2sql_gen(right,
497
                    left_table, table_is_esc), '~=')
498
                conds[(table, right)] = None # filter query by no match
499
500 2122 aaronmk
        def join(entry):
501 2127 aaronmk
            '''Parses non-USING joins'''
502 2124 aaronmk
            right_col, left_col = entry
503 2173 aaronmk
504 2179 aaronmk
            # Parse special values
505 2176 aaronmk
            if left_col == None: left_col = (left_col,)
506
                # for None values, tuple is optional
507 2254 aaronmk
            elif left_col is join_using: left_col = right_col
508 2179 aaronmk
509 2238 aaronmk
            return cond((sql_gen.col2sql_gen(right_col, table, table_is_esc),
510
                sql_gen.cond2sql_gen(left_col, left_table, table_is_esc,
511
                    assume_col=True)))
512 2122 aaronmk
513 2187 aaronmk
        # Create join condition and determine join type
514 2254 aaronmk
        if reduce(operator.and_, (v is join_using for v in joins.itervalues())):
515 2179 aaronmk
            # all cols w/ USING, so can use simpler USING syntax
516 2187 aaronmk
            join_cond = 'USING ('+(', '.join(joins.iterkeys()))+')'
517
        else: join_cond = 'ON '+(' AND '.join(map(join, joins.iteritems())))
518 2127 aaronmk
519 2187 aaronmk
        # Create join
520 2238 aaronmk
        if left_join: query += ' LEFT'
521 2187 aaronmk
        query += ' JOIN '+table+' '+join_cond
522
523 2122 aaronmk
        left_table = table
524
525 865 aaronmk
    missing = True
526 89 aaronmk
    if conds != {}:
527 2122 aaronmk
        query += ' WHERE '+(' AND '.join(map(cond, conds.iteritems())))
528 2056 aaronmk
        params += conds.values()
529 865 aaronmk
        missing = False
530 2227 aaronmk
    if order_by != None:
531
        query += ' ORDER BY '+sql_gen.col2sql_gen(order_by, table0,
532
            table_is_esc).to_str(db)
533 865 aaronmk
    if limit != None: query += ' LIMIT '+str(limit); missing = False
534
    if start != None:
535
        if start != 0: query += ' OFFSET '+str(start)
536
        missing = False
537
    if missing: warnings.warn(DbWarning(
538
        'SELECT statement missing a WHERE, LIMIT, or OFFSET clause: '+query))
539
540 2056 aaronmk
    return (query, params)
541 11 aaronmk
542 2054 aaronmk
def select(db, *args, **kw_args):
543
    '''For params, see mk_select() and run_query()'''
544
    recover = kw_args.pop('recover', None)
545
    cacheable = kw_args.pop('cacheable', True)
546
547
    query, params = mk_select(db, *args, **kw_args)
548
    return run_query(db, query, params, recover, cacheable)
549
550 2066 aaronmk
def mk_insert_select(db, table, cols=None, select_query=None, params=None,
551 2070 aaronmk
    returning=None, embeddable=False, table_is_esc=False):
552 1960 aaronmk
    '''
553
    @param returning str|None An inserted column (such as pkey) to return
554 2070 aaronmk
    @param embeddable Whether the query should be embeddable as a nested SELECT.
555 2073 aaronmk
        Warning: If you set this and cacheable=True when the query is run, the
556
        query will be fully cached, not just if it raises an exception.
557 1960 aaronmk
    @param table_is_esc Whether the table name has already been escaped
558
    '''
559 2063 aaronmk
    if select_query == None: select_query = 'DEFAULT VALUES'
560
    if cols == []: cols = None # no cols (all defaults) = unknown col names
561 1960 aaronmk
    if not table_is_esc: check_name(table)
562 2063 aaronmk
563
    # Build query
564
    query = 'INSERT INTO '+table
565
    if cols != None:
566
        map(check_name, cols)
567
        query += ' ('+', '.join(cols)+')'
568
    query += ' '+select_query
569
570
    if returning != None:
571
        check_name(returning)
572
        query += ' RETURNING '+returning
573
574 2070 aaronmk
    if embeddable:
575
        # Create function
576 2189 aaronmk
        function_name = '_'.join(map(clean_name, ['insert', table] + cols))
577 2070 aaronmk
        return_type = 'SETOF '+table+'.'+returning+'%TYPE'
578 2189 aaronmk
        while True:
579
            try:
580 2194 aaronmk
                function = function_name
581
                if not db.debug: function = 'pg_temp.'+function
582
583 2189 aaronmk
                function_query = '''\
584
CREATE FUNCTION '''+function+'''() RETURNS '''+return_type+'''
585 2070 aaronmk
    LANGUAGE sql
586
    AS $$'''+mogrify(db, query, params)+''';$$;
587
'''
588 2189 aaronmk
                run_query(db, function_query, recover=True, cacheable=True)
589
                break # this version was successful
590
            except DuplicateFunctionException, e:
591
                function_name = next_version(function_name)
592
                # try again with next version of name
593 2070 aaronmk
594
        # Return query that uses function
595 2134 aaronmk
        return mk_select(db, function+'() AS f ('+returning+')', start=0,
596
            order_by=None, table_is_esc=True)# AS clause requires function alias
597 2070 aaronmk
598 2066 aaronmk
    return (query, params)
599
600
def insert_select(db, *args, **kw_args):
601 2085 aaronmk
    '''For params, see mk_insert_select() and run_query_into()
602 2152 aaronmk
    @param into_ref List with name of temp table to place RETURNING values in
603 2072 aaronmk
    '''
604 2151 aaronmk
    into_ref = kw_args.pop('into_ref', None)
605
    if into_ref != None: kw_args['embeddable'] = True
606 2066 aaronmk
    recover = kw_args.pop('recover', None)
607
    cacheable = kw_args.pop('cacheable', True)
608
609
    query, params = mk_insert_select(db, *args, **kw_args)
610 2153 aaronmk
    return run_query_into(db, query, params, into_ref, recover=recover,
611
        cacheable=cacheable)
612 2063 aaronmk
613 2066 aaronmk
default = object() # tells insert() to use the default value for a column
614
615 2063 aaronmk
def insert(db, table, row, *args, **kw_args):
616 2085 aaronmk
    '''For params, see insert_select()'''
617 1960 aaronmk
    if lists.is_seq(row): cols = None
618
    else:
619
        cols = row.keys()
620
        row = row.values()
621
    row = list(row) # ensure that "!= []" works
622
623 1961 aaronmk
    # Check for special values
624
    labels = []
625
    values = []
626
    for value in row:
627 2254 aaronmk
        if value is default: labels.append('DEFAULT')
628 1961 aaronmk
        else:
629
            labels.append('%s')
630
            values.append(value)
631
632
    # Build query
633 2063 aaronmk
    if values != []: query = ' VALUES ('+(', '.join(labels))+')'
634
    else: query = None
635 1554 aaronmk
636 2064 aaronmk
    return insert_select(db, table, cols, query, values, *args, **kw_args)
637 11 aaronmk
638 135 aaronmk
def last_insert_id(db):
639 1849 aaronmk
    module = util.root_module(db.db)
640 135 aaronmk
    if module == 'psycopg2': return value(run_query(db, 'SELECT lastval()'))
641
    elif module == 'MySQLdb': return db.insert_id()
642
    else: return None
643 13 aaronmk
644 1968 aaronmk
def truncate(db, table, schema='public'):
645
    return run_query(db, 'TRUNCATE '+qual_name(db, schema, table)+' CASCADE')
646 832 aaronmk
647
##### Database structure queries
648
649 2084 aaronmk
def pkey(db, table, recover=None, table_is_esc=False):
650 832 aaronmk
    '''Assumed to be first column in table'''
651 2120 aaronmk
    return col_names(select(db, table, limit=0, order_by=None, recover=recover,
652 2084 aaronmk
        table_is_esc=table_is_esc)).next()
653 832 aaronmk
654 853 aaronmk
def index_cols(db, table, index):
655
    '''Can also use this for UNIQUE constraints, because a UNIQUE index is
656
    automatically created. When you don't know whether something is a UNIQUE
657
    constraint or a UNIQUE index, use this function.'''
658
    check_name(table)
659
    check_name(index)
660 1909 aaronmk
    module = util.root_module(db.db)
661
    if module == 'psycopg2':
662
        return list(values(run_query(db, '''\
663 853 aaronmk
SELECT attname
664 866 aaronmk
FROM
665
(
666
        SELECT attnum, attname
667
        FROM pg_index
668
        JOIN pg_class index ON index.oid = indexrelid
669
        JOIN pg_class table_ ON table_.oid = indrelid
670
        JOIN pg_attribute ON attrelid = indrelid AND attnum = ANY (indkey)
671
        WHERE
672
            table_.relname = %(table)s
673
            AND index.relname = %(index)s
674
    UNION
675
        SELECT attnum, attname
676
        FROM
677
        (
678
            SELECT
679
                indrelid
680
                , (regexp_matches(indexprs, E':varattno (\\\\d+)', 'g'))[1]::int
681
                    AS indkey
682
            FROM pg_index
683
            JOIN pg_class index ON index.oid = indexrelid
684
            JOIN pg_class table_ ON table_.oid = indrelid
685
            WHERE
686
                table_.relname = %(table)s
687
                AND index.relname = %(index)s
688
        ) s
689
        JOIN pg_attribute ON attrelid = indrelid AND attnum = indkey
690
) s
691 853 aaronmk
ORDER BY attnum
692
''',
693 1909 aaronmk
            {'table': table, 'index': index}, cacheable=True)))
694
    else: raise NotImplementedError("Can't list index columns for "+module+
695
        ' database')
696 853 aaronmk
697 464 aaronmk
def constraint_cols(db, table, constraint):
698
    check_name(table)
699
    check_name(constraint)
700 1849 aaronmk
    module = util.root_module(db.db)
701 464 aaronmk
    if module == 'psycopg2':
702
        return list(values(run_query(db, '''\
703
SELECT attname
704
FROM pg_constraint
705
JOIN pg_class ON pg_class.oid = conrelid
706
JOIN pg_attribute ON attrelid = conrelid AND attnum = ANY (conkey)
707
WHERE
708
    relname = %(table)s
709
    AND conname = %(constraint)s
710
ORDER BY attnum
711
''',
712
            {'table': table, 'constraint': constraint})))
713
    else: raise NotImplementedError("Can't list constraint columns for "+module+
714
        ' database')
715
716 2096 aaronmk
row_num_col = '_row_num'
717
718 2086 aaronmk
def add_row_num(db, table):
719 2117 aaronmk
    '''Adds a row number column to a table. Its name is in row_num_col. It will
720
    be the primary key.'''
721 2086 aaronmk
    check_name(table)
722 2096 aaronmk
    run_query(db, 'ALTER TABLE '+table+' ADD COLUMN '+row_num_col
723 2117 aaronmk
        +' serial NOT NULL PRIMARY KEY')
724 2086 aaronmk
725 1968 aaronmk
def tables(db, schema='public', table_like='%'):
726 1849 aaronmk
    module = util.root_module(db.db)
727 1968 aaronmk
    params = {'schema': schema, 'table_like': table_like}
728 832 aaronmk
    if module == 'psycopg2':
729 1968 aaronmk
        return values(run_query(db, '''\
730
SELECT tablename
731
FROM pg_tables
732
WHERE
733
    schemaname = %(schema)s
734
    AND tablename LIKE %(table_like)s
735
ORDER BY tablename
736
''',
737
            params, cacheable=True))
738
    elif module == 'MySQLdb':
739
        return values(run_query(db, 'SHOW TABLES LIKE %(table_like)s', params,
740
            cacheable=True))
741 832 aaronmk
    else: raise NotImplementedError("Can't list tables for "+module+' database')
742 830 aaronmk
743 833 aaronmk
##### Database management
744
745 1968 aaronmk
def empty_db(db, schema='public', **kw_args):
746
    '''For kw_args, see tables()'''
747
    for table in tables(db, schema, **kw_args): truncate(db, table, schema)
748 833 aaronmk
749 832 aaronmk
##### Heuristic queries
750
751 2104 aaronmk
def put(db, table, row, pkey_=None, row_ct_ref=None):
752 1554 aaronmk
    '''Recovers from errors.
753 2077 aaronmk
    Only works under PostgreSQL (uses INSERT RETURNING).
754
    '''
755 2104 aaronmk
    if pkey_ == None: pkey_ = pkey(db, table, recover=True)
756
757 471 aaronmk
    try:
758 2149 aaronmk
        cur = insert(db, table, row, pkey_, recover=True)
759 1554 aaronmk
        if row_ct_ref != None and cur.rowcount >= 0:
760
            row_ct_ref[0] += cur.rowcount
761
        return value(cur)
762 471 aaronmk
    except DuplicateKeyException, e:
763 2104 aaronmk
        return value(select(db, table, [pkey_],
764 1069 aaronmk
            util.dict_subset_right_join(row, e.cols), recover=True))
765 471 aaronmk
766 473 aaronmk
def get(db, table, row, pkey, row_ct_ref=None, create=False):
767 830 aaronmk
    '''Recovers from errors'''
768 2209 aaronmk
    try: return value(select(db, table, [pkey], row, limit=1, recover=True))
769 14 aaronmk
    except StopIteration:
770 40 aaronmk
        if not create: raise
771 471 aaronmk
        return put(db, table, row, pkey, row_ct_ref) # insert new row
772 2078 aaronmk
773 2134 aaronmk
def put_table(db, out_table, in_tables, mapping, limit=None, start=0,
774
    row_ct_ref=None, table_is_esc=False):
775 2078 aaronmk
    '''Recovers from errors.
776
    Only works under PostgreSQL (uses INSERT RETURNING).
777 2131 aaronmk
    @param in_tables The main input table to select from, followed by a list of
778
        tables to join with it using the main input table's pkey
779 2133 aaronmk
    @return (table, col) Where the pkeys (from INSERT RETURNING) are made
780 2078 aaronmk
        available
781
    '''
782 2251 aaronmk
    temp_prefix = clean_name(out_table)
783
    pkeys_ref = [temp_prefix+'_pkeys']
784 2131 aaronmk
785 2132 aaronmk
    # Join together input tables
786 2131 aaronmk
    in_tables = in_tables[:] # don't modify input!
787
    in_tables0 = in_tables.pop(0) # first table is separate
788
    in_pkey = pkey(db, in_tables0, recover=True, table_is_esc=table_is_esc)
789 2236 aaronmk
    in_pkey_col = sql_gen.col2sql_gen(in_pkey, in_tables0, table_is_esc)
790 2178 aaronmk
    insert_joins = [in_tables0]+[(t, {in_pkey: join_using}) for t in in_tables]
791 2131 aaronmk
792 2142 aaronmk
    out_pkey = pkey(db, out_table, recover=True, table_is_esc=table_is_esc)
793 2236 aaronmk
    out_pkey_col = sql_gen.col2sql_gen(out_pkey, out_table, table_is_esc)
794 2142 aaronmk
795 2236 aaronmk
    pkeys = [in_pkey, out_pkey]
796
    pkeys_cols = [in_pkey_col, out_pkey_col]
797
798 2201 aaronmk
    pkeys_table_exists_ref = [False]
799
    def run_query_into_pkeys(query, params):
800
        if pkeys_table_exists_ref[0]:
801 2236 aaronmk
            insert_select(db, pkeys_ref[0], pkeys, query, params)
802 2201 aaronmk
        else:
803
            run_query_into(db, query, params, into_ref=pkeys_ref)
804
            pkeys_table_exists_ref[0] = True
805
806 2208 aaronmk
    conds = {}
807 2233 aaronmk
    distinct_on = []
808 2207 aaronmk
    def mk_main_select(cols):
809 2208 aaronmk
        return mk_select(db, insert_joins, cols, conds, distinct_on,
810 2227 aaronmk
            order_by=None, limit=limit, start=start, table_is_esc=table_is_esc,
811
            default_table=in_tables0)
812 2132 aaronmk
813 2245 aaronmk
    def log_ignore(in_col, value):
814
        db.log_debug('Ignoring rows with '+in_col+' = '+value)
815
816 2206 aaronmk
    # Do inserts and selects
817 2257 aaronmk
    join_cols = {}
818
    filter_join_added = False
819 2251 aaronmk
    out_pkeys_ref = [temp_prefix+'_out_pkeys']
820 2252 aaronmk
    in_pkeys_ref = [temp_prefix+'_in_pkeys']
821 2206 aaronmk
    while True:
822
        try:
823 2257 aaronmk
            if join_cols != {}:
824
                select_joins = insert_joins + [(out_table, join_cols)]
825
                db.log_debug('Getting pkeys of already existing rows')
826
                run_query_into_pkeys(*mk_select(db, select_joins, pkeys_cols,
827
                    order_by=None, start=0, table_is_esc=table_is_esc))
828
829
                # Prepare to insert new rows
830
                input_join_cols = join_cols.values()
831
                distinct_on = filter(util.is_str, input_join_cols)
832
                if not filter_join_added:
833
                    insert_joins.append((pkeys_ref[0], {in_pkey: filter_out}))
834
                    filter_join_added = True
835
836 2253 aaronmk
            db.log_debug('Inserting new rows')
837 2206 aaronmk
            cur = insert_select(db, out_table, mapping.keys(),
838 2207 aaronmk
                *mk_main_select(mapping.values()), returning=out_pkey,
839 2206 aaronmk
                into_ref=out_pkeys_ref, recover=True, table_is_esc=table_is_esc)
840
            if row_ct_ref != None and cur.rowcount >= 0:
841
                row_ct_ref[0] += cur.rowcount
842
                add_row_num(db, out_pkeys_ref[0]) # for joining with input pkeys
843
844 2253 aaronmk
            db.log_debug('Getting input pkeys corresponding to rows in insert')
845 2207 aaronmk
            run_query_into(db, *mk_main_select([in_pkey]),
846
                into_ref=in_pkeys_ref)
847 2206 aaronmk
            add_row_num(db, in_pkeys_ref[0]) # for joining with output pkeys
848
849 2253 aaronmk
            db.log_debug('Joining together output and input pkeys')
850 2206 aaronmk
            run_query_into_pkeys(*mk_select(db, [in_pkeys_ref[0],
851 2236 aaronmk
                (out_pkeys_ref[0], {row_num_col: join_using})], pkeys, start=0))
852 2206 aaronmk
853
            break # insert successful
854
        except DuplicateKeyException, e:
855 2258 aaronmk
            old_join_cols = join_cols.copy()
856 2257 aaronmk
            join_cols.update(util.dict_subset_right_join(mapping, e.cols))
857 2245 aaronmk
            db.log_debug('Ignoring existing rows, comparing on '+str(join_cols))
858 2258 aaronmk
            assert join_cols != old_join_cols # avoid infinite loops
859 2206 aaronmk
            # rerun loop with additional constraints
860 2230 aaronmk
        except NullValueException, e:
861
            out_col, = e.cols
862
            try: in_col = mapping[out_col]
863
            except KeyError: # no mapping for missing col, so every row invalid
864 2245 aaronmk
                db.log_debug('Missing mapping for '+out_col)
865 2230 aaronmk
                run_query_into_pkeys(*mk_select(db, insert_joins,
866 2236 aaronmk
                    [in_pkey_col, sql_gen.NamedCode(out_pkey, None)],
867 2230 aaronmk
                    order_by=None, start=0, table_is_esc=table_is_esc))
868
                break
869
            else:
870 2245 aaronmk
                log_ignore(in_col, 'NULL')
871 2230 aaronmk
                conds[in_col] = sql_gen.CompareCond(None, '!=')
872
                # rerun loop with additional constraints
873 2243 aaronmk
        except FunctionValueException, e:
874
            assert e.name == sql_gen.unescape_table(out_table)
875
            out_col = 'value' # assume function param was named "value"
876
            in_col = mapping[out_col]
877 2245 aaronmk
            value = e.value
878
            log_ignore(in_col, value)
879
            conds[in_col] = sql_gen.CompareCond(value, '!=')
880 2243 aaronmk
            # rerun loop with additional constraints
881 2132 aaronmk
882 2154 aaronmk
    return (pkeys_ref[0], out_pkey)
883 2115 aaronmk
884
##### Data cleanup
885
886
def cleanup_table(db, table, cols, table_is_esc=False):
887
    def esc_name_(name): return esc_name(db, name)
888
889
    if not table_is_esc: check_name(table)
890
    cols = map(esc_name_, cols)
891
892
    run_query(db, 'UPDATE '+table+' SET\n'+(',\n'.join(('\n'+col
893
        +' = nullif(nullif(trim(both from '+col+"), %(null0)s), %(null1)s)"
894
            for col in cols))),
895
        dict(null0='', null1=r'\N'))