def connect(self):
   try:
     return self._connect()
   except dbexceptions.OperationalError as e:
     vtdb_logger.get_logger().vtclient_exception(
         self.keyspace, self.shard, self.db_type, e)
     raise

  def test_topo_read_threshold(self):
    before_topo_rtt = vtdb_logger.get_logger().get_topo_rtt()
    # Check original state.
    keyspace_obj = topology.get_keyspace('test_keyspace')
    self.assertNotEqual(keyspace_obj, None, "test_keyspace should be not None")
    self.assertEqual(keyspace_obj.sharding_col_type, keyrange_constants.KIT_UINT64, "ShardingColumnType be %s" % keyrange_constants.KIT_UINT64)

    # Change the keyspace object.
    utils.run_vtctl(['SetKeyspaceShardingInfo', '-force', 'test_keyspace',
                     'keyspace_id', keyrange_constants.KIT_BYTES])
    utils.run_vtctl(['RebuildKeyspaceGraph', 'test_keyspace'], auto_log=True)

    # sleep throttle interval and check values again.
    # the keyspace should have changed and also caused a rtt to topo server.
    time.sleep(self.keyspace_fetch_throttle)
    topology.refresh_keyspace(self.vtgate_client, 'test_keyspace')
    keyspace_obj = topology.get_keyspace('test_keyspace')
    after_1st_clear = vtdb_logger.get_logger().get_topo_rtt()
    self.assertEqual(after_1st_clear - before_topo_rtt, 1, "One additional round-trips to topo server")
    self.assertEqual(keyspace_obj.sharding_col_type, keyrange_constants.KIT_BYTES, "ShardingColumnType be %s" % keyrange_constants.KIT_BYTES)

    # Refresh without sleeping for throttle time shouldn't cause additional rtt.
    topology.refresh_keyspace(self.vtgate_client, 'test_keyspace')
    keyspace_obj = topology.get_keyspace('test_keyspace')
    after_2nd_clear = vtdb_logger.get_logger().get_topo_rtt()
    self.assertEqual(after_2nd_clear - after_1st_clear, 0, "No additional round-trips to topo server")

  def _run_with_reconnect(self, *args, **kargs):
    attempt = 0
    while True:
      try:
        return method(self, *args, **kargs)
      except (dbexceptions.RetryError, dbexceptions.FatalError, dbexceptions.TxPoolFull) as e:
        attempt += 1
        # Execution attempt failed with OperationalError, re-read the keyspace.
        if not isinstance(e, dbexceptions.TxPoolFull):
          self.resolve_topology()

        if attempt >= self.max_attempts or self.in_txn:
          self.close()
          vtdb_logger.get_logger().vtclient_exception(self.keyspace, self.shard, self.db_type, e)
          raise dbexceptions.FatalError(*e.args)
        if method.__name__ == 'begin':
          time.sleep(BEGIN_RECONNECT_DELAY)
        else:
          time.sleep(RECONNECT_DELAY)
        if not isinstance(e, dbexceptions.TxPoolFull):
          logging.info("Attempting to reconnect, %d", attempt)
          self.close()
          self.connect()
          logging.info("Successfully reconnected to %s", str(self.conn))
        else:
          logging.info("Waiting to retry for dbexceptions.TxPoolFull to %s, attempt %d", str(self.conn), attempt)

