root/trunk/storage/db.py

"""Base classes and tools for writing database Storage Managers.

Unit type -> [SQL repr ->] DB -> incoming Python value -> Unit type


DATA TYPES
==========
Since Dejavu relies on external database servers for its persistence,
Python datatypes must be converted to column types in the DB. When writing
a StorageManager, you should make sure that your type conversions can handle
at least the following limitations. If possible, implement the type with no
limits. Also, follow UnitProperty.hints['bytes'] where possible. A value
of zero for hints['bytes'] implies no limit. If no value is given, try to
assume no limit, although you may choose whatever default size you wish
(255 is common for strings).
"""

import time
import weakref
import Queue
from types import FunctionType
import dejavu
from dejavu import codewalk, logic, storage

try:
    import cPickle as pickle
except ImportError:
    import pickle

import datetime

try:
    import fixedpoint
except ImportError:
    pass

try:
    # Builtin in Python 2.5?
    decimal
except NameError:
    try:
        # Module in Python 2.3, 2.4
        import decimal
    except ImportError:
        pass

import warnings
import threading


class FieldTypeAdapter(object):
    """For a Python type, return the SQL column type name.
    
    This base class is designed to work out-of-the-box with PostgreSQL 8.
    """
    
    # 1000 is the max precision for NUMERIC columns for PostgreSQL 8.
    # Override in subclasses.
    numeric_max_precision = 1000
    
    def coerce(self, cls, key):
        """coerce(cls, key) -> SQL typename for valuetype."""
        valuetype = cls.property_type(key)
        mod = valuetype.__module__
        if mod == "__builtin__":
            xform = "coerce_%s" % valuetype.__name__
        else:
            xform = "coerce_%s_%s" % (mod, valuetype.__name__)
        xform = xform.replace(".", "_")
        try:
            xform = getattr(self, xform)
        except AttributeError:
            raise TypeError("'%s' is not handled by %s." %
                            (valuetype, self.__class__))
        return xform(cls, key)
    
    def coerce_float(self, cls, key):
        prop = getattr(cls, key)
        bytes = int(prop.hints.get(u'bytes', '8'))
        if bytes < 5:
            # In MySQL, REAL is still probably 8 bytes. Meh.
            return u"REAL"
        else:
            # Python floats are implemented using C doubles;
            # actual precision depends on platform. PostgreSQL
            # DOUBLE is 8 bytes (15 decimal-digit precision).
            return u"DOUBLE PRECISION"
    
    def coerce_str(self, cls, key):
        # The bytes hint shall not reflect the usual 4-byte base for varchar.
        prop = getattr(cls, key)
        bytes = int(prop.hints.get(u'bytes', '0'))
        if bytes:
            return u"VARCHAR(%s)" % bytes
        else:
            # TEXT is not an SQL standard, but it's common.
            return u"TEXT"
    
    def coerce_dict(self, cls, key):
        return self.coerce_str(cls, key)
    def coerce_list(self, cls, key):
        return self.coerce_str(cls, key)
    def coerce_tuple(self, cls, key):
        return self.coerce_str(cls, key)
    def coerce_unicode(self, cls, key):
        return self.coerce_str(cls, key)
    
    def coerce_bool(self, cls, key): return u"BOOLEAN"
    
    def coerce_datetime_datetime(self, cls, key): return u"TIMESTAMP"
    def coerce_datetime_date(self, cls, key): return u"DATE"
    def coerce_datetime_time(self, cls, key): return u"TIME"
    
    # I was seriously disinterested in writing a parser for interval.
    def coerce_datetime_timedelta(self, cls, key):
        return self.coerce_float(cls, key)
    
    def coerce_decimal_Decimal(self, cls, key):
        prop = getattr(cls, key)
        precision = int(prop.hints.get('precision', '0'))
        if precision == 0:
            precision = decimal.getcontext().prec
        if precision > self.numeric_max_precision:
            warnings.warn("Decimal precision %s > maximum %s for %s.%s, "
                          "using %s. Values may be stored incorrectly."
                          % (precision, self.numeric_max_precision,
                             cls.__name__, key, self.__class__.__name__))
            precision = self.numeric_max_precision
        # Assume most people use decimal for money; default scale = 2.
        scale = int(prop.hints.get(u'scale', 2))
        return u"NUMERIC(%s, %s)" % (precision, scale)
    
    def coerce_decimal(self, cls, key):
        # If decimal ever becomes a builtin. Python 2.5?
        return self.coerce_decimal_Decimal(cls, key)
    
    def coerce_fixedpoint_FixedPoint(self, cls, key):
        prop = getattr(cls, key)
        precision = int(prop.hints.get('precision', '0'))
        if precision == 0:
            precision = self.numeric_max_precision
        # Assume most people use decimal for money; default scale = 2.
        scale = int(prop.hints.get(u'scale', 2))
        return u"NUMERIC(%s, %s)" % (precision, scale)
    
    def coerce_long(self, cls, key):
        prop = getattr(cls, key)
        bytes = int(prop.hints.get(u'bytes', 0))
        if bytes <= 4:
            return self.coerce_int(cls, key)
        elif bytes <= 8:
            # BIGINT is usually 8 bytes
            return "BIGINT"
        # Anything larger than 8 bytes, use decimal/numeric.
        if bytes > self.numeric_max_precision:
            warnings.warn("Long bytes %s > maximum %s for %s.%s, "
                          "using %s. Values may be stored incorrectly."
                          % (bytes, self.numeric_max_precision,
                             cls.__name__, key, self.__class__.__name__))
            bytes = self.numeric_max_precision
        return "NUMERIC(%s, 0)" % bytes
    
    def coerce_int(self, cls, key):
        prop = getattr(cls, key)
        bytes = int(prop.hints.get(u'bytes', '4'))
        if bytes == 1:
            return "BOOLEAN"
        elif bytes == 2:
            return "SMALLINT"
