def testbadargs(self):
        manager = pool.manage(mock_dbapi)

        try:
            connection = manager.connect(None)
        except:
            pass

    def __init__(self, host, database, user, password, debug=False):
        self.host = host
        self.database = database
        # 用于判断是否打印sql
        self.debug = debug

        args = {'database': database}
        if user is not None:
            args['user'] = user
        if password is not None:
            args['password'] = password

        pair = host.split(':')
        if len(pair) == 2:
            args['host'] = pair[0]
            args['port'] = int(pair[1])
        else:
            args['host'] = pair[0]
            args['port'] = 3388

        self.args = args
        self._db = None
        # 单位为秒
        self._db_operation_timeout = 300

        # 连接池
        self.conn_pool = pool.manage(
            MySQLdb, pool_size=20, max_overflow=40, timeout=self._db_operation_timeout, recycle=1800)

        self.reconnection()

    def testnonthreadlocalmanager(self):
        manager = pool.manage(mock_dbapi, use_threadlocal=False)

        connection = manager.connect("foo.db")
        connection2 = manager.connect("foo.db")

        self.assert_(connection.cursor() is not None)
        self.assert_(connection is not connection2)

    def test_non_thread_local_manager(self):
        manager = pool.manage(MockDBAPI(), use_threadlocal=False)

        connection = manager.connect('foo.db')
        connection2 = manager.connect('foo.db')

        self.assert_(connection.cursor() is not None)
        self.assert_(connection is not connection2)

def patch_postgresql():
    try:
        from django.db.backends.postgresql_psycopg2 import base as pgsql_base
    except (ImproperlyConfigured, ImportError) as e:
        return

    if not hasattr(pgsql_base, "_Database"):
        pgsql_base._Database = pgsql_base.Database
        pgsql_base.Database = manage(pgsql_base._Database, **POOL_SETTINGS)

def patch_oracle():
    try:
        from django.db.backends.oracle import base as oracle_base
    except (ImproperlyConfigured, ImportError) as e:
        return

    if not hasattr(oracle_base, "_Database"):
        oracle_base._Database = oracle_base.Database
        oracle_base.Database = manage(oracle_base._Database, **POOL_SETTINGS)

def patch_sqlite3():
    try:
        from django.db.backends.sqlite3 import base as sqlite3_base
    except (ImproperlyConfigured, ImportError) as e:
        return

    if not hasattr(sqlite3_base, "_Database"):
        sqlite3_base._Database = sqlite3_base.Database
        sqlite3_base.Database = manage(sqlite3_base._Database, **POOL_SETTINGS)

    def testmanager(self):
        manager = pool.manage(mock_dbapi, use_threadlocal=True)

        connection = manager.connect("foo.db")
        connection2 = manager.connect("foo.db")
        connection3 = manager.connect("bar.db")

        self.assert_(connection.cursor() is not None)
        self.assert_(connection is connection2)
        self.assert_(connection2 is not connection3)

    def testnonthreadlocalmanager(self):
        manager = pool.manage(mock_dbapi, use_threadlocal = False)
        
        connection = manager.connect('foo.db')
        connection2 = manager.connect('foo.db')

        self.echo( "connection " + repr(connection))

        self.assert_(connection.cursor() is not None)
        self.assert_(connection is not connection2)

    def testmanager(self):
        manager = pool.manage(mock_dbapi, use_threadlocal=True)

        connection = manager.connect('foo.db')
        connection2 = manager.connect('foo.db')
        connection3 = manager.connect('bar.db')

        print "connection " + repr(connection)
        self.assert_(connection.cursor() is not None)
        self.assert_(connection is connection2)
        self.assert_(connection2 is not connection3)

def get_pool():
    "Creates one and only one pool using the configured settings."
    global MYSQLPOOL
    if MYSQLPOOL is None:
        backend = getattr(settings, 'MYSQLPOOL_BACKEND', MYSQLPOOL_BACKEND)
        backend = getattr(pool, backend)
        kwargs = getattr(settings, 'MYSQLPOOL_ARGUMENTS', {})
        kwargs.setdefault('poolclass', backend)
        # The user can override this, but set it by default for safety.
        kwargs.setdefault('recycle', MYSQLPOOL_TIMEOUT)
        MYSQLPOOL = pool.manage(OldDatabase, **kwargs)
    return MYSQLPOOL

 def _connect(self):
     if cfg['CFG_MISCUTIL_SQL_USE_SQLALCHEMY']:
         try:
             import sqlalchemy.pool as pool
             import MySQLdb as mysqldb
             mysqldb = pool.manage(mysqldb, use_threadlocal=True)
             connect = mysqldb.connect
         except ImportError:
             cfg['CFG_MISCUTIL_SQL_USE_SQLALCHEMY'] = False
             from MySQLdb import connect
     else:
         from MySQLdb import connect
     return connect

def _get_pool():
    """
    Private function to get the database pool instance, or create one if it's not available yet.

    :returns: The connection pool instance
    """
    global _pool

    if _pool is None:
        # FIXME: Use pool size and other parameters from config
        _pool = dbpool.manage(MySQLdb, pool_size=1, use_threadlocal=True)

    return _pool

    def test_manager_with_key(self):
        class NoKws(object):
            def connect(self, arg):
                return MockConnection()

        manager = pool.manage(NoKws(), use_threadlocal=True)

        c1 = manager.connect('foo.db', sa_pool_key="a")
        c2 = manager.connect('foo.db', sa_pool_key="b")
        c3 = manager.connect('bar.db', sa_pool_key="a")

        assert c1.cursor() is not None
        assert c1 is not c2
        assert c1 is  c3

