






import os
import tempfile
from functools import partial
from unittest import TestCase
from unittest.mock import patch

import datasets
import datasets.config

from .utils import require_beam



class 				_A			(     datasets.BeamBasedBuilder				):
    """simple docstring"""

    def _a							(				self      :     List[Any]      )      ->							List[str]:
           return datasets.DatasetInfo(
               features=datasets.Features({"""content""": datasets.Value("""string"""      )}      ) ,     supervised_keys=__lowerCAmelCase ,     )



    def _a							(				self      :     Optional[Any] ,     __SCREAMING_SNAKE_CASE      :     List[str] ,     __SCREAMING_SNAKE_CASE      :     Dict      )      ->							List[Any]:
           return [datasets.SplitGenerator(name=datasets.Split.TRAIN ,     gen_kwargs={"""examples""": get_test_dummy_examples()}      )]



    def _a							(				self      :     List[Any] ,     __SCREAMING_SNAKE_CASE      :     Optional[int] ,     __SCREAMING_SNAKE_CASE      :     Tuple      )      ->							List[str]:
           import apache_beam as beam

           return pipeline | "Load Examples" >> beam.Create(__lowerCAmelCase      )




class 				_A			(     datasets.BeamBasedBuilder				):
    """simple docstring"""

    def _a							(				self      :     Tuple      )      ->							int:
           return datasets.DatasetInfo(
               features=datasets.Features({"""a""": datasets.Sequence({"""b""": datasets.Value("""string"""      )}      )}      ) ,     supervised_keys=__lowerCAmelCase ,     )



    def _a							(				self      :     List[str] ,     __SCREAMING_SNAKE_CASE      :     Optional[int] ,     __SCREAMING_SNAKE_CASE      :     Optional[int]      )      ->							Dict:
           return [
               datasets.SplitGenerator(name=datasets.Split.TRAIN ,     gen_kwargs={"""examples""": get_test_nested_examples()}      )
           ]



    def _a							(				self      :     Optional[int] ,     __SCREAMING_SNAKE_CASE      :     Tuple ,     __SCREAMING_SNAKE_CASE      :     Any      )      ->							str:
           import apache_beam as beam

           return pipeline | "Load Examples" >> beam.Create(__lowerCAmelCase      )







def       lowercase__	(					)	->				Any:







       """simple docstring"""






       return [(i, {"content": content}) for i, content in enumerate(["""foo""", """bar""", """foobar"""]		)]







def       lowercase__	(					)	->				Tuple:







       """simple docstring"""






       return [(i, {"a": {"b": [content]}}) for i, content in enumerate(["""foo""", """bar""", """foobar"""]		)]





