def _test(self, kwargs, expected_values=None, expected_err=None):
    with self.cached_session() as sess:
      if expected_err:
        with self.assertRaisesWithPredicateMatch(expected_err[0],
                                                 expected_err[1]):
          out = parsing_ops.parse_single_example(**kwargs)
          sess.run(flatten_values_tensors_or_sparse(out.values()))
        return
      else:
        # Returns dict w/ Tensors and SparseTensors.
        out = parsing_ops.parse_single_example(**kwargs)
        # Also include a test with the example names specified to retain
        # code coverage of the unfused version, and ensure that the two
        # versions produce the same results.
        out_with_example_name = parsing_ops.parse_single_example(
            example_names="name", **kwargs)
        for result_dict in [out, out_with_example_name]:
          result = flatten_values_tensors_or_sparse(result_dict.values())
          # Check values.
          tf_result = self.evaluate(result)
          _compare_output_to_expected(self, result_dict, expected_values,
                                      tf_result)

      for k, f in kwargs["features"].items():
        if isinstance(f, parsing_ops.FixedLenFeature) and f.shape is not None:
          self.assertEqual(tuple(out[k].get_shape().as_list()), f.shape)
        elif isinstance(f, parsing_ops.VarLenFeature):
          self.assertEqual(
              tuple(out[k].indices.get_shape().as_list()), (None, 1))
          self.assertEqual(tuple(out[k].values.get_shape().as_list()), (None,))
          self.assertEqual(
              tuple(out[k].dense_shape.get_shape().as_list()), (1,))

  def decode(self, serialized_example, items=None):
    """Decodes the given serialized TF-example.

    Args:
      serialized_example: a serialized TF-example tensor.
      items: the list of items to decode. These must be a subset of the item
        keys in self._items_to_handlers. If `items` is left as None, then all
        of the items in self._items_to_handlers are decoded.

    Returns:
      the decoded items, a list of tensor.
    """
    example = parsing_ops.parse_single_example(
        serialized_example,
        self._keys_to_features)

    # Reshape non-sparse elements just once:
    for k in self._keys_to_features:
      v = self._keys_to_features[k]
      if isinstance(v, parsing_ops.FixedLenFeature):
        example[k] = array_ops.reshape(example[k], v.shape)

    if not items:
      items = self._items_to_handlers.keys()

    outputs = []
    for item in items:
      handler = self._items_to_handlers[item]
      keys_to_tensors = {key: example[key] for key in handler.keys}
      outputs.append(handler.tensors_to_item(keys_to_tensors))
    return outputs

