# Database access

import copy

import operator

import re

import warnings

import exc

import dicts

import iters

import lists

from Proxy import Proxy

import rand

import sql_gen

import strings

import util

##### Exceptions

def get_cur_query(cur, input_query=None, input_params=None):

    raw_query = None

    if hasattr(cur, 'query'): raw_query = cur.query

    elif hasattr(cur, '_last_executed'): raw_query = cur._last_executed

    if raw_query != None: return raw_query

    else: return '[input] '+strings.ustr(input_query)+' % '+repr(input_params)

def _add_cursor_info(e, *args, **kw_args):

    '''For params, see get_cur_query()'''

    exc.add_msg(e, 'query: '+strings.ustr(get_cur_query(*args, **kw_args)))

class DbException(exc.ExceptionWithCause):

    def __init__(self, msg, cause=None, cur=None):

        exc.ExceptionWithCause.__init__(self, msg, cause, cause_newline=True)

        if cur != None: _add_cursor_info(self, cur)

class ExceptionWithName(DbException):

    def __init__(self, name, cause=None):

        DbException.__init__(self, 'for name: '+strings.as_tt(str(name)), cause)

        self.name = name

class ExceptionWithNameValue(DbException):

    def __init__(self, name, value, cause=None):

        DbException.__init__(self, 'for name: '+strings.as_tt(str(name))

            +'; value: '+strings.as_tt(repr(value)), cause)

        self.name = name

        self.value = value

class ConstraintException(DbException):

    def __init__(self, name, cols, cause=None):

        DbException.__init__(self, 'Violated '+strings.as_tt(name)

            +' constraint on columns: '+strings.as_tt(', '.join(cols)), cause)

        self.name = name

        self.cols = cols

class MissingCastException(DbException):

    def __init__(self, type_, col, cause=None):

        DbException.__init__(self, 'Missing cast to type '+strings.as_tt(type_)

            +' on column: '+strings.as_tt(col), cause)

        self.type = type_

        self.col = col

class NameException(DbException): pass

class DuplicateKeyException(ConstraintException): pass

class NullValueException(ConstraintException): pass

class FunctionValueException(ExceptionWithNameValue): pass

class DuplicateTableException(ExceptionWithName): pass

class DuplicateFunctionException(ExceptionWithName): pass

class EmptyRowException(DbException): pass

##### Warnings

class DbWarning(UserWarning): pass

##### Result retrieval

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

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

def consume_rows(cur):

    '''Used to fetch all rows so result will be cached'''

    iters.consume_iter(rows(cur))

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

def row(cur):

    row_ = next_row(cur)

    consume_rows(cur)

    return row_

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

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

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

def value_or_none(cur):

    try: return value(cur)

    except StopIteration: return None

##### Escaping

def esc_name_by_module(module, name):

    if module == 'psycopg2' or module == None: quote = '"'

    elif module == 'MySQLdb': quote = '`'

    else: raise NotImplementedError("Can't escape name for "+module+' database')

    return sql_gen.esc_name(name, quote)

def esc_name_by_engine(engine, name, **kw_args):

    return esc_name_by_module(db_engines[engine][0], name, **kw_args)

def esc_name(db, name, **kw_args):

    return esc_name_by_module(util.root_module(db.db), name, **kw_args)

def qual_name(db, schema, table):

    def esc_name_(name): return esc_name(db, name)

    table = esc_name_(table)

    if schema != None: return esc_name_(schema)+'.'+table

    else: return table

##### Database connections

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

db_engines = {

    'MySQL': ('MySQLdb', {'password': 'passwd', 'database': 'db'}),

    'PostgreSQL': ('psycopg2', {}),

}

DatabaseErrors_set = set([DbException])

DatabaseErrors = tuple(DatabaseErrors_set)

def _add_module(module):

    DatabaseErrors_set.add(module.DatabaseError)

    global DatabaseErrors

    DatabaseErrors = tuple(DatabaseErrors_set)

def db_config_str(db_config):

    return db_config['engine']+' database '+db_config['database']

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

log_debug_none = lambda msg, level=2: None