##        elif bytes == 3:
##            return "MEDIUMINT"
        else:
            return u"INTEGER"


class AdapterToSQL(object):
    """Coerce Expression constants to SQL.
    
    This base class is designed to work out-of-the-box with PostgreSQL 8.
    """
    
    # Notice these are ordered pairs. Escape \ before introducing new ones.
    escapes = [("'", "''"), ("\\", r"\\")]
    like_escapes = [("%", r"\%"), ("_", r"\_")]
    
    # These are not the same as coerce_bool (which is used on one side of 
    # a comparison). Instead, these are used when the whole (sub)expression
    # is True or False, e.g. "WHERE TRUE", or "WHERE TRUE and 'a'.'b' = 3".
    bool_true = "TRUE"
    bool_false = "FALSE"
    
    def escape_like(self, value):
        """Prepare a string value for use in a LIKE comparison."""
        # Notice we strip leading and trailing quote-marks.
        value = value.strip("'\"")
        for pat, repl in self.like_escapes:
            value = value.replace(pat, repl)
        return value
    
    def coerce(self, value, valuetype=None):
        """coerce(value, valuetype=None) -> value, coerced by valuetype."""
        if valuetype is None:
            valuetype = type(value)
        
        mod = valuetype.__module__
        if mod == "__builtin__":
            xform = "coerce_%s" % valuetype.__name__
        else:
            xform = "coerce_%s_%s" % (mod, valuetype.__name__)
        xform = xform.replace(".", "_")
        try:
            xform = getattr(self, xform)
        except AttributeError:
            raise TypeError("'%s' is not handled by %s." %
                            (valuetype, self.__class__))
        return xform(value)
    
    def tostr(self, value):
        return str(value)
    
    def coerce_NoneType(self, value):
        return "NULL"
    
    def coerce_bool(self, value):
        if value:
            return 'TRUE'
        return 'FALSE'
    
    # The great thing about these 3 date coercers is that you can use
    # them with (VAR)CHAR columns just as well as with DATETIME, etc.
    # and comparisons will still work!
    def coerce_datetime_datetime(self, value):
        return (u"'%04d-%02d-%02d %02d:%02d:%02d'" %
                (value.year, value.month, value.day,
                 value.hour, value.minute, value.second))
    
    def coerce_datetime_date(self, value):
        return u"'%04d-%02d-%02d'" % (value.year, value.month, value.day)
    
    def coerce_datetime_time(self, value):
        return u"'%02d:%02d:%02d'" % (value.hour, value.minute, value.second)
    
    def coerce_datetime_timedelta(self, value):
        float_val = value.days + (value.seconds / 86400.0)
        return repr(float_val)
    
    coerce_decimal = tostr
    coerce_decimal_Decimal = tostr
    
    def do_pickle(self, value):
        return self.coerce_str(pickle.dumps(value))
    
    coerce_dict = do_pickle
    
    coerce_fixedpoint_FixedPoint = tostr
    coerce_float = tostr
    coerce_int = tostr
    
    coerce_list = do_pickle
    
    coerce_long = tostr
    
    def coerce_str(self, value):
        for pat, repl in self.escapes:
            value = value.replace(pat, repl)
        return "'" + value + "'"
    
    coerce_tuple = do_pickle
    
    coerce_unicode = coerce_str