class 				_A			(     _A				):
    """simple docstring"""

    @require_beam
    def _a							(				self      :     Union[str, Any]      )      ->							List[str]:
           __UpperCAmelCase     				=len(get_test_dummy_examples()      )
           with tempfile.TemporaryDirectory() as tmp_cache_dir:
                  __UpperCAmelCase     				=DummyBeamDataset(cache_dir=__lowerCAmelCase ,     beam_runner="""DirectRunner"""      )
                  builder.download_and_prepare()
                  self.assertTrue(
                      os.path.exists(
                          os.path.join(__lowerCAmelCase ,     builder.name ,     """default""" ,     """0.0.0""" ,     f'''{builder.name}-train.arrow'''      )      )      )
                  self.assertDictEqual(builder.info.features ,     datasets.Features({"""content""": datasets.Value("""string"""      )}      )      )
                  __UpperCAmelCase     				=builder.as_dataset()
                  self.assertEqual(dset["""train"""].num_rows ,     __lowerCAmelCase      )
                  self.assertEqual(dset["""train"""].info.splits["""train"""].num_examples ,     __lowerCAmelCase      )
                  self.assertDictEqual(dset["""train"""][0] ,     get_test_dummy_examples()[0][1]      )
                  self.assertDictEqual(
                      dset["""train"""][expected_num_examples - 1] ,     get_test_dummy_examples()[expected_num_examples - 1][1]      )
                  self.assertTrue(
                      os.path.exists(os.path.join(__lowerCAmelCase ,     builder.name ,     """default""" ,     """0.0.0""" ,     """dataset_info.json"""      )      )      )
                  del dset



    @require_beam
    def _a							(				self      :     int      )      ->							str:
           import apache_beam as beam

           __UpperCAmelCase     				=beam.io.parquetio.WriteToParquet

           __UpperCAmelCase     				=len(get_test_dummy_examples()      )
           with tempfile.TemporaryDirectory() as tmp_cache_dir:
                  __UpperCAmelCase     				=DummyBeamDataset(cache_dir=__lowerCAmelCase ,     beam_runner="""DirectRunner"""      )
                  with patch("""apache_beam.io.parquetio.WriteToParquet"""      ) as write_parquet_mock:
                         __UpperCAmelCase     				=partial(__lowerCAmelCase ,     num_shards=2      )
                         builder.download_and_prepare()
                  self.assertTrue(
                      os.path.exists(
                          os.path.join(
                              __lowerCAmelCase ,     builder.name ,     """default""" ,     """0.0.0""" ,     f'''{builder.name}-train-00000-of-00002.arrow'''      )      )      )
                  self.assertTrue(
                      os.path.exists(
                          os.path.join(
                              __lowerCAmelCase ,     builder.name ,     """default""" ,     """0.0.0""" ,     f'''{builder.name}-train-00000-of-00002.arrow'''      )      )      )
                  self.assertDictEqual(builder.info.features ,     datasets.Features({"""content""": datasets.Value("""string"""      )}      )      )
                  __UpperCAmelCase     				=builder.as_dataset()
                  self.assertEqual(dset["""train"""].num_rows ,     __lowerCAmelCase      )
                  self.assertEqual(dset["""train"""].info.splits["""train"""].num_examples ,     __lowerCAmelCase      )
                  # Order is not preserved when sharding, so we just check that all the elements are there
                  self.assertListEqual(sorted(dset["""train"""]["""content"""]      ) ,     sorted(["""foo""", """bar""", """foobar"""]      )      )
                  self.assertTrue(
                      os.path.exists(os.path.join(__lowerCAmelCase ,     builder.name ,     """default""" ,     """0.0.0""" ,     """dataset_info.json"""      )      )      )
                  del dset



    @require_beam
    def _a							(				self      :     Optional[Any]      )      ->							Optional[int]:
           with tempfile.TemporaryDirectory() as tmp_cache_dir:
                  __UpperCAmelCase     				=DummyBeamDataset(cache_dir=__lowerCAmelCase      )
                  self.assertRaises(datasets.builder.MissingBeamOptions ,     builder.download_and_prepare      )



    @require_beam
    def _a							(				self      :     Any      )      ->							int:
           __UpperCAmelCase     				=len(get_test_nested_examples()      )
           with tempfile.TemporaryDirectory() as tmp_cache_dir:
                  __UpperCAmelCase     				=NestedBeamDataset(cache_dir=__lowerCAmelCase ,     beam_runner="""DirectRunner"""      )
                  builder.download_and_prepare()
                  self.assertTrue(
                      os.path.exists(
                          os.path.join(__lowerCAmelCase ,     builder.name ,     """default""" ,     """0.0.0""" ,     f'''{builder.name}-train.arrow'''      )      )      )
                  self.assertDictEqual(
                      builder.info.features ,     datasets.Features({"""a""": datasets.Sequence({"""b""": datasets.Value("""string"""      )}      )}      )      )
                  __UpperCAmelCase     				=builder.as_dataset()
                  self.assertEqual(dset["""train"""].num_rows ,     __lowerCAmelCase      )
                  self.assertEqual(dset["""train"""].info.splits["""train"""].num_examples ,     __lowerCAmelCase      )
                  self.assertDictEqual(dset["""train"""][0] ,     get_test_nested_examples()[0][1]      )
                  self.assertDictEqual(
                      dset["""train"""][expected_num_examples - 1] ,     get_test_nested_examples()[expected_num_examples - 1][1]      )
                  self.assertTrue(
                      os.path.exists(os.path.join(__lowerCAmelCase ,     builder.name ,     """default""" ,     """0.0.0""" ,     """dataset_info.json"""      )      )      )
                  del dset









def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:int				)  ->     int:
   if not isinstance(_snake_case	,   _snake_case				):
      raise ValueError('''Input must be an integer'''				)
   if input_num <= 0:
      raise ValueError('''Input must be positive'''				)
   return sum(
       divisor for divisor in range(1	,   input_num // 2 + 1				) if input_num % divisor == 0				)


if __name__ == "__main__":
   import doctest

   doctest.testmod()