class DbConn:

    def __init__(self, db_config, serializable=True, autocommit=False,

        caching=True, log_debug=log_debug_none):

        self.db_config = db_config

        self.serializable = serializable

        self.autocommit = autocommit

        self.caching = caching

        self.log_debug = log_debug

        self.debug = log_debug != log_debug_none

        self.__db = None

        self.query_results = {}

        self._savepoint = 0

        self._notices_seen = set()

    def __getattr__(self, name):

        if name == '__dict__': raise Exception('getting __dict__')

        if name == 'db': return self._db()

        else: raise AttributeError()

    def __getstate__(self):

        state = copy.copy(self.__dict__) # shallow copy

        state['log_debug'] = None # don't pickle the debug callback

        state['_DbConn__db'] = None # don't pickle the connection

        return state

    def connected(self): return self.__db != None

    def _db(self):

        if self.__db == None:

            # Process db_config

            db_config = self.db_config.copy() # don't modify input!

            schemas = db_config.pop('schemas', None)

            module_name, mappings = db_engines[db_config.pop('engine')]

            module = __import__(module_name)

            _add_module(module)

            for orig, new in mappings.iteritems():

                try: util.rename_key(db_config, orig, new)

                except KeyError: pass

            # Connect

            self.__db = module.connect(**db_config)

            # Configure connection

            if self.serializable and not self.autocommit: self.run_query(

                'SET TRANSACTION ISOLATION LEVEL SERIALIZABLE', log_level=4)

            if schemas != None:

                schemas_ = ''.join((esc_name(self, s)+', '

                    for s in schemas.split(',')))

                self.run_query("SELECT set_config('search_path', "

                    +self.esc_value(schemas_)

                    +" || current_setting('search_path'), false)", log_level=3)

        return self.__db

    class DbCursor(Proxy):

        def __init__(self, outer):

            Proxy.__init__(self, outer.db.cursor())

            self.outer = outer

            self.query_results = outer.query_results

            self.query_lookup = None

            self.result = []

        def execute(self, query, params=None):

            assert params == None or params == [] or params == ()

            if params == [] or params == (): params = None

                # None turns off the mogrifier

            self._is_insert = query.startswith('INSERT')

            self.query_lookup = query

            try:

                try:

                    cur = self.inner.execute(query, params)

                    self.outer.do_autocommit()

                finally: self.query = get_cur_query(self.inner, query, params)

            except Exception, e:

                _add_cursor_info(e, self, query, params)

                self.result = e # cache the exception as the result

                self._cache_result()

                raise

            # Fetch all rows so result will be cached

            if self.rowcount == 0 and not self._is_insert: consume_rows(self)

            return cur

        def fetchone(self):

            row = self.inner.fetchone()

            if row != None: self.result.append(row)

            # otherwise, fetched all rows

            else: self._cache_result()

            return row

        def _cache_result(self):

            # For inserts, only cache exceptions since inserts are not

            # idempotent, but an invalid insert will always be invalid

            if self.query_results != None and (not self._is_insert

                or isinstance(self.result, Exception)):

                assert self.query_lookup != None

                self.query_results[self.query_lookup] = self.CacheCursor(

                    util.dict_subset(dicts.AttrsDictView(self),

                    ['query', 'result', 'rowcount', 'description']))

        class CacheCursor:

            def __init__(self, cached_result): self.__dict__ = cached_result

            def execute(self, *args, **kw_args):

                if isinstance(self.result, Exception): raise self.result

                # otherwise, result is a rows list

                self.iter = iter(self.result)

            def fetchone(self):

                try: return self.iter.next()

                except StopIteration: return None

    def esc_value(self, value):

        try: str_ = self.mogrify('%s', [value])

        except NotImplementedError, e:

            module = util.root_module(self.db)

            if module == 'MySQLdb':

                import _mysql

                str_ = _mysql.escape_string(value)

            else: raise e

        return strings.to_unicode(str_)

    def esc_name(self, name): return esc_name(self, name) # calls global func

    def can_mogrify(self):

        module = util.root_module(self.db)

        return module == 'psycopg2'

    def mogrify(self, query, params=None):

        if self.can_mogrify(): return self.db.cursor().mogrify(query, params)

        else: raise NotImplementedError("Can't mogrify query")

    def print_notices(self):

        if hasattr(self.db, 'notices'):

            for msg in self.db.notices:

                if msg not in self._notices_seen:

                    self._notices_seen.add(msg)

                    self.log_debug(msg, level=2)

    def run_query(self, query, cacheable=False, log_level=2,

        debug_msg_ref=None):

        '''

        @param log_ignore_excs The log_level will be increased by 2 if the query

            throws one of these exceptions.

        @param debug_msg_ref If specified, the log message will be returned in

            this instead of being output. This allows you to filter log messages

            depending on the result of the query.

        '''

        assert query != None

        if not self.caching: cacheable = False

        used_cache = False

        def log_msg(query):

            if used_cache: cache_status = 'cache hit'

            elif cacheable: cache_status = 'cache miss'

            else: cache_status = 'non-cacheable'

            return 'DB query: '+cache_status+':\n'+strings.as_code(query, 'SQL')

        try:

            # Get cursor

            if cacheable:

                try:

                    cur = self.query_results[query]

                    used_cache = True

                except KeyError: cur = self.DbCursor(self)

            else: cur = self.db.cursor()

            # Log query

            if self.debug and debug_msg_ref == None: # log before running

                self.log_debug(log_msg(query), log_level)

            # Run query

            cur.execute(query)

        finally:

            self.print_notices()

            if self.debug and debug_msg_ref != None: # return after running

                debug_msg_ref[0] = log_msg(str(get_cur_query(cur, query)))

        return cur

    def is_cached(self, query): return query in self.query_results

    def with_autocommit(self, func, autocommit=True):

        prev_autocommit = self.db.autocommit

        self.db.autocommit = autocommit

        try: return func()

        finally: self.db.autocommit = prev_autocommit

    def with_savepoint(self, func):

        savepoint = 'level_'+str(self._savepoint)

        self.run_query('SAVEPOINT '+savepoint, log_level=4)

        self._savepoint += 1

        try:

            try: return_val = func()

            finally:

                self._savepoint -= 1

                assert self._savepoint >= 0

        except:

            self.run_query('ROLLBACK TO SAVEPOINT '+savepoint, log_level=4)

            raise

        else:

            self.run_query('RELEASE SAVEPOINT '+savepoint, log_level=4)

            self.do_autocommit()

            return return_val

    def do_autocommit(self):

        '''Autocommits if outside savepoint'''

        assert self._savepoint >= 0

        if self.autocommit and self._savepoint == 0:

            self.log_debug('Autocommitting')

            self.db.commit()

    def col_default(self, col):

        table = sql_gen.Table('columns', 'information_schema')

        conds = [('table_name', col.table.name), ('column_name', col.name)]

        schema = col.table.schema

        if schema != None: conds.append(('table_schema', schema))

        return sql_gen.as_Code(value(select(self, table, ['column_default'],

            conds, order_by='table_schema', limit=1, log_level=3)))

            # TODO: order_by search_path schema order

