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 2199 aaronmk
distinct_on_all = object() # tells mk_select() to SELECT DISTINCT ON all columns
417
418 2233 aaronmk
def mk_select(db, tables, fields=None, conds=None, distinct_on=[], limit=None,
419 2293 aaronmk
    start=None, order_by=order_by_pkey, default_table=None):
420 1981 aaronmk
    '''
421 2121 aaronmk
    @param tables The single table to select from, or a list of tables to join
422 2280 aaronmk
        together, with tables after the first being sql_gen.Join objects
423 1981 aaronmk
    @param fields Use None to select all fields in the table
424 2218 aaronmk
    @param conds WHERE conditions: dict(compare_left_side=compare_right_side):
425
        * compare_left_side: Code|str (for col name)
426
        * compare_right_side: ValueCond|literal value
427 2199 aaronmk
    @param distinct_on The columns to SELECT DISTINCT ON, or distinct_on_all to
428
        use all columns
429 2054 aaronmk
    @return tuple(query, params)
430 1981 aaronmk
    '''
431 2121 aaronmk
    if not lists.is_seq(tables): tables = [tables]
432 2141 aaronmk
    tables = list(tables) # don't modify input! (list() copies input)
433 2121 aaronmk
    table0 = tables.pop(0) # first table is separate
434
435 1135 aaronmk
    if conds == None: conds = {}
436 135 aaronmk
    assert limit == None or type(limit) == int
437 865 aaronmk
    assert start == None or type(start) == int
438 2293 aaronmk
    table0 = sql_gen.as_Table(table0)
439 2291 aaronmk
    if order_by is order_by_pkey: order_by = pkey(db, table0, recover=True)
440 865 aaronmk
441 2056 aaronmk
    params = []
442
443 2227 aaronmk
    def parse_col(field):
444 2056 aaronmk
        '''Parses fields'''
445 2286 aaronmk
        return sql_gen.as_Col(field, default_table).to_str(db)
446 2056 aaronmk
447 2200 aaronmk
    query = 'SELECT'
448
449
    # DISTINCT ON columns
450 2233 aaronmk
    if distinct_on != []:
451 2200 aaronmk
        query += ' DISTINCT'
452 2254 aaronmk
        if distinct_on is not distinct_on_all:
453 2200 aaronmk
            query += ' ON ('+(', '.join(map(parse_col, distinct_on)))+')'
454
455
    # Columns
456
    query += ' '
457 1135 aaronmk
    if fields == None: query += '*'
458 2056 aaronmk
    else: query += ', '.join(map(parse_col, fields))
459 2200 aaronmk
460
    # Main table
461 2271 aaronmk
    query += ' FROM '+table0.to_str(db)
462 865 aaronmk
463 2122 aaronmk
    # Add joins
464 2271 aaronmk
    left_table = table0
465 2263 aaronmk
    for join_ in tables:
466 2277 aaronmk
        # Parse special values
467 2263 aaronmk
        table = join_.table
468 2267 aaronmk
        if join_.type_ is sql_gen.filter_out:
469 2277 aaronmk
            conds[sql_gen.Col(pkey(db, table), table)] = None # filter no match
470 2238 aaronmk
471 2277 aaronmk
        query += ' '+join_.to_str(db, left_table)
472 2122 aaronmk
473
        left_table = table
474
475 865 aaronmk
    missing = True
476 89 aaronmk
    if conds != {}:
477 2284 aaronmk
        query += ' WHERE '+(' AND '.join((sql_gen.as_ValueCond(r).to_str(db, l)
478
            for l, r in conds.iteritems())))
479 2056 aaronmk
        params += conds.values()
480 865 aaronmk
        missing = False
481 2227 aaronmk
    if order_by != None:
482 2285 aaronmk
        query += ' ORDER BY '+sql_gen.as_Col(order_by, table0).to_str(db)
483 865 aaronmk
    if limit != None: query += ' LIMIT '+str(limit); missing = False
484
    if start != None:
485
        if start != 0: query += ' OFFSET '+str(start)
486
        missing = False
487
    if missing: warnings.warn(DbWarning(
488
        'SELECT statement missing a WHERE, LIMIT, or OFFSET clause: '+query))
489
490 2056 aaronmk
    return (query, params)
491 11 aaronmk
492 2054 aaronmk
def select(db, *args, **kw_args):
493
    '''For params, see mk_select() and run_query()'''
494
    recover = kw_args.pop('recover', None)
495
    cacheable = kw_args.pop('cacheable', True)
496
497
    query, params = mk_select(db, *args, **kw_args)
