def _create_v2_request_with_keyranges(
    sql, new_binds, keyspace_name, tablet_type, keyranges, not_in_transaction):
  """Make a request dict from arguments.

  Args:
    sql: Str sql with format tokens.
    new_binds: Dict of bind variables.
    keyspace_name: Str keyspace name.
    tablet_type: Str tablet_type.
    keyranges: A list of keyrange.KeyRange objects.
    not_in_transaction: Bool True if a transaction should not be started
      (generally used when sql is not a write).

  Returns:
    A (str: value) dict.
  """
  sql, new_binds = dbapi.prepare_query_bind_vars(sql, new_binds)
  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()),
      'KeyRanges': keyranges,
      'NotInTransaction': not_in_transaction,
  }
  return req

  def _stream_execute(self, sql, bind_variables):
    req = {
      'Query': {
        'Sql': sql,
        'BindVariables': field_types.convert_bind_vars(bind_variables),
        'SessionId': self.session_id,
        'TransactionId': self.transaction_id
      },
      'ImmediateCallerID': {'Username': self.caller_id}
    }

    self._stream_fields = []
    self._stream_conversions = []
    self._stream_result = None
    self._stream_result_index = 0
    try:
      self.client.stream_call('SqlQuery.StreamExecute2', req)
      first_response = self.client.stream_next()
      reply = first_response.reply
      if reply.get('Err'):
        self.__drain_conn_after_streaming_app_error()
        raise gorpc.AppError(reply['Err'].get(
            'Message', 'Missing error message'))

      for field in reply['Fields']:
        self._stream_fields.append((field['Name'], field['Type']))
        self._stream_conversions.append(
            field_types.conversions.get(field['Type']))
    except gorpc.GoRpcError as e:
      self.logger_object.log_private_data(bind_variables)
      raise convert_exception(e, str(self), sql)
    except:
      logging.exception('gorpc low-level error')
      raise
    return None, 0, 0, self._stream_fields

def _create_v2_request_with_keyspace_ids(sql, new_binds, keyspace_name, tablet_type, keyspace_ids, not_in_transaction):
    """Make a request dict from arguments.

  Args:
    sql: Str sql with format tokens.
    new_binds: Dict of bind variables.
    keyspace_name: Str keyspace name.
    tablet_type: Str tablet_type.
    keyspace_ids: Bytes list of keyspace IDs.
    not_in_transaction: Bool True if a transaction should not be started
      (generally used when sql is not a write).

  Returns:
    A (str: value) dict.
  """
    # keyspace_ids are Keyspace Ids packed to byte[]
    sql, new_binds = dbapi.prepare_query_bind_vars(sql, new_binds)
    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()),
        "KeyspaceIds": keyspace_ids,
        "NotInTransaction": not_in_transaction,
    }
    return req

    def _stream_execute2(self, sql, bind_variables):
        new_binds = field_types.convert_bind_vars(bind_variables)
        query = self._make_req()
        query["Sql"] = sql
        query["BindVariables"] = new_binds
        req = {"Query": query}

        self._stream_fields = []
        self._stream_conversions = []
        self._stream_result = None
        self._stream_result_index = 0
        try:
            self.client.stream_call("SqlQuery.StreamExecute2", req)
            first_response = self.client.stream_next()
            reply = first_response.reply
            if reply.get("Err"):
                self.__drain_conn_after_streaming_app_error()
                raise gorpc.AppError(reply["Err"].get("Message", "Missing error message"))

            for field in reply["Fields"]:
                self._stream_fields.append((field["Name"], field["Type"]))
                self._stream_conversions.append(field_types.conversions.get(field["Type"]))
        except gorpc.GoRpcError as e:
            self.logger_object.log_private_data(bind_variables)
            raise convert_exception(e, str(self), sql)
        except:
            logging.exception("gorpc low-level error")
            raise
        return None, 0, 0, self._stream_fields

  def _execute(self, sql, bind_variables):
    req = {
        'QueryRequest': {
            'Sql': sql,
            'BindVariables': field_types.convert_bind_vars(bind_variables),
            'SessionId': self.session_id,
            'TransactionId': self.transaction_id
        },
        'ImmediateCallerID': {'Username': self.caller_id}
    }

    fields = []
    conversions = []
    results = []
    try:
      response = self.rpc_call_and_extract_error('SqlQuery.Execute2', req)
      reply = response.reply

      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']
    except gorpc.GoRpcError 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 _execute_batch(self, sql_list, bind_variables_list):
    query_list = []
    for sql, bind_vars in zip(sql_list, bind_variables_list):
      req = self._make_req()
      req['Sql'] = sql
      req['BindVariables'] = field_types.convert_bind_vars(bind_vars)
      query_list.append(req)

    rowsets = []

    try:
      req = {'List': query_list}
      response = self.client.call('SqlQuery.ExecuteBatch', req)
      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 as e:
      raise convert_exception(e, str(self), sql_list, bind_variables_list)
    except:
      logging.exception('gorpc low-level error')
      raise
    return rowsets

  def _stream_execute(self, sql, bind_variables):
    new_binds = field_types.convert_bind_vars(bind_variables)
    req = {
        'Sql': sql,
        'BindVariables': new_binds,
        'Keyspace': self.keyspace,
        'TabletType': self.tablet_type,
        'Shards': [self.shard],
    }
    self._add_session(req)

    self._stream_fields = []
    self._stream_conversions = []
    self._stream_result = None
    self._stream_result_index = 0
    try:
      self.client.stream_call('VTGate.StreamExecuteShard', req)
      first_response = self.client.stream_next()
      reply = first_response.reply

      for field in reply['Fields']:
        self._stream_fields.append((field['Name'], field['Type']))
        self._stream_conversions.append(field_types.conversions.get(field['Type']))
    except gorpc.GoRpcError as e:
      raise convert_exception(e, str(self), sql, bind_variables)
    except:
      logging.exception('gorpc low-level error')
      raise
    return None, 0, 0, self._stream_fields

  def _stream_execute2(self, sql, bind_variables):
    new_binds = field_types.convert_bind_vars(bind_variables)
    query = self._make_req()
    query['Sql'] = sql
    query['BindVariables'] = new_binds
    req = {'Query': query}

    self._stream_fields = []
    self._stream_conversions = []
    self._stream_result = None
    self._stream_result_index = 0
    try:
      self.client.stream_call('SqlQuery.StreamExecute2', req)
      first_response = self.client.stream_next()
      reply = first_response.reply
      if reply.get('Err'):
        self.__drain_conn_after_streaming_app_error()
        raise gorpc.AppError(reply['Err'].get('Message', 'Missing error message'))

      for field in reply['Fields']:
        self._stream_fields.append((field['Name'], field['Type']))
        self._stream_conversions.append(field_types.conversions.get(field['Type']))
    except gorpc.GoRpcError as e:
      self.logger_object.log_private_data(bind_variables)
      raise convert_exception(e, str(self), sql)
    except:
      logging.exception('gorpc low-level error')
      raise
    return None, 0, 0, self._stream_fields