connect = DbConn

##### Querying

def mogrify(db, query, params):

    module = util.root_module(db.db)

    if module == 'psycopg2': return db.db.cursor().mogrify(query, params)

    else: raise NotImplementedError("Can't mogrify query for "+module+

        ' database')

##### Recoverable querying

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

def run_query(db, query, recover=None, cacheable=False, log_level=2,

    log_ignore_excs=None, **kw_args):

    '''For params, see DbConn.run_query()'''

    if recover == None: recover = False

    if log_ignore_excs == None: log_ignore_excs = ()

    log_ignore_excs = tuple(log_ignore_excs)

    debug_msg_ref = None # usually, db.run_query() logs query before running it

    # But if filtering with log_ignore_excs, wait until after exception parsing

    if log_ignore_excs != () or not db.can_mogrify(): debug_msg_ref = [None] 

    try:

        try:

            def run(): return db.run_query(query, cacheable, log_level,

                debug_msg_ref, **kw_args)

            if recover and not db.is_cached(query):

                return with_savepoint(db, run)

            else: return run() # don't need savepoint if cached

        except Exception, e:

            if not recover: raise # need savepoint to run index_cols()

            msg = exc.str_(e)

            match = re.search(r'duplicate key value violates unique constraint '

                r'"((_?[^\W_]+)_.+?)"', msg)

            if match:

                constraint, table = match.groups()

                try: cols = index_cols(db, table, constraint)

                except NotImplementedError: raise e

                else: raise DuplicateKeyException(constraint, cols, e)

            match = re.search(r'null value in column "(.+?)" violates not-null'

                r' constraint', msg)

            if match: raise NullValueException('NOT NULL', [match.group(1)], e)

            match = re.search(r'\b(?:invalid input (?:syntax|value)\b.*?'

                r'|date/time field value out of range): "(.+?)"\n'

                r'(?:(?s).*?)\bfunction "(.+?)"', msg)

            if match:

                value, name = match.groups()

                raise FunctionValueException(name, strings.to_unicode(value), e)

            match = re.search(r'column "(.+?)" is of type (.+?) but expression '

                r'is of type', msg)

            if match:

                col, type_ = match.groups()

                raise MissingCastException(type_, col, e)

            match = re.search(r'relation "(.+?)" already exists', msg)

            if match: raise DuplicateTableException(match.group(1), e)

            match = re.search(r'function "(.+?)" already exists', msg)

            if match: raise DuplicateFunctionException(match.group(1), e)

            raise # no specific exception raised

    except log_ignore_excs:

        log_level += 2

        raise

    finally:

        if debug_msg_ref != None and debug_msg_ref[0] != None:

            db.log_debug(debug_msg_ref[0], log_level)