498
    return run_query(db, query, params, recover, cacheable)
499
500 2066 aaronmk
def mk_insert_select(db, table, cols=None, select_query=None, params=None,
501 2292 aaronmk
    returning=None, embeddable=False):
502 1960 aaronmk
    '''
503
    @param returning str|None An inserted column (such as pkey) to return
504 2070 aaronmk
    @param embeddable Whether the query should be embeddable as a nested SELECT.
505 2073 aaronmk
        Warning: If you set this and cacheable=True when the query is run, the
506
        query will be fully cached, not just if it raises an exception.
507 1960 aaronmk
    '''
508 2063 aaronmk
    if select_query == None: select_query = 'DEFAULT VALUES'
509
    if cols == []: cols = None # no cols (all defaults) = unknown col names
510 2292 aaronmk
    table = sql_gen.as_Table(table).to_str(db)
511 2063 aaronmk
512
    # Build query
513
    query = 'INSERT INTO '+table
514
    if cols != None:
515
        map(check_name, cols)
516
        query += ' ('+', '.join(cols)+')'
517
    query += ' '+select_query
518
519
    if returning != None:
520
        check_name(returning)
521
        query += ' RETURNING '+returning
522
523 2070 aaronmk
    if embeddable:
524
        # Create function
525 2189 aaronmk
        function_name = '_'.join(map(clean_name, ['insert', table] + cols))
526 2070 aaronmk
        return_type = 'SETOF '+table+'.'+returning+'%TYPE'
527 2189 aaronmk
        while True:
528
            try:
529 2194 aaronmk
                function = function_name
530
                if not db.debug: function = 'pg_temp.'+function
531
532 2189 aaronmk
                function_query = '''\
533
CREATE FUNCTION '''+function+'''() RETURNS '''+return_type+'''
534 2070 aaronmk
    LANGUAGE sql
535
    AS $$'''+mogrify(db, query, params)+''';$$;
536
'''
537 2189 aaronmk
                run_query(db, function_query, recover=True, cacheable=True)
538
                break # this version was successful
539
            except DuplicateFunctionException, e:
540
                function_name = next_version(function_name)
541
                # try again with next version of name
542 2070 aaronmk
543
        # Return query that uses function
544 2271 aaronmk
        return mk_select(db, sql_gen.CustomCode(function+'() AS f ('+returning
545
            +')'), start=0, order_by=None) # AS clause requires function alias
546 2070 aaronmk
547 2066 aaronmk
    return (query, params)
548
549
def insert_select(db, *args, **kw_args):
550 2085 aaronmk
    '''For params, see mk_insert_select() and run_query_into()
551 2152 aaronmk
    @param into_ref List with name of temp table to place RETURNING values in
552 2072 aaronmk
    '''
553 2151 aaronmk
    into_ref = kw_args.pop('into_ref', None)
554
    if into_ref != None: kw_args['embeddable'] = True
555 2066 aaronmk
    recover = kw_args.pop('recover', None)
556
    cacheable = kw_args.pop('cacheable', True)
557
558
    query, params = mk_insert_select(db, *args, **kw_args)
559 2153 aaronmk
    return run_query_into(db, query, params, into_ref, recover=recover,
560
        cacheable=cacheable)
561 2063 aaronmk
562 2066 aaronmk
default = object() # tells insert() to use the default value for a column
563
564 2063 aaronmk
def insert(db, table, row, *args, **kw_args):
565 2085 aaronmk
    '''For params, see insert_select()'''
566 1960 aaronmk
    if lists.is_seq(row): cols = None
567
    else:
568
        cols = row.keys()
569
        row = row.values()
570
    row = list(row) # ensure that "!= []" works
571
572 1961 aaronmk
    # Check for special values
573
    labels = []
574
    values = []
575
    for value in row:
576 2254 aaronmk
        if value is default: labels.append('DEFAULT')
577 1961 aaronmk
        else:
578
            labels.append('%s')
579
            values.append(value)
580
581
    # Build query
582 2063 aaronmk
    if values != []: query = ' VALUES ('+(', '.join(labels))+')'
583
    else: query = None
584 1554 aaronmk
585 2064 aaronmk
    return insert_select(db, table, cols, query, values, *args, **kw_args)
586 11 aaronmk
587 135 aaronmk
def last_insert_id(db):
588 1849 aaronmk
    module = util.root_module(db.db)
589 135 aaronmk
    if module == 'psycopg2': return value(run_query(db, 'SELECT lastval()'))
590
    elif module == 'MySQLdb': return db.insert_id()
591
    else: return None
