def _execute(
      self, sql, bind_variables, tablet_type, not_in_transaction=False):
    req = _create_req(sql, bind_variables, tablet_type, not_in_transaction)
    self._add_session(req)

    fields = []
    conversions = []
    results = []
    rowcount = 0
    lastrowid = 0
    try:
      response = self._get_client().call('VTGate.Execute', req)
      self._update_session(response)
      vtgate_utils.extract_rpc_error('VTGate.Execute', response)
      reply = response.reply

      if reply.get('Result'):
        res = reply['Result']
        for field in res['Fields']:
          fields.append((field['Name'], field['Type']))
          conversions.append(field_types.conversions.get(field['Type']))

        for row in res['Rows']:
          results.append(tuple(_make_row(row, conversions)))

        rowcount = res['RowsAffected']
        lastrowid = res['InsertId']
    except (gorpc.GoRpcError, vtgate_utils.VitessError) as e:
      self.logger_object.log_private_data(bind_variables)
      raise convert_exception(e, str(self), sql)
    except Exception:
      logging.exception('gorpc low-level error')
      raise
    return results, rowcount, lastrowid, fields

 def make_execute_batch_call(self, query_list, uses_keyspace_ids):
   """Make an ExecuteBatch call for KeyspaceIds or Shards queries."""
   filtered_query_list = [
       query for query in query_list
       if query_uses_keyspace_ids(query) == uses_keyspace_ids]
   rowsets = []
   if not filtered_query_list:
     return rowsets
   try:
     req = {
         'Queries': filtered_query_list,
         'TabletType': topodata_pb2.TabletType.Value(tablet_type.upper()),
         'AsTransaction': as_transaction,
     }
     self._add_caller_id(req, effective_caller_id)
     self._add_session(req)
     if uses_keyspace_ids:
       exec_method = 'VTGate.ExecuteBatchKeyspaceIds'
     else:
       exec_method = 'VTGate.ExecuteBatchShard'
     response = self._get_client().call(exec_method, req)
     self._update_session(response)
     vtgate_utils.extract_rpc_error(exec_method, response)
     for query_result in response.reply['List']:
       rowsets.append(self._get_rowset_from_query_result(query_result))
   except (gorpc.GoRpcError, vtgate_utils.VitessError) as e:
     self.logger_object.log_private_data(bind_variables_list)
     raise self._convert_exception(
         e, sql_list, exec_method,
         keyspace='', tablet_type=tablet_type)
   except Exception:
     logging.exception('gorpc low-level error')
     raise
   return rowsets

 def begin(self, effective_caller_id=None):
   try:
     req = {}
     self._add_caller_id(req, effective_caller_id)
     response = self._get_client().call('VTGate.Begin2', req)
     vtgate_utils.extract_rpc_error('VTGate.Begin2', response)
     self.effective_caller_id = effective_caller_id
     self.session = None
     self._update_session(response)
   except (gorpc.GoRpcError, vtgate_utils.VitessError) as e:
     raise self._convert_exception(e)

 def rollback(self):
   try:
     req = {}
     self._add_caller_id(req, self.effective_caller_id)
     self._add_session(req)
     response = self._get_client().call('VTGate.Rollback2', req)
     vtgate_utils.extract_rpc_error('VTGate.Rollback2', response)
   except (gorpc.GoRpcError, vtgate_utils.VitessError) as e:
     raise self._convert_exception(e)
   finally:
     self.session = None
     self.effective_caller_id = None

 def get_srv_keyspace(self, name):
     try:
         response = self._get_client().call("VTGate.GetSrvKeyspace", {"Keyspace": name})
         vtgate_utils.extract_rpc_error("VTGate.GetSrvKeyspace", response)
         # response.reply is a proto3 encoded in bson RPC.
         # we need to make it back to what keyspace.Keyspace expects
         return keyspace.Keyspace(name, keyrange_constants.srv_keyspace_proto3_to_old(response.reply))
     except gorpc.GoRpcError as e:
         raise self._convert_exception(e, keyspace=name)
     except:
         logging.exception("gorpc low-level error")
         raise

  def _execute_batch(
      self, sql_list, bind_variables_list, tablet_type, as_transaction):
    query_list = []
    for sql, bind_vars in zip(sql_list, bind_variables_list):
      query = {}
      query['Sql'] = sql
      query['BindVariables'] = field_types.convert_bind_vars(bind_vars)
      query_list.append(query)

    rowsets = []

    try:
      req = {
          'Queries': query_list,
          'TabletType': tablet_type,
          'AsTransaction': as_transaction,
      }
      self._add_session(req)
      response = self._get_client().call('VTGate.ExecuteBatch', req)
      self._update_session(response)
      vtgate_utils.extract_rpc_error('VTGate.ExecuteBatch', response)
      for reply in response.reply['List']:
        fields = []
        conversions = []
        results = []
        rowcount = 0

        for field in reply['Fields']:
          fields.append((field['Name'], field['Type']))
          conversions.append(field_types.conversions.get(field['Type']))

        for row in reply['Rows']:
          results.append(tuple(_make_row(row, conversions)))

        rowcount = reply['RowsAffected']
        lastrowid = reply['InsertId']
        rowsets.append((results, rowcount, lastrowid, fields))
    except (gorpc.GoRpcError, vtgate_utils.VitessError) as e:
      self.logger_object.log_private_data(bind_variables_list)
      raise convert_exception(e, str(self), sql_list)
    except Exception:
      logging.exception('gorpc low-level error')
      raise
    return rowsets

    def _execute_entity_ids(
        self,
        sql,
        bind_variables,
        keyspace,
        tablet_type,
        entity_keyspace_id_map,
        entity_column_name,
        not_in_transaction=False,
        effective_caller_id=None,
    ):
        sql, new_binds = dbapi.prepare_query_bind_vars(sql, bind_variables)
        new_binds = field_types.convert_bind_vars(new_binds)
        req = {
            "Sql": sql,
            "BindVariables": new_binds,
            "Keyspace": keyspace,
            "TabletType": tablet_type,
            "EntityKeyspaceIDs": [
                {"ExternalID": xid, "KeyspaceID": kid} for xid, kid in entity_keyspace_id_map.iteritems()
            ],
            "EntityColumnName": entity_column_name,
            "NotInTransaction": not_in_transaction,
        }

        self._add_caller_id(req, effective_caller_id)
        self._add_session(req)
        try:
            exec_method = "VTGate.ExecuteEntityIds"
            response = self._get_client().call(exec_method, req)
            self._update_session(response)
            vtgate_utils.extract_rpc_error(exec_method, response)
            reply = response.reply
            return self._get_rowset_from_query_result(reply.get("Result"))
        except (gorpc.GoRpcError, vtgate_utils.VitessError) as e:
            self.logger_object.log_private_data(bind_variables)
            raise self._convert_exception(e, sql, entity_keyspace_id_map, keyspace=keyspace, tablet_type=tablet_type)
        except Exception:
            logging.exception("gorpc low-level error")
            raise

    def _execute(
        self,
        sql,
        bind_variables,
        keyspace,
        tablet_type,
        keyspace_ids=None,
        keyranges=None,
        not_in_transaction=False,
        effective_caller_id=None,
    ):
        exec_method = None
        req = None
        if keyspace_ids is not None:
            req = _create_req_with_keyspace_ids(
                sql, bind_variables, keyspace, tablet_type, keyspace_ids, not_in_transaction
            )
            exec_method = "VTGate.ExecuteKeyspaceIds"
        elif keyranges is not None:
            req = _create_req_with_keyranges(sql, bind_variables, keyspace, tablet_type, keyranges, not_in_transaction)
            exec_method = "VTGate.ExecuteKeyRanges"
        else:
            raise dbexceptions.ProgrammingError("_execute called without specifying keyspace_ids or keyranges")

        self._add_caller_id(req, effective_caller_id)
        self._add_session(req)
        try:
            response = self._get_client().call(exec_method, req)
            self._update_session(response)
            vtgate_utils.extract_rpc_error(exec_method, response)
            reply = response.reply
            return self._get_rowset_from_query_result(reply.get("Result"))
        except (gorpc.GoRpcError, vtgate_utils.VitessError) as e:
            self.logger_object.log_private_data(bind_variables)
            raise self._convert_exception(e, sql, keyspace_ids, keyranges, keyspace=keyspace, tablet_type=tablet_type)
        except Exception:
            logging.exception("gorpc low-level error")
            raise

 def test_reply_is_string(self):
   response = gorpc.GoRpcResponse()
   response.reply = 'foo'
   vtgate_utils.extract_rpc_error('method', response)

 def test_reply_is_empty_string(self):
   response = gorpc.GoRpcResponse()
   vtgate_utils.extract_rpc_error('method', response)

 def test_reply_is_none(self):
   vtgate_utils.extract_rpc_error('method', gorpc.GoRpcResponse())

 def test_reply_has_err_code(self):
   response = gorpc.GoRpcResponse()
   response.reply = {'Err': {'Code': vtrpc_pb2.TRANSIENT_ERROR}}
   with self.assertRaisesRegexp(vtgate_utils.VitessError,
                                'TRANSIENT_ERROR'):
     vtgate_utils.extract_rpc_error('method', response)

 def test_reply_has_missing_err_message(self):
   response = gorpc.GoRpcResponse()
   response.reply = {'Err': {'foo': 'bar'}}
   with self.assertRaisesRegexp(vtgate_utils.VitessError,
                                'Missing error message'):
     vtgate_utils.extract_rpc_error('method', response)

 def test_reply_has_non_dict_err(self):
   response = gorpc.GoRpcResponse()
   response.reply = {'Err': 1}
   with self.assertRaisesRegexp(vtgate_utils.VitessError, 'UNKNOWN_ERROR'):
     vtgate_utils.extract_rpc_error('method', response)

 def test_reply_is_dict(self):
   response = gorpc.GoRpcResponse()
   response.reply = {'foo': 'bar'}
   vtgate_utils.extract_rpc_error('method', response)

  def _execute(
      self, sql, bind_variables, keyspace_name, tablet_type,
      keyspace_ids=None, keyranges=None,
      entity_keyspace_id_map=None, entity_column_name=None,
      not_in_transaction=False, effective_caller_id=None):
    """Execute query.

    Args:
      sql: The sql text, with %(format)s-style tokens.
      bind_variables: (str: value) dict of bind variables corresponding
        to sql %(format)s tokens.
      keyspace_name: Str name of keyspace.
      tablet_type: Str tablet type (e.g. master, rdonly, replica).
      keyspace_ids: bytes list of keyspace ID lists.
      keyranges: KeyRange objects.
      entity_keyspace_id_map: (column value: bytes) map from a column
        to a keyspace id. If defined, vtgate adds a per-shard expression
        to the WHERE clause, and ignores keyspace_ids and keyranges
        parameters.
      entity_column_name: Str name of entity column used by
        entity_keyspace_id_map.
      not_in_transaction: bool.
      effective_caller_id: CallerID object.

    Returns:
      The (results, rowcount, lastrowid, fields) tuple.
    """

    routing_kwargs = {}
    exec_method = None
    req = None
    if entity_keyspace_id_map is not None:
      # This supercedes keyspace_ids and keyranges.
      routing_kwargs['entity_keyspace_id_map'] = entity_keyspace_id_map
      routing_kwargs['entity_column_name'] = entity_column_name
      if entity_column_name is None:
        raise dbexceptions.ProgrammingError(
            '_execute called with entity_keyspace_id_map and no '
            'entity_column_name')
      sql, new_binds = dbapi.prepare_query_bind_vars(sql, bind_variables)
      new_binds = field_types.convert_bind_vars(new_binds)
      req = {
          'Sql': sql,
          'BindVariables': new_binds,
          'Keyspace': keyspace_name,
          'TabletType': topodata_pb2.TabletType.Value(tablet_type.upper()),
          'EntityKeyspaceIDs': [
              {'ExternalID': xid, 'KeyspaceID': kid}
              for xid, kid in entity_keyspace_id_map.iteritems()],
          'EntityColumnName': entity_column_name,
          'NotInTransaction': not_in_transaction,
      }
      exec_method = 'VTGate.ExecuteEntityIds'
    elif keyspace_ids is not None:
      if keyranges is not None:
        raise dbexceptions.ProgrammingError(
            '_execute called with keyspace_ids and keyranges both defined')
      routing_kwargs['keyspace_ids'] = keyspace_ids
      req = _create_v2_request_with_keyspace_ids(
          sql, bind_variables, keyspace_name, tablet_type, keyspace_ids,
          not_in_transaction)
      exec_method = 'VTGate.ExecuteKeyspaceIds'
    elif keyranges is not None:
      routing_kwargs['keyranges'] = keyranges
      req = _create_v2_request_with_keyranges(
          sql, bind_variables, keyspace_name, tablet_type, keyranges,
          not_in_transaction)
      exec_method = 'VTGate.ExecuteKeyRanges'
    else:
      req = _create_v3_request(
          sql, bind_variables, tablet_type, not_in_transaction)
      if keyspace_name is not None:
        raise dbexceptions.ProgrammingError(
            '_execute called with keyspace_name but no routing args')
      exec_method = 'VTGate.Execute'

    self._add_caller_id(req, effective_caller_id)
    if not not_in_transaction:
      self._add_session(req)
    try:
      response = self._get_client().call(exec_method, req)
      self._update_session(response)
      vtgate_utils.extract_rpc_error(exec_method, response)
      reply = response.reply
      return self._get_rowset_from_query_result(reply.get('Result'))
    except (gorpc.GoRpcError, vtgate_utils.VitessError) as e:
      self.logger_object.log_private_data(bind_variables)
      raise self._convert_exception(
          e, sql, keyspace=keyspace_name, tablet_type=tablet_type,
          **routing_kwargs)
    except Exception:
      logging.exception('gorpc low-level error')
      raise

 def test_reply_has_err_message(self):
   response = gorpc.GoRpcResponse()
   response.reply = {'Err': {'Message': 'bar'}}
   with self.assertRaisesRegexp(vtgate_utils.VitessError,
                                'UNKNOWN_ERROR.+bar'):
     vtgate_utils.extract_rpc_error('method', response)

    def _execute(
        self,
        sql,
        bind_variables,
        keyspace,
        tablet_type,
        keyspace_ids=None,
        keyranges=None,
        entity_keyspace_id_map=None,
        entity_column_name=None,
        not_in_transaction=False,
        effective_caller_id=None,
    ):

        routing_kwargs = {}
        exec_method = None
        req = None
        if keyspace_ids is not None:
            routing_kwargs["keyspace_ids"] = keyspace_ids
            req = _create_req_with_keyspace_ids(
                sql, bind_variables, keyspace, tablet_type, keyspace_ids, not_in_transaction
            )
            exec_method = "VTGate.ExecuteKeyspaceIds"
        elif keyranges is not None:
            routing_kwargs["keyranges"] = keyranges
            req = _create_req_with_keyranges(sql, bind_variables, keyspace, tablet_type, keyranges, not_in_transaction)
            exec_method = "VTGate.ExecuteKeyRanges"
        elif entity_keyspace_id_map is not None:
            routing_kwargs["entity_keyspace_id_map"] = entity_keyspace_id_map
            routing_kwargs["entity_column_name"] = entity_column_name
            if entity_column_name is None:
                raise dbexceptions.ProgrammingError(
                    "_execute called with entity_keyspace_id_map and no " "entity_column_name"
                )
            sql, new_binds = dbapi.prepare_query_bind_vars(sql, bind_variables)
            new_binds = field_types.convert_bind_vars(new_binds)
            req = {
                "Sql": sql,
                "BindVariables": new_binds,
                "Keyspace": keyspace,
                "TabletType": tablet_type,
                "EntityKeyspaceIDs": [
                    {"ExternalID": xid, "KeyspaceID": kid} for xid, kid in entity_keyspace_id_map.iteritems()
                ],
                "EntityColumnName": entity_column_name,
                "NotInTransaction": not_in_transaction,
            }
            exec_method = "VTGate.ExecuteEntityIds"
        else:
            raise dbexceptions.ProgrammingError(
                "_execute called with no keyspace_ids, keyranges, or " "entity_keyspace_id_map"
            )

        def check_incompatible_args(arg_name):
            if arg_name not in routing_kwargs:
                raise dbexceptions.ProgrammingError(
                    "_execute called with routing_args=%s, "
                    "incompatible routing arg=%s" % (sorted(routing_kwargs), arg_name)
                )

        if keyranges is not None:
            check_incompatible_args("keyranges")
        if entity_column_name is not None:
            check_incompatible_args("entity_column_name")
        if entity_keyspace_id_map is not None:
            check_incompatible_args("entity_keyspace_id_map")

        self._add_caller_id(req, effective_caller_id)
        self._add_session(req)
        try:
            response = self._get_client().call(exec_method, req)
            self._update_session(response)
            vtgate_utils.extract_rpc_error(exec_method, response)
            reply = response.reply
            return self._get_rowset_from_query_result(reply.get("Result"))
        except (gorpc.GoRpcError, vtgate_utils.VitessError) as e:
            self.logger_object.log_private_data(bind_variables)
            raise self._convert_exception(e, sql, keyspace=keyspace, tablet_type=tablet_type, **routing_kwargs)
        except Exception:
            logging.exception("gorpc low-level error")
            raise