def _create_v2_request_with_shards(
    sql, new_binds, keyspace_name, tablet_type, shards,
    not_in_transaction):
  """Make a request dict from arguments.

  Args:
    sql: Str sql with format tokens.
    new_binds: Dict of bind variables.
    keyspace_name: Str keyspace name.
    tablet_type: Str tablet_type.
    shards: String list of shards.
    not_in_transaction: Bool True if a transaction should not be started
      (generally used when sql is not a write).

  Returns:
    A (str: value) dict.
  """
  # keyspace_ids are Keyspace Ids packed to byte[]
  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()),
      'Shards': shards,
      'NotInTransaction': not_in_transaction,
  }
  return req

    def _execute(self, sql, bind_variables):
        new_binds = field_types.convert_bind_vars(bind_variables)
        req = self._make_req()
        req["Sql"] = sql
        req["BindVariables"] = new_binds

        fields = []
        conversions = []
        results = []
        try:
            response = self.client.call("VTGate.ExecuteShard", req)
            reply = response.reply

            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"]
        except gorpc.GoRpcError as e:
            raise convert_exception(e, str(self), sql, bind_variables)
        except:
            logging.exception("gorpc low-level error")
            raise
        return results, rowcount, lastrowid, fields

  def _execute(self, sql, bind_variables):
    new_binds = field_types.convert_bind_vars(bind_variables)
    req = self._make_req()
    req['Sql'] = sql
    req['BindVariables'] = new_binds

    fields = []
    conversions = []
    results = []
    try:
      response = self.client.call('SqlQuery.Execute', req)
      reply = response.reply

      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']
    except gorpc.GoRpcError as e:
      self.logger_object.log_private_data(bind_variables)
      raise convert_exception(e, str(self), sql)
    except:
      logging.exception('gorpc low-level error')
      raise
    return results, rowcount, lastrowid, fields

  def _stream_execute(self, sql, bind_variables):
    new_binds = field_types.convert_bind_vars(bind_variables)
    req = self._make_req()
    req['Sql'] = sql
    req['BindVariables'] = new_binds

    self._stream_fields = []
    self._stream_conversions = []
    self._stream_result = None
    self._stream_result_index = 0
    try:
      self.client.stream_call('SqlQuery.StreamExecute', req)
      first_response = self.client.stream_next()
      reply = first_response.reply

      for field in reply['Fields']:
        self._stream_fields.append((field['Name'], field['Type']))
        self._stream_conversions.append(field_types.conversions.get(field['Type']))
    except gorpc.GoRpcError as e:
      self.logger_object.log_private_data(bind_variables)
      raise convert_exception(e, str(self), sql)
    except:
      logging.exception('gorpc low-level error')
      raise
    return None, 0, 0, self._stream_fields

  def _stream_execute(self, sql, bind_variables, keyspace, tablet_type, keyspace_ids=None, keyranges=None):
    new_binds = field_types.convert_bind_vars(bind_variables)
    exec_method = None
    req = None
    if keyspace_ids is not None:
      req = self._create_req_with_keyspace_ids(sql, new_binds, keyspace, tablet_type, keyspace_ids)
      exec_method = 'VTGate.StreamExecuteKeyspaceIds'
    elif keyrange is not None:
      req = self._create_req_with_keyranges(sql, new_binds, keyspace, tablet_type, keyranges)
      exec_method = 'VTGate.StreamExecuteKeyRanges'
    else:
      raise dbexceptions.ProgrammingError('_execute called without specifying keyspace_ids or keyranges')

    self._add_session(req)

    self._stream_fields = []
    self._stream_conversions = []
    self._stream_result = None
    self._stream_result_index = 0
    try:
      self.client.stream_call(exec_method, req)
      first_response = self.client.stream_next()
      reply = first_response.reply

      for field in reply['Fields']:
        self._stream_fields.append((field['Name'], field['Type']))
        self._stream_conversions.append(field_types.conversions.get(field['Type']))
    except gorpc.GoRpcError as e:
      raise convert_exception(e, str(self), sql, bind_variables)
    except:
      logging.exception('gorpc low-level error')
      raise
    return None, 0, 0, self._stream_fields