##### Basic queries

def next_version(name):

    version = 1 # first existing name was version 0

    match = re.match(r'^(.*)#(\d+)$', name)

    if match:

        name, version = match.groups()

        version = int(version)+1

    return sql_gen.add_suffix(name, '#'+str(version))

def run_query_into(db, query, into=None, add_indexes_=False, **kw_args):

    '''Outputs a query to a temp table.

    For params, see run_query().

    '''

    if into == None: return run_query(db, query, **kw_args)

    assert isinstance(into, sql_gen.Table)

    kw_args['recover'] = True

    kw_args.setdefault('log_ignore_excs', (DuplicateTableException,))

    temp = not db.autocommit # tables are permanent in autocommit mode

    # "temporary tables cannot specify a schema name", so remove schema

    if temp: into.schema = None

    # Create table

    while True:

        create_query = 'CREATE'

        if temp: create_query += ' TEMP'

        create_query += ' TABLE '+into.to_str(db)+' AS\n'+query

        try:

            cur = run_query(db, create_query, **kw_args)

                # CREATE TABLE AS sets rowcount to # rows in query

            break

        except DuplicateTableException, e:

            into.name = next_version(into.name)

            # try again with next version of name

    if add_indexes_: add_indexes(db, into)

    return cur

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

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

def mk_select(db, tables, fields=None, conds=None, distinct_on=[], limit=None,

    start=None, order_by=order_by_pkey, default_table=None):

    '''

    @param tables The single table to select from, or a list of tables to join

        together, with tables after the first being sql_gen.Join objects

    @param fields Use None to select all fields in the table

    @param conds WHERE conditions: [(compare_left_side, compare_right_side),...]

        * container can be any iterable type

        * compare_left_side: sql_gen.Code|str (for col name)

        * compare_right_side: sql_gen.ValueCond|literal value

    @param distinct_on The columns to SELECT DISTINCT ON, or distinct_on_all to

        use all columns

    @return query

    '''

    # Parse tables param

    if not lists.is_seq(tables): tables = [tables]

    tables = list(tables) # don't modify input! (list() copies input)

    table0 = sql_gen.as_Table(tables.pop(0)) # first table is separate

    # Parse other params

    if conds == None: conds = []

    elif dicts.is_dict(conds): conds = conds.items()

    conds = list(conds) # don't modify input! (list() copies input)

    assert limit == None or type(limit) == int

    assert start == None or type(start) == int

    if order_by is order_by_pkey:

        if distinct_on != []: order_by = None

        else: order_by = pkey(db, table0, recover=True)

    query = 'SELECT'

    def parse_col(col): return sql_gen.as_Col(col, default_table).to_str(db)

    # DISTINCT ON columns

    if distinct_on != []:

        query += '\nDISTINCT'

        if distinct_on is not distinct_on_all:

            query += ' ON ('+(', '.join(map(parse_col, distinct_on)))+')'

    # Columns

    query += '\n'

    if fields == None: query += '*'

    else:

        assert fields != []

        query += '\n, '.join(map(parse_col, fields))

    # Main table

    query += '\nFROM '+table0.to_str(db)

    # Add joins

    left_table = table0

    for join_ in tables:

        table = join_.table

        # Parse special values

        if join_.type_ is sql_gen.filter_out: # filter no match

            conds.append((sql_gen.Col(table_not_null_col(db, table), table),

                None))

        query += '\n'+join_.to_str(db, left_table)

        left_table = table

    missing = True

    if conds != []:

        if len(conds) == 1: whitespace = ' '

        else: whitespace = '\n'

        query += '\n'+sql_gen.combine_conds([sql_gen.ColValueCond(l, r)

            .to_str(db) for l, r in conds], 'WHERE')

        missing = False

    if order_by != None:

        query += '\nORDER BY '+sql_gen.as_Col(order_by, table0).to_str(db)

    if limit != None: query += '\nLIMIT '+str(limit); missing = False

    if start != None:

        if start != 0: query += '\nOFFSET '+str(start)

        missing = False

    if missing: warnings.warn(DbWarning(

        'SELECT statement missing a WHERE, LIMIT, or OFFSET clause: '+query))

    return query