class AdapterFromDB(object):
    """Coerce incoming values from DB types to Dejavu datatypes.
    
    You might notice that we pass coltype around a lot, but don't
    refer to it in this base class. It's there so subclasses can
    make decisions about coercion when they don't have control over
    the types of database columns, and must make do with legacy
    databases implementations.
    
    This base class is designed to work out-of-the-box with PostgreSQL 8.
    """
    
    def coerce(self, value, coltype, valuetype=None):
        """coerce(value, coltype, valuetype=None) -> value, coerced by valuetype."""
        if value is None:
            return None
        
        if valuetype is None:
            valuetype = type(value)
        
        mod = valuetype.__module__
        if mod == "__builtin__":
            xform = "coerce_%s" % valuetype.__name__
        else:
            xform = "coerce_%s_%s" % (mod, valuetype.__name__)
        xform = xform.replace(".", "_")
        try:
            xform = getattr(self, xform)
        except AttributeError:
            raise TypeError("'%s' is not handled by %s." %
                            (valuetype, self.__class__))
        return xform(value, coltype)
    
    def consume(self, unit, key, value, coltype):
        expectedType = unit.__class__.property_type(key)
        value = self.coerce(value, coltype, expectedType)
        unit._properties[key] = value
    
    def do_pickle(self, value, coltype):
        # Coerce to str for pickle.loads restriction.
        value = str(value)
        return pickle.loads(value)
    
    def coerce_bool(self, value, coltype):
        return bool(value)
    
    def coerce_datetime_datetime(self, value, coltype):
        chunks = (value[0:4], value[5:7], value[8:10],
                  value[11:13], value[14:16], value[17:19])
        return datetime.datetime(*map(int, chunks))
    
    def coerce_datetime_date(self, value, coltype):
        chunks = (value[0:4], value[5:7], value[8:10])
        return datetime.date(*map(int, chunks))
    
    def coerce_datetime_time(self, value, coltype):
        chunks = (value[0:2], value[3:5], value[6:8])
        return datetime.time(*map(int, chunks))
    
    def coerce_datetime_timedelta(self, value, coltype):
        days, seconds = divmod(value, 1)
        return datetime.timedelta(days, int(seconds * 86400))
    
    def coerce_decimal(self, value, coltype):
        return decimal(str(value))
    
    def coerce_decimal_Decimal(self, value, coltype):
        return decimal.Decimal(str(value))
    
    coerce_dict = do_pickle
    
    def coerce_fixedpoint_FixedPoint(self, value, coltype):
        return fixedpoint.FixedPoint(value)
    
    def coerce_float(self, value, coltype):
        return float(value)
    
    def coerce_int(self, value, coltype):
        return int(value)
    
    coerce_list = do_pickle
    
    def coerce_long(self, value, coltype):
        return long(value)
    
    def coerce_str(self, value, coltype):
        return str(value)
    
    coerce_tuple = do_pickle
    
    def coerce_unicode(self, value, coltype):
        # You should REALLY check into your DB's encoding and override this.
        return unicode(value, "utf8")


# -------------------------- SQL DECOMPILATION -------------------------- #

class ConstWrapper(unicode):
    """Wraps a constant for use in SQLDecompiler's stack.
    
    When we hit LOAD_CONST while decompiling, we occasionally need to keep
    both the base and the coerced value around (see COMPARE_OP for use
    of ConstWrapper.basevalue).
    """
    def __new__(self, basevalue, coerced_value):
        newobj = unicode.__new__(ConstWrapper, coerced_value)
        newobj.basevalue = basevalue
        return newobj
    
    def __str__(self):
        return str(self)
    
    def __unicode__(self):
        return self
    
    def __repr__(self):
        return self


# Stack sentinels
class Sentinel(object):
    
    def __init__(self, name):
        self.name = name
    
    def __repr__(self):
        return 'Stack Sentinel: %s' % self.name

table_arg = Sentinel('Table Arg')
kw_arg = Sentinel('Keyword Arg')
# cannot_represent exists so that a portion of an Expression can be
# labeled imperfect. For example, the function dejavu.iscurrentweek
# rarely has an SQL equivalent. All Units (which match the rest of 
# the Expression) will be recalled; they can then be compared in
# expr.evaluate(unit).
cannot_represent = Sentinel('Cannot Repr')