def _create_req(sql, new_binds, tablet_type):
  new_binds = field_types.convert_bind_vars(new_binds)
  req = {
        'Sql': sql,
        'BindVariables': new_binds,
        'TabletType': tablet_type,
        }
  return req

def _create_req(sql, new_binds, tablet_type, not_in_transaction):
  new_binds = field_types.convert_bind_vars(new_binds)
  req = {
      'Sql': sql,
      'BindVariables': new_binds,
      'TabletType': tablet_type,
      'NotInTransaction': not_in_transaction,
  }
  return req

def _create_req_with_keyranges(sql, new_binds, keyspace, tablet_type, keyranges):
  sql, new_binds = dbapi.prepare_query_bind_vars(sql, new_binds)
  new_binds = field_types.convert_bind_vars(new_binds)
  req = {
        'Sql': sql,
        'BindVariables': new_binds,
        'Keyspace': keyspace,
        'TabletType': tablet_type,
        'KeyRanges': [keyrange.KeyRange(kr) for kr in keyranges],
        }
  return req

def _create_req_with_keyspace_ids(sql, new_binds, keyspace, tablet_type, keyspace_ids):
  sql, new_binds = dbapi.prepare_query_bind_vars(sql, new_binds)
  new_binds = field_types.convert_bind_vars(new_binds)
  req = {
        'Sql': sql,
        'BindVariables': new_binds,
        'Keyspace': keyspace,
        'TabletType': tablet_type,
        'KeyspaceIds': [str(kid) for kid in keyspace_ids],
        }
  return req

  def _execute_batch(
      self, sql_list, bind_variables_list, keyspace_list, keyspace_ids_list,
      tablet_type, as_transaction, effective_caller_id=None):
    query_list = []
    for sql, bind_vars, keyspace, keyspace_ids in zip(
        sql_list, bind_variables_list, keyspace_list, keyspace_ids_list):
      sql, bind_vars = dbapi.prepare_query_bind_vars(sql, bind_vars)
      query = {}
      query['Sql'] = sql
      query['BindVariables'] = field_types.convert_bind_vars(bind_vars)
      query['Keyspace'] = keyspace
      query['KeyspaceIds'] = keyspace_ids
      query_list.append(query)

    rowsets = []

    try:
      req = {
          'Queries': query_list,
          'TabletType': tablet_type,
          'AsTransaction': as_transaction,
      }
      self._add_caller_id(req, effective_caller_id)
      self._add_session(req)
      response = self._get_client().call('VTGate.ExecuteBatchKeyspaceIds', req)
      self._update_session(response)
      if response.reply.get('Error'):
        raise gorpc.AppError(
            response.reply['Error'], 'VTGate.ExecuteBatchKeyspaceIds')
      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 as e:
      self.logger_object.log_private_data(bind_variables_list)
      raise self._convert_exception(
          e, sql_list, keyspace_ids_list,
          keyspace='', tablet_type=tablet_type)
    except Exception:
      logging.exception('gorpc low-level error')
      raise
    return rowsets

def _create_req_with_keyranges(sql, new_binds, keyspace, tablet_type, keyranges):
  # keyranges are keyspace.KeyRange objects with start/end packed to byte[]
  sql, new_binds = dbapi.prepare_query_bind_vars(sql, new_binds)
  new_binds = field_types.convert_bind_vars(new_binds)
  req = {
        'Sql': sql,
        'BindVariables': new_binds,
        'Keyspace': keyspace,
        'TabletType': tablet_type,
        'KeyRanges': keyranges,
        }
  return req

def _create_req_with_keyspace_ids(sql, new_binds, keyspace, tablet_type, keyspace_ids):
  # keyspace_ids are Keyspace Ids packed to byte[]
  sql, new_binds = dbapi.prepare_query_bind_vars(sql, new_binds)
  new_binds = field_types.convert_bind_vars(new_binds)
  req = {
        'Sql': sql,
        'BindVariables': new_binds,
        'Keyspace': keyspace,
        'TabletType': tablet_type,
        'KeyspaceIds': keyspace_ids,
        }
  return req