def parse_single_example(serialized, features, name=None, example_names=None):
  """Parses a single `Example` proto.

  See tf.parse_single_example.

  Args:
    serialized: A scalar string Tensor or LabeledTensor, a single serialized
      Example.
    features: A `dict` mapping feature keys to `labeled_tensor.FixedLenFeature`
      values.
    name: A name for this operation (optional).
    example_names: (Optional) A scalar string Tensor, the associated name.

  Returns:
    A `dict` mapping feature keys to `LabeledTensor` values.

  Raises:
    ValueError: if any feature is invalid.
  """
  serialized = core.convert_to_labeled_tensor(serialized)
  unlabeled_features = _labeled_to_unlabeled_features(features)

  unlabeled_parsed = parsing_ops.parse_single_example(
      serialized.tensor, unlabeled_features, name, example_names)

  parsed = {}
  for name, parsed_feature in unlabeled_parsed.items():
    parsed[name] = core.LabeledTensor(parsed_feature, features[name].axes)

  return parsed

 def parse_single_example_fn(x):
   features = {
       "dense_int": parsing_ops.FixedLenFeature((), dtypes.int64, 0),
       "dense_str": parsing_ops.FixedLenFeature((), dtypes.string, ""),
       "sparse_int": parsing_ops.VarLenFeature(dtypes.int64),
       "sparse_str": parsing_ops.VarLenFeature(dtypes.string),
   }
   return parsing_ops.parse_single_example(x, features)

  def test_keyed_parse_json(self):
    gfile.Glob = self._orig_glob
    filename = self._create_temp_file(
        '{"features": {"feature": {"age": {"int64_list": {"value": [0]}}}}}\n'
        '{"features": {"feature": {"age": {"int64_list": {"value": [1]}}}}}\n'
        '{"features": {"feature": {"age": {"int64_list": {"value": [2]}}}}}\n')

    batch_size = 1
    queue_capacity = 5
    name = "my_batch"

    with ops.Graph().as_default() as g, self.test_session(graph=g) as session:
      dtypes = {"age": parsing_ops.FixedLenFeature([1], dtypes_lib.int64)}
      parse_fn = lambda example: parsing_ops.parse_single_example(  # pylint: disable=g-long-lambda
          parsing_ops.decode_json_example(example), dtypes)
      keys, inputs = graph_io.read_keyed_batch_examples(
          filename,
          batch_size,
          reader=io_ops.TextLineReader,
          randomize_input=False,
          num_epochs=1,
          queue_capacity=queue_capacity,
          parse_fn=parse_fn,
          name=name)
      self.assertAllEqual((None,), keys.get_shape().as_list())
      self.assertEqual(1, len(inputs))
      self.assertAllEqual((None, 1), inputs["age"].get_shape().as_list())
      session.run(variables.local_variables_initializer())

      coord = coordinator.Coordinator()
      threads = queue_runner_impl.start_queue_runners(session, coord=coord)

      key, age = session.run([keys, inputs["age"]])
      self.assertAllEqual(age, [[0]])
      self.assertAllEqual(key, [filename.encode("utf-8") + b":1"])
      key, age = session.run([keys, inputs["age"]])
      self.assertAllEqual(age, [[1]])
      self.assertAllEqual(key, [filename.encode("utf-8") + b":2"])
      key, age = session.run([keys, inputs["age"]])
      self.assertAllEqual(age, [[2]])
      self.assertAllEqual(key, [filename.encode("utf-8") + b":3"])
      with self.assertRaises(errors.OutOfRangeError):
        session.run(inputs)

      coord.request_stop()
      coord.join(threads)

 def testReadWithEquivalentDataset(self):
   features = {
       "file": parsing_ops.FixedLenFeature([], dtypes.int64),
       "record": parsing_ops.FixedLenFeature([], dtypes.int64),
   }
   dataset = (
       core_readers.TFRecordDataset(self.test_filenames)
       .map(lambda x: parsing_ops.parse_single_example(x, features))
       .repeat(10).batch(2))
   next_element = self.getNext(dataset)
   for file_batch, _, _, _, record_batch, _ in self._next_expected_batch(
       range(self._num_files), 2, 10):
     actual_batch = self.evaluate(next_element())
     self.assertAllEqual(file_batch, actual_batch["file"])
     self.assertAllEqual(record_batch, actual_batch["record"])
   with self.assertRaises(errors.OutOfRangeError):
     self.evaluate(next_element())

  def testReadWithEquivalentDataset(self):
    features = {
        "file": parsing_ops.FixedLenFeature([], dtypes.int64),
        "record": parsing_ops.FixedLenFeature([], dtypes.int64),
    }
    dataset = (core_readers.TFRecordDataset(self.test_filenames)
               .map(lambda x: parsing_ops.parse_single_example(x, features))
               .repeat(10).batch(2))
    iterator = dataset.make_initializable_iterator()
    init_op = iterator.initializer
    next_element = iterator.get_next()

    with self.test_session() as sess:
      sess.run(init_op)
      for file_batch, _, _, _, record_batch in self._next_expected_batch(
          range(self._num_files), 2, 10):
        actual_batch = sess.run(next_element)
        self.assertAllEqual(file_batch, actual_batch["file"])
        self.assertAllEqual(record_batch, actual_batch["record"])
      with self.assertRaises(errors.OutOfRangeError):
        sess.run(next_element)

 def loop_fn(i):
   example_proto = array_ops.gather(examples, i)
   f = parsing_ops.parse_single_example(example_proto, features)
   return f