class SQLDecompiler(codewalk.LambdaDecompiler):
    """SQLDecompiler(tablename, expr, adapter=AdapterToSQL()).
    
    Produce SQL from a supplied Expression object, with a lambda of the form:
        lambda x, **kw: ...
    
    Attributes of x (or whatever the name of the first argument is) will be
    mapped to table columns. Keyword arguments should be bound using
    Expression.bind_args before calling this decompiler.
    """
    
    # Some constants are function or class objects,
    # which should not be coerced.
    no_coerce = (FunctionType,
                 type,
                 type(len),       # <type 'builtin_function_or_method'>
                 )
    
    sql_cmp_op = ('<', '<=', '=', '!=', '>', '>=', 'in', 'not in')
    
    def __init__(self, tablename, expr, adapter=AdapterToSQL()):
        self.tablename = tablename
        self.expr = expr
        self.adapter = adapter
        obj = expr.func
        codewalk.LambdaDecompiler.__init__(self, obj)
    
    def code(self):
        self.imperfect = False
        self.walk()
        # After walk(), self.stack should be reduced to a single string,
        # which is the SQL representation of our Expression.
        result = self.stack[0]
        if result is cannot_represent:
            # The entire expression could not be evaluated.
            result = self.adapter.bool_true
        if result == self.adapter.coerce_bool(True):
            result = self.adapter.bool_true
        if result == self.adapter.coerce_bool(False):
            result = self.adapter.bool_false
        return result
    
    def visit_target(self, terms):
        """A target is an AND or OR test."""
        comp = self.stack.pop()
        trueval = self.adapter.bool_true
        falseval = self.adapter.bool_false
        while terms:
            term, operation = terms.pop()
            if term is cannot_represent:
                term = trueval
            if comp is cannot_represent:
                comp = trueval
            
            # Blurg. SQL Server is *so* picky.
            if term == self.adapter.coerce_bool(True):
                term = trueval
            if term == self.adapter.coerce_bool(False):
                term = falseval
            if comp == self.adapter.coerce_bool(True):
                comp = trueval
            if comp == self.adapter.coerce_bool(False):
                comp = falseval
            
            comp = "(%s) %s (%s)" % (term, operation.upper(), comp)
        self.stack.append(comp)
    
    def visit_LOAD_DEREF(self, lo, hi):
        raise ValueError("Illegal reference found in %s." % self.expr)
    
    def visit_LOAD_GLOBAL(self, lo, hi):
        raise ValueError("Illegal global found in %s." % self.expr)
    
    def visit_LOAD_FAST(self, lo, hi):
        if lo + (hi << 8) < self.co_argcount:
            self.stack.append(table_arg)
        else:
            self.stack.append(kw_arg)
    
    def visit_LOAD_ATTR(self, lo, hi):
        name = self.co_names[lo + (hi << 8)]
        tos = self.stack.pop()
        if tos is table_arg:
            # Call another function to make subclassing easier.
            self.stack.append(self.column_name(name))
        else:
            # tos.name will reference an attribute of the tos object.
            # Stick the tos and name in a tuple for later processing.
            self.stack.append((tos, name))
    
    def visit_LOAD_CONST(self, lo, hi):
        val = self.co_consts[lo + (hi << 8)]
        if not isinstance(val, self.no_coerce):
            val = ConstWrapper(val, self.adapter.coerce(val))
        self.stack.append(val)
    
    def visit_BUILD_TUPLE(self, lo, hi):
        terms = ", ".join([self.stack.pop() for i in range(lo + hi << 8)])
        self.stack.append("(" + terms + ")")
    
    visit_BUILD_LIST = visit_BUILD_TUPLE
    
    def visit_CALL_FUNCTION(self, lo, hi):
        kwargs = {}
        for i in xrange(hi):
            val = self.stack.pop()
            key = self.stack.pop()
            kwargs[key] = val
        kwargs = [k + "=" + v for k, v in kwargs.iteritems()]
        
        args = []
        for i in xrange(lo):
            arg = self.stack.pop()
            args.append(arg)
        args.reverse()
        
        if kwargs:
            args += kwargs
        
        func = self.stack.pop()
        
        # Handle function objects.
        if isinstance(func, tuple):
            tos, name = func
            dispatch = getattr(self, "attr_" + name, None)
            if dispatch:
                self.stack.append(dispatch(tos, *args))
                return
        else:
            funcname = func.__module__ + "_" + func.__name__
            funcname = funcname.replace(".", "_")
            if funcname.startswith("_"):
                funcname = "func" + funcname
            dispatch = getattr(self, funcname, None)
            if dispatch:
                self.stack.append(dispatch(*args))
                return
        
        self.stack.append(cannot_represent)
        self.imperfect = True
    
    def visit_COMPARE_OP(self, lo, hi):
        op2, op1 = self.stack.pop(), self.stack.pop()
        if op1 is cannot_represent or op2 is cannot_represent:
            self.stack.append(cannot_represent)
            return
        
        op = lo + (hi << 8)
        if op in (6, 7):     # in, not in
            value = self.containedby(op1, op2)
            if op == 7:
                value = "NOT " + value
            self.stack.append(value)
        elif op1 == 'NULL':
            if op == 2:
                self.stack.append(op2 + " IS NULL")
            elif op == 3:
                self.stack.append(op2 + " IS NOT NULL")
            else:
                raise ValueError("Non-equality Null comparisons not allowed.")
        elif op2 == 'NULL':
            if op == 2:
                self.stack.append(op1 + " IS NULL")
            elif op == 3:
                self.stack.append(op1 + " IS NOT NULL")
            else:
                raise ValueError("Non-equality Null comparisons not allowed.")
        else:
            # Comparison operators for strings are case-sensitive in PG et al.
            self.stack.append(op1 + " " + self.sql_cmp_op[op] + " " + op2)
    
    def binary_op(self, op):
        op2, op1 = self.stack.pop(), self.stack.pop()
        self.stack.append(op1 + " " + op + " " + op2)
    
    def visit_BINARY_SUBSCR(self):
        # The only BINARY_SUBSCR used in Expressions should be kwargs[key].
        name = self.stack.pop()
        tos = self.stack.pop()
        if tos is not kw_arg:
            raise ValueError("Subscript %s of %s object not allowed."
                             % (name, tos))
        # name, since formed in LOAD_CONST, may have extraneous quotes.
        name = name.strip("'\"")
        value = self.expr.kwargs[name]
        if not isinstance(value, self.no_coerce):
            value = ConstWrapper(value, self.adapter.coerce(value))
        self.stack.append(value)
    
    def visit_UNARY_NOT(self):
        op = self.stack.pop()
        if op is cannot_represent:
            self.stack.append(cannot_represent)
        else:
            self.stack.append("NOT (" + op + ")")
    
    def column_name(self, name):
        # This is valid SQL for PostgreSQL only and should be overridden.
        # If you want to use a map from UnitProperty names to legacy DB
        # column names, override this method. You will probably also
        # want to override StorageManager.identifier and perform the
        # same map lookup there.
        return '%s."%s"' % (self.tablename, name)
    
    # --------------------------- Dispatchees --------------------------- #
    
    def attr_startswith(self, tos, arg):
        return tos + " LIKE '" + self.adapter.escape_like(arg) + "%'"
    
    def attr_endswith(self, tos, arg):
        return tos + " LIKE '%" + self.adapter.escape_like(arg) + "'"
    
    def containedby(self, op1, op2):
        if isinstance(op1, ConstWrapper):
            # Looking for text in a field. Use Like (reverse terms).
            return op2 + " LIKE '%" + self.adapter.escape_like(op1) + "%'"
        else:
            # Looking for field in (a, b, c)
            atoms = [self.adapter.coerce(x) for x in op2.basevalue]
            return op1 + " IN (" + ", ".join(atoms) + ")"
    
    def dejavu_icontainedby(self, op1, op2):
        if isinstance(op1, ConstWrapper):
            # Looking for text in a field. Use Like (reverse terms).
            return ("LOWER(" + op2 + ") LIKE '%" +
                    self.adapter.escape_like(op1).lower() + "%'")
        else:
            # Looking for field in (a, b, c).
            # Force all args to lowercase for case-insensitive comparison.
            atoms = [self.adapter.coerce(x).lower() for x in op2.basevalue]
            return "LOWER(%s) IN (%s)" % (op1, ", ".join(atoms))
    
    def dejavu_icontains(self, x, y):
        return self.dejavu_icontainedby(y, x)
    
    def dejavu_istartswith(self, x, y):
        return "LOWER(" + x + ") LIKE '" + self.adapter.escape_like(y) + "%'"
    
    def dejavu_iendswith(self, x, y):
        return "LOWER(" + x + ") LIKE '%" + self.adapter.escape_like(y) + "'"
    
    def dejavu_ieq(self, x, y):
        return "LOWER(" + x + ") = LOWER(" + y + ")"
    
    def dejavu_now(self):
        return "NOW()"
    
    def dejavu_today(self):
        return "CURRENT_DATE"
    
    def dejavu_year(self, x):
        return "YEAR(" + x + ")"
    
    def func__builtin___len(self, x):
        return "LENGTH(" + x + ")"