def read_topology(zkocc_client, read_fqdb_keys=True):
  fqdb_keys = []
  db_keys = []
  keyspace_list = zkocc_client.get_srv_keyspace_names('local')
  # validate step
  if len(keyspace_list) == 0:
    vtdb_logger.get_logger().topo_empty_keyspace_list()
    raise Exception('zkocc returned empty keyspace list')
  for keyspace_name in keyspace_list:
    try:
      ks = keyspace.read_keyspace(zkocc_client, keyspace_name)
      __set_keyspace(ks)
      for db_type in ks.db_types:
        for shard_name in ks.get_shard_names(db_type):
          db_key_parts = [ks.name, shard_name, db_type]
          db_key = '.'.join(db_key_parts)
          db_keys.append(db_key)

          if read_fqdb_keys:
            db_instances = len(get_host_port_by_name(zkocc_client, db_key))
            for db_i in xrange(db_instances):
              fqdb_keys.append('.'.join(db_key_parts + [str(db_i)]))
    except Exception:
      vtdb_logger.get_logger().topo_bad_keyspace_data(keyspace_name)
  return db_keys, fqdb_keys

 def _execute(self, sql, bind_variables):
   sql, bind_variables = dbapi.prepare_query_bind_vars(sql, bind_variables)
   try:
     result = self.conn._execute(sql, bind_variables)
   except dbexceptions.IntegrityError as e:
     vtdb_logger.get_logger().integrity_error(e)
     raise
   return result

  def test_keyspace_reresolve_on_conn_failure(self):
    before_topo_rtt = vtdb_logger.get_logger().get_topo_rtt()
    self.replica_tablet.kill_vttablet()
    time.sleep(self.keyspace_fetch_throttle)
    try:
      replica_conn = get_connection(
          db_type='replica', shard_index=self.shard_index)
    except Exception:
      pass

    after_tablet_conn_error = vtdb_logger.get_logger().get_topo_rtt()
    self.assertEqual(
        after_tablet_conn_error - before_topo_rtt, 1,
        'One additional round-trips to topo server')
    self.replica_tablet.start_vttablet()

  def _execute_batch(self, sql_list, bind_variables_list):
    sane_sql_list = []
    sane_bind_vars_list = []
    for sql, bind_variables in zip(sql_list, bind_variables_list):
      sane_sql, sane_bind_vars = dbapi.prepare_query_bind_vars(sql, bind_variables)
      sane_sql_list.append(sane_sql)
      sane_bind_vars_list.append(sane_bind_vars)

    try:
      result = self.conn._execute_batch(sane_sql_list, sane_bind_vars_list)
    except dbexceptions.IntegrityError as e:
      vtdb_logger.get_logger().integrity_error(e)
      raise

    return result

 def __init__(self, addr, timeout, user=None, password=None,
              keyfile=None, certfile=None):
   self.addr = addr
   self.timeout = timeout
   self.client = bsonrpc.BsonRpcClient(
       addr, timeout, user, password, keyfile=keyfile, certfile=certfile)
   self.logger_object = vtdb_logger.get_logger()

 def setUp(self):
   self.shard_index = 1
   self.keyrange = get_keyrange(shard_names[self.shard_index])
   self.master_tablet = shard_1_master
   self.replica_tablet = shard_1_replica
   vtdb_logger.register_vtdb_logger(VTGateTestLogger())
   self.logger = vtdb_logger.get_logger()

 def __init__(self, addr, tablet_type, keyspace, shard, timeout, user=None, password=None, encrypted=False, keyfile=None, certfile=None):
   self.addr = addr
   self.tablet_type = tablet_type
   self.keyspace = keyspace
   self.shard = shard
   self.timeout = timeout
   self.client = bsonrpc.BsonRpcClient(addr, timeout, user, password, encrypted=encrypted, keyfile=keyfile, certfile=certfile)
   self.logger_object = vtdb_logger.get_logger()

def refresh_keyspace(zkocc_client, name):
  global __keyspace_fetch_throttle

  time_last_fetch = get_time_last_fetch(name)
  if time_last_fetch is None:
    return

  if (time_last_fetch + __keyspace_fetch_throttle) > time.time():
    return

  start_time = time.time()
  ks = keyspace.read_keyspace(zkocc_client, name)
  topo_rtt = time.time() - start_time
  if ks is not None:
    __set_keyspace(ks)

  vtdb_logger.get_logger().topo_keyspace_fetch(name, topo_rtt)

 def test_keyspace_reresolve_on_conn_failure(self):
   before_topo_rtt = vtdb_logger.get_logger().get_topo_rtt()
   self.replica_tablet.kill_vttablet()
   time.sleep(self.keyspace_fetch_throttle)
   try:
     replica_conn = get_connection(db_type='replica', shard_index=self.shard_index)
   except Exception, e:
     pass

 def __init__(self, addr, timeout, user=None, password=None,
              keyfile=None, certfile=None):
   super(VTGateConnection, self).__init__(addr, timeout)
   self.user = user
   self.password = password
   self.keyfile = keyfile
   self.certfile = certfile
   self.client = self._create_client()
   self.logger_object = vtdb_logger.get_logger()