592 13 aaronmk
593 1968 aaronmk
def truncate(db, table, schema='public'):
594
    return run_query(db, 'TRUNCATE '+qual_name(db, schema, table)+' CASCADE')
595 832 aaronmk
596
##### Database structure queries
597
598 2291 aaronmk
def pkey(db, table, recover=None):
599 832 aaronmk
    '''Assumed to be first column in table'''
600 2291 aaronmk
    return col_names(select(db, table, limit=0, order_by=None,
601
        recover=recover)).next()
602 832 aaronmk
603 853 aaronmk
def index_cols(db, table, index):
604
    '''Can also use this for UNIQUE constraints, because a UNIQUE index is
605
    automatically created. When you don't know whether something is a UNIQUE
606
    constraint or a UNIQUE index, use this function.'''
607
    check_name(table)
608
    check_name(index)
609 1909 aaronmk
    module = util.root_module(db.db)
610
    if module == 'psycopg2':
611
        return list(values(run_query(db, '''\
612 853 aaronmk
SELECT attname
613 866 aaronmk
FROM
614
(
615
        SELECT attnum, attname
616
        FROM pg_index
617
        JOIN pg_class index ON index.oid = indexrelid
618
        JOIN pg_class table_ ON table_.oid = indrelid
619
        JOIN pg_attribute ON attrelid = indrelid AND attnum = ANY (indkey)
620
        WHERE
621
            table_.relname = %(table)s
622
            AND index.relname = %(index)s
623
    UNION
624
        SELECT attnum, attname
625
        FROM
626
        (
627
            SELECT
628
                indrelid
629
                , (regexp_matches(indexprs, E':varattno (\\\\d+)', 'g'))[1]::int
630
                    AS indkey
631
            FROM pg_index
632
            JOIN pg_class index ON index.oid = indexrelid
633
            JOIN pg_class table_ ON table_.oid = indrelid
634
            WHERE
635
                table_.relname = %(table)s
636
                AND index.relname = %(index)s
637
        ) s
638
        JOIN pg_attribute ON attrelid = indrelid AND attnum = indkey
639
) s
640 853 aaronmk
ORDER BY attnum
641
''',
642 1909 aaronmk
            {'table': table, 'index': index}, cacheable=True)))
643
    else: raise NotImplementedError("Can't list index columns for "+module+
644
        ' database')
645 853 aaronmk
646 464 aaronmk
def constraint_cols(db, table, constraint):
647
    check_name(table)
648
    check_name(constraint)
649 1849 aaronmk
    module = util.root_module(db.db)
650 464 aaronmk
    if module == 'psycopg2':
651
        return list(values(run_query(db, '''\
652
SELECT attname
653
FROM pg_constraint
654
JOIN pg_class ON pg_class.oid = conrelid
655
JOIN pg_attribute ON attrelid = conrelid AND attnum = ANY (conkey)
656
WHERE
657
    relname = %(table)s
658
    AND conname = %(constraint)s
659
ORDER BY attnum
660
''',
661
            {'table': table, 'constraint': constraint})))
662
    else: raise NotImplementedError("Can't list constraint columns for "+module+
663
        ' database')
664
665 2096 aaronmk
row_num_col = '_row_num'
666
667 2086 aaronmk
def add_row_num(db, table):
668 2117 aaronmk
    '''Adds a row number column to a table. Its name is in row_num_col. It will
669
    be the primary key.'''
670 2086 aaronmk
    check_name(table)
671 2096 aaronmk
    run_query(db, 'ALTER TABLE '+table+' ADD COLUMN '+row_num_col
672 2117 aaronmk
        +' serial NOT NULL PRIMARY KEY')
673 2086 aaronmk
674 1968 aaronmk
def tables(db, schema='public', table_like='%'):
675 1849 aaronmk
    module = util.root_module(db.db)
676 1968 aaronmk
    params = {'schema': schema, 'table_like': table_like}
677 832 aaronmk
    if module == 'psycopg2':
678 1968 aaronmk
        return values(run_query(db, '''\
679
SELECT tablename
680
FROM pg_tables
681
WHERE
682
    schemaname = %(schema)s
683
    AND tablename LIKE %(table_like)s
684
ORDER BY tablename
685
''',
686
            params, cacheable=True))
687
    elif module == 'MySQLdb':
688
        return values(run_query(db, 'SHOW TABLES LIKE %(table_like)s', params,
689
            cacheable=True))
690 832 aaronmk
    else: raise NotImplementedError("Can't list tables for "+module+' database')