class ConnectionWrapper(object):
    def __init__(self, conn=None):
        self.conn = conn
    
    def __getattr__(self, attr):
        return getattr(self.conn, attr)


class StorageManagerDB(storage.StorageManager):
    """StoreManager to save and retrieve Units using a DB."""
    
    identifier_length = 64
    identifier_caseless = False
    close_connection_method = 'close'
    
    decompiler = SQLDecompiler
    typeAdapter = FieldTypeAdapter()
    toAdapter = AdapterToSQL()
    fromAdapter = AdapterFromDB()
    debug_connections = False
    
    def __init__(self, name, arena, allOptions={}):
        storage.StorageManager.__init__(self, name, arena, allOptions)
        
        # Adapter Overrides
        def get_adapter_option(name):
            adapter_class = allOptions.get(name)
            if isinstance(adapter_class, basestring):
                import xray
                adapter_class = xray.classes(adapter_class)
            return adapter_class
        
        adapter = get_adapter_option(u'Type Adapter')
        if adapter: self.typeAdapter = adapter
        adapter = get_adapter_option(u'To Adapter')
        if adapter: self.toAdapter = adapter
        adapter = get_adapter_option(u'From Adapter')
        if adapter: self.fromAdapter = adapter
        
        self.CreateIfMissing = bool(allOptions.get(u'Create If Missing', 'True'))
        self.pool_size = int(allOptions.get(u'Pool Size', '10'))
        
        self.refs = {}
        if self.pool_size > 0:
            self.pool = Queue.Queue(self.pool_size)
        else:
            self.pool = None
        self.retry = 5
        self.threaded = True
        
        self.prefix = allOptions.get(u'Prefix', u"djv")
        self.reserve_lock = threading.Lock()
        
        ec = {}
        for prop in allOptions.get(u'Expanded Columns', '').split(","):
            if prop:
                cls, type = [x.strip() for x in prop.split(":", 1)]
                lastdot = cls.rfind(".")
                clsname, key = cls[:lastdot], cls[lastdot + 1:]
                ec[(clsname, key)] = type
        self.expanded_columns = ec
    
    def identifier(self, *atoms):
        ident = ''.join(map(str, atoms)).replace('"', '""')
        if self.identifier_caseless:
            ident = ident.lower()
        idlen = self.identifier_length
        if idlen and len(ident) > idlen:
            warnings.warn("Identifier is longer than %s characters." % idlen)
            ident = ident[:idlen]
        return '"' + ident + '"'
    
    def tablename(self, cls):
        if isinstance(cls, type):
            name = cls.__name__
        elif isinstance(cls, dejavu.Unit):
            name = cls.__class__.__name__
        elif isinstance(cls, basestring):
            name = cls
        else:
            raise TypeError("Cannot form tablenames from %s" % cls)
        return self.identifier(self.prefix, name)
    
    def _get_conn(self):
        # Override this with the connection call for your DB. Example follows:
##        try:
##            conn = libpq.PQconnectdb(self.connstring)
##        except Exception, x:
##            if self.CreateIfMissing:
##                self.create_database()
##                conn = libpq.PQconnectdb(self.connstring)
##            else:
##                raise
##        return conn
        raise NotImplementedError
    
    def connection(self):
        if not self.threaded:
            try:
                return self.refs['conn']
            except KeyError:
                self.refs['conn'] = conn = self._get_conn()
                return conn
        
        retry = 0
        while True:
            if self.pool is not None:
                try:
                    conn = self.pool.get_nowait()
                    # Okay, this is freaky. If we wrap here, all goes well.
                    # If we wrap on Queue.put(), mysql crashes after 1700
                    # or so inserts (when migrating Access tables to MySQL).
                    # Go figure.
                    w = ConnectionWrapper(conn)
                    self.refs[weakref.ref(w, self.release)] = w.conn
                    if self.debug_connections:
                        print "-->get %s" % self.__class__.__name__
                    return w
                except Queue.Empty:
                    pass
            
            try:
                conn = self._get_conn()
                w = ConnectionWrapper(conn)
                self.refs[weakref.ref(w, self.release)] = w.conn
                if self.debug_connections:
                    print "create %s" % self.__class__.__name__
                return w
            except OutOfConnectionsError:
                retry += 1
                if retry < self.retry:
                    time.sleep(retry * 1)
                    conn = None
                    continue
                raise OutOfConnectionsError()
    
    def release(self, ref):
        conn = self.refs[ref]
        del self.refs[ref]
        
        if self.pool is not None:
            try:
                self.pool.put_nowait(conn)
                if self.debug_connections:
                    print "<--put %s" % self.__class__.__name__
                return
            except Queue.Full:
                pass
        
        getattr(conn, self.close_connection_method)()
        if self.debug_connections:
            print "___close___ %s" % self.__class__.__name__
    
    def shutdown(self):
        # Empty the pool.
        if self.pool:
            while True:
                try:
                    self.pool.get(True, 0.5)
                except Queue.Empty:
                    break
        
        # Empty self.refs.
        while self.refs:
            ref, conn = self.refs.popitem()
            getattr(conn, self.close_connection_method)()
    
    def create_database(self):
        self.execute("CREATE DATABASE %s;" % self.identifier(self.dbname))
    
    def drop_database(self):
        self.execute("DROP DATABASE %s;" % self.identifier(self.dbname))
    
    def create_storage(self, unitClass):
        tablename = self.tablename(unitClass)
        
        coerce = self.typeAdapter.coerce
        fields = []
        for key in unitClass.properties():
            fields.append(u'%s %s' % (self.identifier(key),
                                      coerce(unitClass, key)))
        self.execute(u'CREATE TABLE %s (%s);' % (tablename, ", ".join(fields)))
        for index in unitClass.indices():
            i = self.identifier(self.prefix, "i", unitClass.__name__, index)
            self.execute(u'CREATE INDEX %s ON %s (%s);' %
                         (i, tablename, self.identifier(index)))
    
    def select(self, unitClass, expr, distinct_fields=None):
        tablename = self.tablename(unitClass)
        if distinct_fields:
            distinct_fields = [self.identifier(x) for x in distinct_fields]
            sql = (u'SELECT DISTINCT %s FROM %s' %
                   (u', '.join(distinct_fields), tablename))
        else:
            sql = u'SELECT * FROM %s' % tablename
        w, i = self.where(unitClass, expr)
        if len(w) > 0:
            w = u" WHERE " + w
        else:
            w = u""
        sql += w + ";"
        return sql, i
    
    def where(self, cls, expr):
        decom = self.decompiler(self.tablename(cls), expr, self.toAdapter)
        return decom.code(), decom.imperfect
    
    def execute(self, query, conn=None):
        """execute(query, conn=None) -> result set."""
        if conn is None:
            conn = self.connection()
        return conn.query(query.encode('utf8'))
    
    def fetch(self, query, conn=None):
        """fetch(query, conn=None) -> rowdata, columns.
        
        This base class uses SQLite3 syntax."""
        res = self.execute(query, conn)
        return res.row_list, res.col_defs
    
    def recall(self, cls, expr=None):
        if expr is None:
            expr = logic.Expression(lambda x: True)
        sql, imperfect = self.select(cls, expr)
        data, col_defs = self.fetch(sql)
        
        columns = dict([(col[0], (index, col[1])) for index, col
                        in enumerate(col_defs)])
        
        # Get specs on properties. Get the ID property first in case other
        # fields depend upon it. See load_expanded, for example.
        props = []
        for key in ['ID'] + [x for x in cls.properties() if x != "ID"]:
            if self.identifier_caseless:
                index, ftype = columns[key.lower()]
            else:
                index, ftype = columns[key]
            subtype = self.expanded_columns.get((cls.__name__, key))
            props.append((key, index, ftype, subtype))
        
        consume = self.fromAdapter.consume
        for row in data:
            unit = cls()
            for key, index, ftype, subtype in props:
                value = row[index]
                if subtype:
                    self.load_expanded(unit, key, subtype)
                else:
                    try:
                        consume(unit, key, value, ftype)
                    except UnicodeDecodeError, x:
                        x.reason += "[%s][%s][%s]" % (key, value, ftype)
                        raise x
                    except Exception, x:
                        x.args += (key, value, ftype)
                        raise x
            
            # If our SQL is imperfect, don't yield it to the
            # caller unless it passes evaluate().
            if (not imperfect) or expr.evaluate(unit):
                unit.cleanse()
                yield unit
    
    def reserve(self, unit):
        """reserve(unit). -> Reserve a persistent slot for unit.
        
        Notice in particular that we do not use the auto-number or
        sequence generation capabilities within some databases, etc.
        The ID should be supplied by UnitSequencers via reserve().
        """
        cls = unit.__class__
        tablename = self.tablename(unit)
        i = self.identifier
        self.reserve_lock.acquire()
        try:
            if unit.ID is None:
                # Examine all existing IDs and grant the "next" one.
                data, cols = self.fetch(u'SELECT %s FROM %s;' % (i('ID'), tablename))
                if data:
                    # sqlite 2, for example, has empty cols tuple if no data.
                    coerce = self.fromAdapter.coerce
                    coltype = cols[0][1]
                    expectedType = cls.property_type("ID")
                    data = [coerce(row[0], coltype, expectedType) for row in data]
                    unit.ID = unit.sequencer.next(data)
                else:
                    unit.ID = unit.sequencer.next([])
                del data
                del cols
            
            fields = []
            values = []
            for key in cls.properties():
                subtype = self.expanded_columns.get((cls.__name__, key))
                if subtype:
                    self.save_expanded(unit, key, subtype)
                else:
                    val = self.toAdapter.coerce(getattr(unit, key))
                    fields.append(i(key))
                    values.append(val)
            
            self.execute('INSERT INTO %s (%s) VALUES (%s);' %
                         (tablename, u", ".join(fields),
                          u", ".join(values)))
            unit.cleanse()
        finally:
            self.reserve_lock.release()
    
    def save(self, unit, forceSave=False):
        """save(unit, forceSave=False) -> Update storage from unit's data."""
        if unit.dirty() or forceSave:
            cls = unit.__class__
            
            parms = []
            for key in cls.properties():
                if key != "ID":
                    subtype = self.expanded_columns.get((cls.__name__, key))
                    if subtype:
                        self.save_expanded(unit, key, subtype)
                    else:
                        val = self.toAdapter.coerce(getattr(unit, key))
                        parms.append('%s = %s' % (self.identifier(key), val))
            
            sql = ('UPDATE %s SET %s WHERE %s = %s;' %
                   (self.tablename(unit), u", ".join(parms),
                    self.identifier("ID"),
                    self.toAdapter.coerce(unit.ID)))
            self.execute(sql)
            unit.cleanse()
    
    def save_expanded(self, unit, key, subtype):
        """save_expanded(unit, key, subtype). Save list in separate table."""
        unitcls = unit.__class__
        table = self.identifier(self.prefix, "_", unitcls.__name__,
                                "_", unit.ID, "_", key)
        
        # Just drop the old table and start with a new one.
        try:
            self.execute(u"DROP TABLE %s;" % table)
        except:
            pass
        
        val = getattr(unit, key)
        if val is None:
            # Don't create a new table at all. This will signal
            # recall() to set the attribute to None on load.
            pass
        else:
            ftype = getattr(self.typeAdapter, "coerce_" + subtype)(unitcls, key)
            self.execute(u"CREATE TABLE %s (EXPVAL %s);" % (table, ftype))
            
            for v in val:
                self.execute(u"INSERT INTO %s (EXPVAL) VALUES ('%s');"
                             % (table, self.toAdapter.coerce(v)))
    
    def load_expanded(self, unit, key, subtype):
        """load_expanded(unit, key, subtype). Load list from separate table."""
        table = self.identifier(self.prefix, "_", unit.__class__.__name__,
                                "_", unit.ID, "_", key)
        try:
            data, col_defs = self.fetch(u"SELECT EXPVAL FROM %s" % table)
        except:
            values = None
        else:
            coltype = col_defs[0][1]
            coercer = getattr(self.fromAdapter, "coerce_" + subtype)
            values = [coercer(row[0], coltype) for row in data]
            
            expected_type = unit.__class__.property_type(key)
            values = expected_type(values)
        
        # Set the attribute directly to avoid __set__ overhead.
        unit._properties[key] = values
    
    def destroy(self, unit):
        """destroy(unit). Delete the unit."""
        self.execute(u'DELETE * FROM %s WHERE %s = %s;' %
                     (self.tablename(unit), self.identifier("ID"),
                      self.toAdapter.coerce(unit.ID)))
    
    def distinct(self, cls, fields, expr=None):
        """distinct(cls, fields, expr=None) -> Distinct values for given fields."""
        if expr is None:
            expr = logic.Expression(lambda x: True)
        
        sql, imperfect = self.select(cls, expr, fields)
        if imperfect:
            # ^%$#@! There's no way to handle imperfect queries without
            # creating all involved Units, which defeats the purpose of
            # distinct, which was a speed issue more than anything.
            warnings.warn("The requested distinct() query for %s Units "
                          "cannot produce perfect SQL with a %s datasource. "
                          "It may take an absurd amount of time to run, "
                          "since each unit must be fully-formed. %s"
                          % (cls.__name__, self.__class__.__name__, expr))
            distvals = {}
            for unit in self.recall(cls, expr):
                # Must use inner tuples for hashability in Sandbox.distinct()
                val = tuple([getattr(unit, f) for f in fields])
                distvals[val] = None
            return distvals.keys()
        else:
            data, columns = self.fetch(sql)
            actualTypes = [x[1] for x in columns]
            expectedTypes = [cls.property_type(x) for x in fields]
            
            coerce = self.fromAdapter.coerce
            # Must use inner tuples for hashability in Sandbox.distinct()
            return [tuple([coerce(val, actualTypes[i], expectedTypes[i])
                           for i, val in enumerate(row)])
                     for row in data]
    
    def multiselect(self, pairs):
        t = self.tablename
        i = self.identifier
        firstcls = pairs[0][0]
        tablenames = []
        # Because various databases may mangle column names, we explicitly
        # order the requested columns (instead of using *).
        columns = []
        wheres = []
        imps = []
        
        for cls, expr in pairs:
            tablenames.append(t(cls))
            # Place the ID property first in case others depend upon it.
            keys = ['ID'] + [k for k in cls.properties() if k != 'ID']
            columns.extend([(cls, k) for k in keys])
            
            if expr is None:
                expr = logic.Expression(lambda x: True)
            w, imp = self.where(cls, expr)
            wheres.append(w)
            imps.append(imp)
            
            if cls is not firstcls:
                spath = self.arena.associations.shortest_path(firstcls, cls)
                # This should be firstcls in every case.
                cls1 = spath.pop(0)
                for cls2 in spath:
                    leftkey, rightkey = cls1._associations[cls2]
                    wheres.append("(%s.%s = %s.%s)" %
                                  (t(cls1), i(leftkey),
                                   t(cls2), i(rightkey)))
                    cls1 = cls2
        
        # Remove any duplicate entries in the where clauses
        # (there may be several from the _join_clauses).
        wheres = dict.fromkeys(wheres).keys()
        
        names = ["%s.%s" % (t(cls), i(key)) for cls, key in columns]
        tbls = u', '.join(tablenames)
        w = u' AND '.join(wheres)
        statement = "SELECT %s FROM %s WHERE %s" % (u', '.join(names), tbls, w)
        return statement, imps, columns
    
    def multirecall(self, *pairs):
        """multirecall(*pairs) -> Full inner join units for each (cls, expr) pair."""
        sql, imps, supplied_cols = self.multiselect(pairs)
        data, recvd_cols = self.fetch(sql)
        
        # Get specs on properties.
        props = []
        for sup, rec in zip(supplied_cols, recvd_cols):
            c, key = sup
            name, ftype = rec[0], rec[1]
            subtype = self.expanded_columns.get((c.__name__, key))
            props.append((c, key, ftype, subtype))
        
        consume = self.fromAdapter.consume
        for row in data:
            index = 0
            units = {}
            for c, key, ftype, subtype in props:
                if c in units:
                    unit = units[c]
                else:
                    units[c] = unit = c()
                value = row[index]
                if subtype:
                    self.load_expanded(unit, key, subtype)
                else:
                    try:
                        consume(unit, key, value, ftype)
                    except UnicodeDecodeError, x:
                        x.reason += "[%s][%s][%s]" % (key, value, ftype)
                        raise x
                    except Exception, x:
                        x.args += (key, value, ftype)
                        raise x
                index += 1
            
            # If our SQL is imperfect, don't yield units to the
            # caller unless they pass evaluate().
            acceptable = True
            unitset = []
            for pair, imp in zip(pairs, imps):
                c, e = pair
                unit = units[c]
                unit.cleanse()
                if imp:
                    acceptable &= e.evaluate(unit)
                    if not acceptable:
                        break
                unitset.append(unit)
            if acceptable:
                yield unitset


class OutOfConnectionsError(dejavu.DejavuError):
    """Exception raised when a database store has run out of connections."""
    pass