def get_pool():
    """Create one and only one pool using the configured settings."""
    global MYSQLPOOL
    if MYSQLPOOL is None:
        backend_name = getattr(settings, "SHIELD_MYSQL_POOL_BACKEND", DEFAULT_BACKEND)
        backend = getattr(local_pool, backend_name)
        kwargs = getattr(settings, "SHIELD_MYSQL_POOL_ARGUMENTS", {})
        kwargs.setdefault("poolclass", backend)
        kwargs.setdefault("recycle", DEFAULT_POOL_TIMEOUT)
        MYSQLPOOL = sa_pool.manage(OldDatabase, **kwargs)
        setattr(MYSQLPOOL, "_pid", os.getpid())

    if getattr(MYSQLPOOL, "_pid", None) != os.getpid():
        sa_pool.clear_managers()
    return MYSQLPOOL

    def test_manager(self):
        manager = pool.manage(MockDBAPI(), use_threadlocal=True)

        c1 = manager.connect('foo.db')
        c2 = manager.connect('foo.db')
        c3 = manager.connect('bar.db')
        c4 = manager.connect("foo.db", bar="bat")
        c5 = manager.connect("foo.db", bar="hoho")
        c6 = manager.connect("foo.db", bar="bat")

        assert c1.cursor() is not None
        assert c1 is c2
        assert c1 is not c3
        assert c4 is c6
        assert c4 is not c5

def get_pool():
    """Create one and only one pool using the configured settings."""
    global MYSQLPOOL
    if MYSQLPOOL is None:
        backend_name = getattr(settings, 'MYSQLPOOL_BACKEND', DEFAULT_BACKEND)
        backend = getattr(pool, backend_name)
        kwargs = getattr(settings, 'MYSQLPOOL_ARGUMENTS', {})
        kwargs.setdefault('poolclass', backend)
        kwargs.setdefault('recycle', DEFAULT_POOL_TIMEOUT)
        MYSQLPOOL = pool.manage(OldDatabase, **kwargs)
        setattr(MYSQLPOOL, '_pid', os.getpid())

    if getattr(MYSQLPOOL, '_pid', None) != os.getpid():
        pool.clear_managers()
    return MYSQLPOOL

def init_pool():
     if not globals().get('pool_initialized', False):
         global pool_initialized
         pool_initialized = True
         try:
             backendname = settings.DATABASES['default']['ENGINE']
             backend = load_backend(backendname)

             #replace the database object with a proxy.
             backend.Database = pool.manage(backend.Database)

             backend.DatabaseError = backend.Database.DatabaseError
             backend.IntegrityError = backend.Database.IntegrityError
             logging.info("Connection Pool initialized")
         except:
             logging.exception("Connection Pool initialization error")

def init_pool():
    """From http://blog.bootstraptoday.com/2012/07/11/django-connection-pooling-using-sqlalchemy-connection-pool/"""
    if not globals().get('pool_initialized', False):
        global pool_initialized
        pool_initialized = True
        try:
            backendname = settings.DATABASES['default']['ENGINE']
            backend = load_backend(backendname)
            
            #replace the database object with a proxy.
            backend.Database = pool.manage(backend.Database, pool_size=settings.DB_POOL_SIZE, max_overflow=-1)
            
            backend.DatabaseError = backend.Database.DatabaseError
            backend.IntegrityError = backend.Database.IntegrityError
            logging.info("Connection Pool initialized")
        except:
            logging.exception("Connection Pool initialization error")

def patch_mysql():
    class hashabledict(dict):
        def __hash__(self):
            return hash(tuple(sorted(self.items())))

    class hashablelist(list):
        def __hash__(self):
            return hash(tuple(sorted(self)))

    class ManagerProxy(object):
        def __init__(self, manager):
            self.manager = manager

        def __getattr__(self, key):
            return getattr(self.manager, key)

        def connect(self, *args, **kwargs):
            if 'conv' in kwargs:
                conv = kwargs['conv']
                if isinstance(conv, dict):
                    items = []
                    for k, v in conv.items():
                        if isinstance(v, list):
                            v = hashablelist(v)
                        items.append((k, v))
                    kwargs['conv'] = hashabledict(items)
            if 'ssl' in kwargs:
                ssl = kwargs['ssl']
                if isinstance(ssl, dict):
                    items = []
                    for k, v in ssl.items():
                        if isinstance(v, list):
                            v = hashablelist(v)
                        items.append((k, v))
                    kwargs['ssl'] = hashabledict(items)
            return self.manager.connect(*args, **kwargs)

    try:
        from django.db.backends.mysql import base as mysql_base
    except (ImproperlyConfigured, ImportError) as e:
        return

    if not hasattr(mysql_base, "_Database"):
        mysql_base._Database = mysql_base.Database
        mysql_base.Database = ManagerProxy(manage(mysql_base._Database, **POOL_SETTINGS))