691 830 aaronmk
692 833 aaronmk
##### Database management
693
694 1968 aaronmk
def empty_db(db, schema='public', **kw_args):
695
    '''For kw_args, see tables()'''
696
    for table in tables(db, schema, **kw_args): truncate(db, table, schema)
697 833 aaronmk
698 832 aaronmk
##### Heuristic queries
699
700 2104 aaronmk
def put(db, table, row, pkey_=None, row_ct_ref=None):
701 1554 aaronmk
    '''Recovers from errors.
702 2077 aaronmk
    Only works under PostgreSQL (uses INSERT RETURNING).
703
    '''
704 2104 aaronmk
    if pkey_ == None: pkey_ = pkey(db, table, recover=True)
705
706 471 aaronmk
    try:
707 2149 aaronmk
        cur = insert(db, table, row, pkey_, recover=True)
708 1554 aaronmk
        if row_ct_ref != None and cur.rowcount >= 0:
709
            row_ct_ref[0] += cur.rowcount
710
        return value(cur)
711 471 aaronmk
    except DuplicateKeyException, e:
712 2104 aaronmk
        return value(select(db, table, [pkey_],
713 1069 aaronmk
            util.dict_subset_right_join(row, e.cols), recover=True))
714 471 aaronmk
715 473 aaronmk
def get(db, table, row, pkey, row_ct_ref=None, create=False):
716 830 aaronmk
    '''Recovers from errors'''
717 2209 aaronmk
    try: return value(select(db, table, [pkey], row, limit=1, recover=True))
718 14 aaronmk
    except StopIteration:
719 40 aaronmk
        if not create: raise
720 471 aaronmk
        return put(db, table, row, pkey, row_ct_ref) # insert new row
721 2078 aaronmk
722 2134 aaronmk
def put_table(db, out_table, in_tables, mapping, limit=None, start=0,
723 2279 aaronmk
    row_ct_ref=None):
724 2078 aaronmk
    '''Recovers from errors.
725
    Only works under PostgreSQL (uses INSERT RETURNING).
726 2131 aaronmk
    @param in_tables The main input table to select from, followed by a list of
727
        tables to join with it using the main input table's pkey
728 2279 aaronmk
    @return sql_gen.Col Where the pkeys (from INSERT RETURNING) are made
729 2078 aaronmk
        available
730
    '''
731 2251 aaronmk
    temp_prefix = clean_name(out_table)
732
    pkeys_ref = [temp_prefix+'_pkeys']
733 2131 aaronmk
734 2132 aaronmk
    # Join together input tables
735 2131 aaronmk
    in_tables = in_tables[:] # don't modify input!
736
    in_tables0 = in_tables.pop(0) # first table is separate
737 2279 aaronmk
    in_pkey = pkey(db, in_tables0, recover=True)
738 2285 aaronmk
    in_pkey_col = sql_gen.as_Col(in_pkey, in_tables0)
739 2278 aaronmk
    insert_joins = [in_tables0]+[sql_gen.Join(v, {in_pkey: sql_gen.join_using})
740
        for v in in_tables]
741 2131 aaronmk
742 2279 aaronmk
    out_pkey = pkey(db, out_table, recover=True)
743 2285 aaronmk
    out_pkey_col = sql_gen.as_Col(out_pkey, out_table)
744 2142 aaronmk
745 2236 aaronmk
    pkeys = [in_pkey, out_pkey]
746
    pkeys_cols = [in_pkey_col, out_pkey_col]
747
748 2201 aaronmk
    pkeys_table_exists_ref = [False]
749
    def run_query_into_pkeys(query, params):
750
        if pkeys_table_exists_ref[0]:
751 2236 aaronmk
            insert_select(db, pkeys_ref[0], pkeys, query, params)
752 2201 aaronmk
        else:
753
            run_query_into(db, query, params, into_ref=pkeys_ref)
754
            pkeys_table_exists_ref[0] = True
755
756 2208 aaronmk
    conds = {}
757 2233 aaronmk
    distinct_on = []
758 2207 aaronmk
    def mk_main_select(cols):
759 2208 aaronmk
        return mk_select(db, insert_joins, cols, conds, distinct_on,
760 2297 aaronmk
            limit=limit, start=start, default_table=in_tables0)
761 2132 aaronmk
762 2245 aaronmk
    def log_ignore(in_col, value):
763
        db.log_debug('Ignoring rows with '+in_col+' = '+value)
764
765 2206 aaronmk
    # Do inserts and selects
766 2257 aaronmk
    join_cols = {}
767
    filter_join_added = False
768 2251 aaronmk
    out_pkeys_ref = [temp_prefix+'_out_pkeys']