def select(db, *args, **kw_args):

    '''For params, see mk_select() and run_query()'''

    recover = kw_args.pop('recover', None)

    cacheable = kw_args.pop('cacheable', True)

    log_level = kw_args.pop('log_level', 2)

    return run_query(db, mk_select(db, *args, **kw_args), recover, cacheable,

        log_level=log_level)

def mk_insert_select(db, table, cols=None, select_query=None, returning=None,

    embeddable=False):

    '''

    @param returning str|None An inserted column (such as pkey) to return

    @param embeddable Whether the query should be embeddable as a nested SELECT.

        Warning: If you set this and cacheable=True when the query is run, the

        query will be fully cached, not just if it raises an exception.

    '''

    table = sql_gen.remove_table_rename(sql_gen.as_Table(table))

    if cols == []: cols = None # no cols (all defaults) = unknown col names

    if cols != None:

        cols = [sql_gen.to_name_only_col(v, table).to_str(db) for v in cols]

    if select_query == None: select_query = 'DEFAULT VALUES'

    if returning != None: returning = sql_gen.as_Col(returning, table)

    # Build query

    first_line = 'INSERT INTO '+table.to_str(db)

    query = first_line

    if cols != None: query += '\n('+', '.join(cols)+')'

    query += '\n'+select_query

    if returning != None:

        query += '\nRETURNING '+sql_gen.to_name_only_col(returning).to_str(db)

    if embeddable:

        assert returning != None

        # Create function

        function_name = sql_gen.clean_name(first_line)

        return_type = 'SETOF '+returning.to_str(db)+'%TYPE'

        while True:

            try:

                function = sql_gen.TempFunction(function_name, db.autocommit)

                function_query = '''\

CREATE FUNCTION '''+function.to_str(db)+'''()

RETURNS '''+return_type+'''

LANGUAGE sql

AS $$

'''+query+''';

$$;

'''

                run_query(db, function_query, recover=True, cacheable=True,

                    log_ignore_excs=(DuplicateFunctionException,))

                break # this version was successful

            except DuplicateFunctionException, e:

                function_name = next_version(function_name)

                # try again with next version of name

        # Return query that uses function

        func_table = sql_gen.NamedTable('f', sql_gen.FunctionCall(function),

            [returning]) # AS clause requires function alias

        return mk_select(db, func_table, start=0, order_by=None)

    return query

