Method _post_new_data

lit_nlp/app.py:341–382

Save datapoints provided, after annotatation, for later retrieval. Args: data: JsonDict of datapoints to add, in dict under key 'inputs', per format for other requests. dataset_name: Dataset containing the format of data to add, necessary for proper datapoint annotat

(
      self,
      data: types.JsonDict,
      dataset_name: Optional[str] = None,
      **unused_kw
  )

    return data['inputs']  # pytype: disable=bad-return-type  # always-use-return-annotations

  def _post_new_data(
      self,
      data: types.JsonDict,
      dataset_name: Optional[str] = None,
      **unused_kw
  ) -> dict[str, str]:
    """Save datapoints provided, after annotatation, for later retrieval.

    Args:
      data: JsonDict of datapoints to add, in dict under key 'inputs', per
        format for other requests.
      dataset_name: Dataset containing the format of data to add, necessary for
        proper datapoint annotation.

    Returns:
      A dict of two URLs (minus the root of the webserver). 'load' value is
      for loading LIT with those datapoints. 'remove' value is for removing
      those new datapoints from this server after they have been loaded, if
      desired.

    Raises:
      KeyError: If the `data` dictionary does not have an "inputs" field.
      ValueError: If a "dataset_name" is not provided.
    """
    if dataset_name is None:
      raise ValueError('Must provide a "dataset_name" to save new datapoints.')
    if 'inputs' not in data:
      raise KeyError('Data dict does not contain "inputs" field.')

    data_with_metadata = [
        {'data': d, 'meta': {'added': True, 'source': 'POST', 'parentId': None}}
        for d in data['inputs']
    ]
    annotation_input = {'inputs': data_with_metadata}
    annotated_data = self._annotate_new_data(annotation_input, dataset_name)
    datapoints_id = utils.get_uuid()
    with self._saved_datapoints_lock:
      self._saved_datapoints[datapoints_id] = annotated_data
    return {
        'load': f'?saved_datapoints_id={datapoints_id}',
        'remove': f'/remove_new_data?saved_datapoints_id={datapoints_id}',
    }

  def _fetch_new_data(
      self, unused_data, saved_datapoints_id: Optional[str] = None, **unused_kw