769 2252 aaronmk
    in_pkeys_ref = [temp_prefix+'_in_pkeys']
770 2206 aaronmk
    while True:
771 2303 aaronmk
        has_joins = join_cols != {}
772
773
        if has_joins:
774
            select_joins = insert_joins+[sql_gen.Join(out_table, join_cols)]
775
            db.log_debug('Getting pkeys of already existing rows')
776
            run_query_into_pkeys(*mk_select(db, select_joins, pkeys_cols,
777
                start=0))
778
779
            # Prepare to insert new rows
780
            input_join_cols = join_cols.values()
781
            distinct_on = filter(util.is_str, input_join_cols)
782
            if not filter_join_added:
783
                insert_joins.append(sql_gen.Join(pkeys_ref[0],
784
                    {in_pkey: in_pkey}, sql_gen.filter_out))
785
                filter_join_added = True
786
787 2206 aaronmk
        try:
788 2253 aaronmk
            db.log_debug('Inserting new rows')
789 2206 aaronmk
            cur = insert_select(db, out_table, mapping.keys(),
790 2207 aaronmk
                *mk_main_select(mapping.values()), returning=out_pkey,
791 2279 aaronmk
                into_ref=out_pkeys_ref, recover=True)
792 2206 aaronmk
        except DuplicateKeyException, e:
793 2258 aaronmk
            old_join_cols = join_cols.copy()
794 2257 aaronmk
            join_cols.update(util.dict_subset_right_join(mapping, e.cols))
795 2245 aaronmk
            db.log_debug('Ignoring existing rows, comparing on '+str(join_cols))
796 2258 aaronmk
            assert join_cols != old_join_cols # avoid infinite loops
797 2206 aaronmk
            # rerun loop with additional constraints
798 2230 aaronmk
        except NullValueException, e:
799
            out_col, = e.cols
800
            try: in_col = mapping[out_col]
801
            except KeyError: # no mapping for missing col, so every row invalid
802 2245 aaronmk
                db.log_debug('Missing mapping for '+out_col)
803 2230 aaronmk
                run_query_into_pkeys(*mk_select(db, insert_joins,
804 2297 aaronmk
                    [in_pkey_col, sql_gen.NamedCode(out_pkey, None)], start=0))
805 2230 aaronmk
                break
806
            else:
807 2245 aaronmk
                log_ignore(in_col, 'NULL')
808 2230 aaronmk
                conds[in_col] = sql_gen.CompareCond(None, '!=')
809
                # rerun loop with additional constraints
810 2243 aaronmk
        except FunctionValueException, e:
811
            assert e.name == sql_gen.unescape_table(out_table)
812
            out_col = 'value' # assume function param was named "value"
813
            in_col = mapping[out_col]
814 2245 aaronmk
            value = e.value
815
            log_ignore(in_col, value)
816
            conds[in_col] = sql_gen.CompareCond(value, '!=')
817 2243 aaronmk
            # rerun loop with additional constraints
818 2303 aaronmk
        else:
819
            if row_ct_ref != None and cur.rowcount >= 0:
820
                row_ct_ref[0] += cur.rowcount
821
                add_row_num(db, out_pkeys_ref[0]) # for joining with input pkeys
822
823
            db.log_debug('Getting input pkeys corresponding to rows in insert')
824
            run_query_into(db, *mk_main_select([in_pkey]),
825
                into_ref=in_pkeys_ref)
826
            add_row_num(db, in_pkeys_ref[0]) # for joining with output pkeys
827
828
            db.log_debug('Joining together output and input pkeys')
829
            run_query_into_pkeys(*mk_select(db, [in_pkeys_ref[0],
830
                sql_gen.Join(out_pkeys_ref[0],
831
                {row_num_col: sql_gen.join_using})], pkeys, start=0))
832
833
            break # insert successful
834 2132 aaronmk
835 2268 aaronmk
    return sql_gen.Col(out_pkey, pkeys_ref[0])
836 2115 aaronmk
837
##### Data cleanup
838
839 2290 aaronmk
def cleanup_table(db, table, cols):
840 2115 aaronmk
    def esc_name_(name): return esc_name(db, name)
841
842 2290 aaronmk
    table = sql_gen.as_Table(table).to_str(db)
843 2115 aaronmk
    cols = map(esc_name_, cols)
844
845
    run_query(db, 'UPDATE '+table+' SET\n'+(',\n'.join(('\n'+col
846
        +' = nullif(nullif(trim(both from '+col+"), %(null0)s), %(null1)s)"
847
            for col in cols))),
848
        dict(null0='', null1=r'\N'))