def convert_exception(exc, *args):
  new_args = exc.args + args
  vtdb_logger.get_logger().vtgatev2_exception(exc)
  if isinstance(exc, gorpc.TimeoutError):
    return dbexceptions.TimeoutError(new_args)
  elif isinstance(exc, gorpc.AppError):
    msg = str(exc[0]).lower()
    match = _errno_pattern.search(msg)
    if match:
      mysql_errno = int(match.group(1))
      if mysql_errno == 1062:
        return dbexceptions.IntegrityError(new_args)
    return dbexceptions.DatabaseError(new_args)
  elif isinstance(exc, gorpc.ProgrammingError):
    return dbexceptions.ProgrammingError(new_args)
  elif isinstance(exc, gorpc.GoRpcError):
    return dbexceptions.FatalError(new_args)
  return exc

 def test_keyspace_reresolve_on_execute(self):
   before_topo_rtt = vtdb_logger.get_logger().get_topo_rtt()
   try:
     replica_conn = get_connection(
         db_type='replica', shard_index=self.shard_index)
   except Exception, e:
     self.fail(
         'Connection to shard %s replica failed with error %s' %
         (shard_names[self.shard_index], str(e)))

    def _log_exception(exc, *args):
        logger_object = vtdb_logger.get_logger()

        new_exception = method(exc, *args)

        if isinstance(new_exception, dbexceptions.IntegrityError):
            logger_object.integrity_error(new_exception)
        else:
            logger_object.vtgatev2_exception(new_exception)
        return new_exception

 def __init__(self, addr, timeout, user=None, password=None, keyfile=None, certfile=None):
     self.session = None
     self.addr = addr
     self.user = user
     self.password = password
     self.keyfile = keyfile
     self.certfile = certfile
     self.timeout = timeout
     self.client = self._create_client()
     self.logger_object = vtdb_logger.get_logger()

 def __init__(self, vtgate_addrs=None, lag_tolerant_mode=False, master_access_disabled=False):
     self.vtgate_addrs = vtgate_addrs
     self.lag_tolerant_mode = lag_tolerant_mode
     self.master_access_disabled = master_access_disabled
     self.vtgate_connection = None
     self.change_master_read_to_replica = False
     self._transaction_stack_depth = 0
     self.event_logger = vtdb_logger.get_logger()
     self.connection_timeout = DEFAULT_CONNECTION_TIMEOUT
     self._tablet_type = None

 def __init__(self, addr, timeout, user=None, password=None,
              keyfile=None, certfile=None):
   # TODO: Merge. This is very similar to vtgatev2.
   self.session = None
   self.addr = addr
   self.user = user
   self.password = password
   self.keyfile = keyfile
   self.certfile = certfile
   self.timeout = timeout
   self.client = self._create_client()
   self.logger_object = vtdb_logger.get_logger()

def get_host_port_by_name(topo_client, db_key):
  parts = db_key.split(':')
  if len(parts) == 2:
    service = parts[1]
  else:
    service = 'mysql'

  db_key = parts[0]
  ks, shard, tablet_type = db_key.split('.')
  try:
    data = topo_client.get_end_points('local', ks, shard, tablet_type)
  except zkocc.ZkOccError as e:
    vtdb_logger.get_logger().topo_zkocc_error('do data', db_key, e)
    return []
  except Exception as e:
    vtdb_logger.get_logger().topo_exception(
        'failed to get or parse topo data', db_key, e)
    return []
  if 'Entries' not in data:
    vtdb_logger.get_logger().topo_exception(
        'topo server returned: ' + str(data), db_key, e)
    raise Exception('zkocc returned: %s' % str(data))

  host_port_list = []
  for entry in data['Entries']:
    if service in entry['PortMap']:
      host_port = (entry['Host'], entry['PortMap'][service])
      host_port_list.append(host_port)
  random.shuffle(host_port_list)
  return host_port_list