def _read_stream()

in awswrangler/distributed/ray/datasources/arrow_csv_datasource.py [0:0]

• 23 lines of code
• 5 McCabe index (conditional complexity)

    def _read_stream(self, f: pa.NativeFile, path: str) -> Iterator[pa.Table]:
        reader = csv.open_csv(
            f,
            read_options=self.read_options,
            parse_options=self.parse_options,
            convert_options=self.convert_options,
        )

        schema = None
        while True:
            try:
                batch = reader.read_next_batch()
                table = pa.Table.from_batches([batch], schema=schema)
                if schema is None:
                    schema = table.schema

                if self.dataset:
                    table = _add_table_partitions(
                        table=table,
                        path=f"s3://{path}",
                        path_root=self.path_root,
                    )

                yield table

            except StopIteration:
                return