def insert_select(db, *args, **kw_args):

    '''For params, see mk_insert_select() and run_query_into()

    @param into sql_gen.Table with suggested name of temp table to put RETURNING

        values in

    '''

    into = kw_args.pop('into', None)

    if into != None: kw_args['embeddable'] = True

    recover = kw_args.pop('recover', None)

    cacheable = kw_args.pop('cacheable', True)

    log_level = kw_args.pop('log_level', 2)

    return run_query_into(db, mk_insert_select(db, *args, **kw_args), into,

        recover=recover, cacheable=cacheable, log_level=log_level)

default = sql_gen.default # tells insert() to use the default value for a column

def insert(db, table, row, *args, **kw_args):

    '''For params, see insert_select()'''

    if lists.is_seq(row): cols = None

    else:

        cols = row.keys()

        row = row.values()

    row = list(row) # ensure that "== []" works

    if row == []: query = None

    else: query = sql_gen.Values(row).to_str(db)

    return insert_select(db, table, cols, query, *args, **kw_args)

def mk_update(db, table, changes=None, cond=None):

    '''

    @param changes [(col, new_value),...]

        * container can be any iterable type

        * col: sql_gen.Code|str (for col name)

        * new_value: sql_gen.Code|literal value

    @param cond sql_gen.Code WHERE condition. e.g. use sql_gen.*Cond objects.

    @return str query

    '''

    query = 'UPDATE '+sql_gen.as_Table(table).to_str(db)+'\nSET\n'

    query += ',\n'.join((sql_gen.to_name_only_col(col, table).to_str(db)+' = '

        +sql_gen.as_Value(new_value).to_str(db) for col, new_value in changes))

    if cond != None: query += '\nWHERE\n'+cond.to_str(db)

    return query

def update(db, *args, **kw_args):

    '''For params, see mk_update() and run_query()'''

    recover = kw_args.pop('recover', None)

    return run_query(db, mk_update(db, *args, **kw_args), recover)

def last_insert_id(db):

    module = util.root_module(db.db)

    if module == 'psycopg2': return value(run_query(db, 'SELECT lastval()'))

    elif module == 'MySQLdb': return db.insert_id()

    else: return None

def mk_flatten_mapping(db, into, cols, preserve=[], as_items=False):

    '''Creates a mapping from original column names (which may have collisions)

    to names that will be distinct among the columns' tables.

    This is meant to be used for several tables that are being joined together.

    @param cols The columns to combine. Duplicates will be removed.

    @param into The table for the new columns.

    @param preserve [sql_gen.Col...] Columns not to rename. Note that these

        columns will be included in the mapping even if they are not in cols.

        The tables of the provided Col objects will be changed to into, so make

        copies of them if you want to keep the original tables.

    @param as_items Whether to return a list of dict items instead of a dict

    @return dict(orig_col=new_col, ...)

        * orig_col: sql_gen.Col(orig_col_name, orig_table)

        * new_col: sql_gen.Col(orig_col_name, into)

        * All mappings use the into table so its name can easily be

          changed for all columns at once

    '''

    cols = lists.uniqify(cols)

    items = []

    for col in preserve:

        orig_col = copy.copy(col)

        col.table = into

        items.append((orig_col, col))

    preserve = set(preserve)

    for col in cols:

        if col not in preserve:

            items.append((col, sql_gen.Col(str(col), into, col.srcs)))

    if not as_items: items = dict(items)

    return items

def flatten(db, into, joins, cols, limit=None, start=None, **kw_args):

    '''For params, see mk_flatten_mapping()

    @return See return value of mk_flatten_mapping()

    '''

    items = mk_flatten_mapping(db, into, cols, as_items=True, **kw_args)

    cols = [sql_gen.NamedCol(new.name, old) for old, new in items]

    run_query_into(db, mk_select(db, joins, cols, limit=limit, start=start),

        into=into)

    return dict(items)

def mk_track_data_error(db, errors_table, cols, value, error_code, error):