From 60b189b5541e4890db254f2343601422bb4b3a39 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 5 Jun 2026 02:25:49 +0000 Subject: [PATCH 1/5] fix(file-based): allow safe trailing empty CSV headers --- .../file_based/file_types/csv_parser.py | 100 ++++++++++++-- .../file_based/file_types/test_csv_parser.py | 123 +++++++++++++++++- 2 files changed, 207 insertions(+), 16 deletions(-) diff --git a/airbyte_cdk/sources/file_based/file_types/csv_parser.py b/airbyte_cdk/sources/file_based/file_types/csv_parser.py index 3b3dc4a0f7..2fb317efcb 100644 --- a/airbyte_cdk/sources/file_based/file_types/csv_parser.py +++ b/airbyte_cdk/sources/file_based/file_types/csv_parser.py @@ -70,7 +70,9 @@ def read_data( file, file_read_mode, config_format.encoding, logger ) as fp: try: - headers = self._get_headers(fp, config_format, dialect_name) + headers, trailing_empty_header_count = ( + self._get_headers_and_trailing_empty_count(fp, config_format, dialect_name) + ) except UnicodeError: raise AirbyteTracedException( message=f"{FileBasedSourceError.ENCODING_ERROR.value} Expected encoding: {config_format.encoding}", @@ -84,10 +86,19 @@ def read_data( self._skip_rows(fp, rows_to_skip) lineno += rows_to_skip - reader = csv.DictReader(fp, dialect=dialect_name, fieldnames=headers) # type: ignore - for row in reader: + reader = csv.reader(fp, dialect=dialect_name) # type: ignore + for values in reader: lineno += 1 + values = self._strip_trailing_empty_header_values( + values, + headers, + trailing_empty_header_count, + file.uri, + lineno, + ) + row = self._row_to_dict(headers, values) + # The row was not properly parsed if any of the values are None. This will most likely occur if there are more columns # than headers or more headers dans columns if None in row: @@ -117,32 +128,103 @@ def read_data( csv.unregister_dialect(dialect_name) def _get_headers(self, fp: IOBase, config_format: CsvFormat, dialect_name: str) -> List[str]: + headers, _ = self._get_headers_and_trailing_empty_count(fp, config_format, dialect_name) + return headers + + def _get_headers_and_trailing_empty_count( + self, fp: IOBase, config_format: CsvFormat, dialect_name: str + ) -> Tuple[List[str], int]: """Assumes the fp is pointing to the beginning of the files and will reset it as such.""" # Note that this method assumes the dialect has already been registered if we're parsing the headers if isinstance(config_format.header_definition, CsvHeaderUserProvided): - return config_format.header_definition.column_names + return config_format.header_definition.column_names, 0 if isinstance(config_format.header_definition, CsvHeaderAutogenerated): self._skip_rows( fp, config_format.skip_rows_before_header + config_format.skip_rows_after_header ) headers = self._auto_generate_headers(fp, dialect_name) + trailing_empty_header_count = 0 else: # Then read the header self._skip_rows(fp, config_format.skip_rows_before_header) reader = csv.reader(fp, dialect=dialect_name) # type: ignore headers = list(next(reader)) - empty_count = sum(1 for h in headers if not h or h.isspace()) - if empty_count: + trailing_empty_header_count = self._get_trailing_empty_header_count(headers) + headers_without_trailing_empty_values = ( + headers[: len(headers) - trailing_empty_header_count] + if trailing_empty_header_count + else headers + ) + interior_empty_count = sum( + 1 for h in headers_without_trailing_empty_values if self._is_empty_csv_value(h) + ) + if interior_empty_count: + raise AirbyteTracedException( + message="CSV header row contains interior empty column name(s).", + internal_message=f"Found {interior_empty_count} empty/whitespace-only column name(s) before trailing blank columns in header: {headers}", + failure_type=FailureType.config_error, + ) + if trailing_empty_header_count == len(headers): raise AirbyteTracedException( - message="CSV header row contains empty column name(s). Remove trailing delimiters or empty columns from the header row.", - internal_message=f"Found {empty_count} empty/whitespace-only column name(s) in header: {headers}", + message="CSV header row contains only empty column name(s).", + internal_message=f"Found {trailing_empty_header_count} empty/whitespace-only column name(s) in header: {headers}", failure_type=FailureType.config_error, ) + headers = headers_without_trailing_empty_values fp.seek(0) - return headers + return headers, trailing_empty_header_count + + def _strip_trailing_empty_header_values( + self, + values: List[str], + headers: List[str], + trailing_empty_header_count: int, + filename: str, + lineno: int, + ) -> List[str]: + if not trailing_empty_header_count: + return values + + strip_start = len(headers) + strip_end = strip_start + trailing_empty_header_count + trailing_values = values[strip_start:strip_end] + non_empty_trailing_values = [ + value for value in trailing_values if not self._is_empty_csv_value(value) + ] + if non_empty_trailing_values: + raise AirbyteTracedException( + message="CSV data row contains non-empty value(s) under trailing empty header column(s).", + internal_message=f"Found {len(non_empty_trailing_values)} non-empty value(s) in stripped trailing column(s) on line {lineno} of file {filename}.", + failure_type=FailureType.config_error, + ) + + return values[:strip_start] + values[strip_end:] + + @staticmethod + def _row_to_dict(headers: List[str], values: List[str]) -> Dict[Any, Any]: + row: Dict[Any, Any] = dict(zip(headers, values)) + if len(values) > len(headers): + row[None] = values[len(headers) :] + elif len(values) < len(headers): + for header in headers[len(values) :]: + row[header] = None + return row + + @staticmethod + def _get_trailing_empty_header_count(headers: List[str]) -> int: + trailing_empty_header_count = 0 + for header in reversed(headers): + if not _CsvReader._is_empty_csv_value(header): + break + trailing_empty_header_count += 1 + return trailing_empty_header_count + + @staticmethod + def _is_empty_csv_value(value: Optional[str]) -> bool: + return not value or value.isspace() def _auto_generate_headers(self, fp: IOBase, dialect_name: str) -> List[str]: """ diff --git a/unit_tests/sources/file_based/file_types/test_csv_parser.py b/unit_tests/sources/file_based/file_types/test_csv_parser.py index 3238a96bb3..7f9b9932c0 100644 --- a/unit_tests/sources/file_based/file_types/test_csv_parser.py +++ b/unit_tests/sources/file_based/file_types/test_csv_parser.py @@ -648,7 +648,7 @@ def test_read_data_with_encoding_error(self) -> None: self._stream_reader.open_file.return_value = ( CsvFileBuilder().with_data(["something"]).build() ) - self._csv_reader._get_headers = Mock( + self._csv_reader._get_headers_and_trailing_empty_count = Mock( side_effect=UnicodeDecodeError("encoding", b"", 0, 1, "reason") ) @@ -657,7 +657,7 @@ def test_read_data_with_encoding_error(self) -> None: assert len(list(data_generator)) == 0 assert "encoding" in ate.value.message - assert self._csv_reader._get_headers.called + assert self._csv_reader._get_headers_and_trailing_empty_count.called def _read_data(self) -> Generator[Dict[str, str], None, None]: data_generator = self._csv_reader.read_data( @@ -791,10 +791,10 @@ def test_encoding_is_passed_to_stream_reader() -> None: @pytest.mark.parametrize( "header_row, expected_empty_count", [ - pytest.param("col1,col2,col3,,,", 3, id="trailing_empty_columns"), pytest.param("col1,,col3", 1, id="middle_empty_column"), pytest.param(",col2,col3", 1, id="leading_empty_column"), - pytest.param("col1,col2, ", 1, id="whitespace_only_column"), + pytest.param("col1, ,col3", 1, id="middle_whitespace_only_column"), + pytest.param(" , , ", 3, id="only_empty_columns"), ], ) def test_get_headers_raises_on_empty_column_names( @@ -825,6 +825,41 @@ def test_get_headers_raises_on_empty_column_names( csv.unregister_dialect(dialect_name) +@pytest.mark.parametrize( + "header_row, expected_headers", + [ + pytest.param("col1,col2,col3,,,", ["col1", "col2", "col3"], id="empty_headers"), + pytest.param( + "col1,col2,col3, , ", + ["col1", "col2", "col3"], + id="whitespace_only_headers", + ), + ], +) +def test_get_headers_strips_trailing_empty_column_names( + header_row: str, expected_headers: List[str] +) -> None: + csv_reader = _CsvReader() + config_format = CsvFormat() + fp = io.StringIO(header_row) + + dialect_name = f"test_{uuid4()}" + csv.register_dialect( + dialect_name, + delimiter=config_format.delimiter, + quotechar=config_format.quote_char, + escapechar=config_format.escape_char, + doublequote=config_format.double_quote, + quoting=csv.QUOTE_MINIMAL, + ) + + try: + headers = csv_reader._get_headers(fp, config_format, dialect_name) + assert headers == expected_headers + finally: + csv.unregister_dialect(dialect_name) + + def test_get_headers_accepts_valid_headers() -> None: csv_reader = _CsvReader() config_format = CsvFormat() @@ -847,7 +882,81 @@ def test_get_headers_accepts_valid_headers() -> None: csv.unregister_dialect(dialect_name) -def test_read_data_raises_on_empty_column_names() -> None: +@pytest.mark.parametrize( + "data, expected_records", + [ + pytest.param( + ["col1,col2,col3,,", "v1,v2,v3,,", "v4,v5,v6"], + [ + {"col1": "v1", "col2": "v2", "col3": "v3"}, + {"col1": "v4", "col2": "v5", "col3": "v6"}, + ], + id="trailing_empty_headers_and_cells", + ), + pytest.param( + ["col1,col2,col3, , ", "v1,v2,v3, ,", "v4,v5,v6"], + [ + {"col1": "v1", "col2": "v2", "col3": "v3"}, + {"col1": "v4", "col2": "v5", "col3": "v6"}, + ], + id="trailing_whitespace_headers_and_cells", + ), + pytest.param( + ["col1,col2,col3", "v1,v2,v3"], + [{"col1": "v1", "col2": "v2", "col3": "v3"}], + id="valid_headers", + ), + ], +) +def test_read_data_strips_trailing_empty_headers_and_values( + data: List[str], expected_records: List[Dict[str, str]] +) -> None: + config_format = CsvFormat() + config = Mock() + config.name = "config_name" + config.format = config_format + + file = RemoteFile(uri="test.csv", last_modified=datetime.now()) + stream_reader = Mock(spec=AbstractFileBasedStreamReader) + logger = Mock(spec=logging.Logger) + csv_reader = _CsvReader() + + stream_reader.open_file.return_value = CsvFileBuilder().with_data(data).build() + + assert ( + list(csv_reader.read_data(config, file, stream_reader, logger, FileReadMode.READ)) + == expected_records + ) + + +@pytest.mark.parametrize( + "data", + [ + pytest.param(["col1,,col3", "v1,,v3"], id="interior_empty_header"), + pytest.param(["col1, ,col3", "v1, ,v3"], id="interior_whitespace_header"), + ], +) +def test_read_data_raises_on_interior_empty_column_names(data: List[str]) -> None: + config_format = CsvFormat() + config = Mock() + config.name = "config_name" + config.format = config_format + + file = RemoteFile(uri="test.csv", last_modified=datetime.now()) + stream_reader = Mock(spec=AbstractFileBasedStreamReader) + logger = Mock(spec=logging.Logger) + csv_reader = _CsvReader() + + stream_reader.open_file.return_value = CsvFileBuilder().with_data(data).build() + + with pytest.raises(AirbyteTracedException) as exc_info: + list(csv_reader.read_data(config, file, stream_reader, logger, FileReadMode.READ)) + + assert exc_info.value.failure_type == FailureType.config_error + assert "interior empty column name" in exc_info.value.message + + +def test_read_data_raises_on_non_empty_values_under_trailing_empty_headers() -> None: config_format = CsvFormat() config = Mock() config.name = "config_name" @@ -859,7 +968,7 @@ def test_read_data_raises_on_empty_column_names() -> None: csv_reader = _CsvReader() stream_reader.open_file.return_value = ( - CsvFileBuilder().with_data(["col1,col2,col3,,,", "v1,v2,v3,v4,v5,v6"]).build() + CsvFileBuilder().with_data(["col1,col2,col3,,", "v1,v2,v3,v4,"]).build() ) with pytest.raises(AirbyteTracedException) as exc_info: @@ -874,7 +983,7 @@ def test_read_data_raises_on_empty_column_names() -> None: ) assert exc_info.value.failure_type == FailureType.config_error - assert "empty column name" in exc_info.value.message + assert "non-empty value" in exc_info.value.message def test_parse_records_preserves_mismatch_error_detail() -> None: From 350f8d89ae454221177b6b244823a03f13456624 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 8 Jun 2026 13:53:31 +0000 Subject: [PATCH 2/5] refactor(csv-parser): remove redundant _get_headers_and_trailing_empty_count Consolidate logic into _get_headers (returns List[str]) and call _get_trailing_empty_header_count separately in read_data where the trailing count is needed. This eliminates the unnecessary wrapper method while keeping the interface clean. Co-Authored-By: bot_apk --- .../file_based/file_types/csv_parser.py | 35 +++++++++++-------- .../file_based/file_types/test_csv_parser.py | 4 +-- 2 files changed, 22 insertions(+), 17 deletions(-) diff --git a/airbyte_cdk/sources/file_based/file_types/csv_parser.py b/airbyte_cdk/sources/file_based/file_types/csv_parser.py index 2fb317efcb..3c180bd0f4 100644 --- a/airbyte_cdk/sources/file_based/file_types/csv_parser.py +++ b/airbyte_cdk/sources/file_based/file_types/csv_parser.py @@ -70,14 +70,25 @@ def read_data( file, file_read_mode, config_format.encoding, logger ) as fp: try: - headers, trailing_empty_header_count = ( - self._get_headers_and_trailing_empty_count(fp, config_format, dialect_name) - ) + headers = self._get_headers(fp, config_format, dialect_name) except UnicodeError: raise AirbyteTracedException( message=f"{FileBasedSourceError.ENCODING_ERROR.value} Expected encoding: {config_format.encoding}", ) + # Compute trailing empty header count for data row stripping. + # _get_headers already stripped trailing empties from the returned headers, + # but we need the count to strip corresponding values from data rows. + trailing_empty_header_count = 0 + if not isinstance( + config_format.header_definition, (CsvHeaderUserProvided, CsvHeaderAutogenerated) + ): + self._skip_rows(fp, config_format.skip_rows_before_header) + raw_reader = csv.reader(fp, dialect=dialect_name) # type: ignore + raw_headers = list(next(raw_reader)) + trailing_empty_header_count = self._get_trailing_empty_header_count(raw_headers) + fp.seek(0) + rows_to_skip = ( config_format.skip_rows_before_header + (1 if config_format.header_definition.has_header_row() else 0) @@ -128,23 +139,17 @@ def read_data( csv.unregister_dialect(dialect_name) def _get_headers(self, fp: IOBase, config_format: CsvFormat, dialect_name: str) -> List[str]: - headers, _ = self._get_headers_and_trailing_empty_count(fp, config_format, dialect_name) - return headers - - def _get_headers_and_trailing_empty_count( - self, fp: IOBase, config_format: CsvFormat, dialect_name: str - ) -> Tuple[List[str], int]: """Assumes the fp is pointing to the beginning of the files and will reset it as such.""" # Note that this method assumes the dialect has already been registered if we're parsing the headers if isinstance(config_format.header_definition, CsvHeaderUserProvided): - return config_format.header_definition.column_names, 0 + fp.seek(0) + return config_format.header_definition.column_names if isinstance(config_format.header_definition, CsvHeaderAutogenerated): self._skip_rows( fp, config_format.skip_rows_before_header + config_format.skip_rows_after_header ) headers = self._auto_generate_headers(fp, dialect_name) - trailing_empty_header_count = 0 else: # Then read the header self._skip_rows(fp, config_format.skip_rows_before_header) @@ -152,13 +157,13 @@ def _get_headers_and_trailing_empty_count( headers = list(next(reader)) trailing_empty_header_count = self._get_trailing_empty_header_count(headers) - headers_without_trailing_empty_values = ( + headers_without_trailing = ( headers[: len(headers) - trailing_empty_header_count] if trailing_empty_header_count else headers ) interior_empty_count = sum( - 1 for h in headers_without_trailing_empty_values if self._is_empty_csv_value(h) + 1 for h in headers_without_trailing if self._is_empty_csv_value(h) ) if interior_empty_count: raise AirbyteTracedException( @@ -172,10 +177,10 @@ def _get_headers_and_trailing_empty_count( internal_message=f"Found {trailing_empty_header_count} empty/whitespace-only column name(s) in header: {headers}", failure_type=FailureType.config_error, ) - headers = headers_without_trailing_empty_values + headers = headers_without_trailing fp.seek(0) - return headers, trailing_empty_header_count + return headers def _strip_trailing_empty_header_values( self, diff --git a/unit_tests/sources/file_based/file_types/test_csv_parser.py b/unit_tests/sources/file_based/file_types/test_csv_parser.py index 7f9b9932c0..cafd673ed7 100644 --- a/unit_tests/sources/file_based/file_types/test_csv_parser.py +++ b/unit_tests/sources/file_based/file_types/test_csv_parser.py @@ -648,7 +648,7 @@ def test_read_data_with_encoding_error(self) -> None: self._stream_reader.open_file.return_value = ( CsvFileBuilder().with_data(["something"]).build() ) - self._csv_reader._get_headers_and_trailing_empty_count = Mock( + self._csv_reader._get_headers = Mock( side_effect=UnicodeDecodeError("encoding", b"", 0, 1, "reason") ) @@ -657,7 +657,7 @@ def test_read_data_with_encoding_error(self) -> None: assert len(list(data_generator)) == 0 assert "encoding" in ate.value.message - assert self._csv_reader._get_headers_and_trailing_empty_count.called + assert self._csv_reader._get_headers.called def _read_data(self) -> Generator[Dict[str, str], None, None]: data_generator = self._csv_reader.read_data( From 1707f08e15d566d1837c4b75d2fc13e707bb5edf Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 8 Jun 2026 14:12:21 +0000 Subject: [PATCH 3/5] refactor(csv-parser): consolidate trailing header logic into _validate_trailing_headers Extract _validate_trailing_headers method that encapsulates all trailing header validation logic (count, strip, interior/all-empty checks) in one place. Add _read_and_validate_headers as the single entry point that reads raw headers from file and validates them. _get_headers delegates to it. Co-Authored-By: bot_apk --- .../file_based/file_types/csv_parser.py | 89 ++++++++++--------- .../file_based/file_types/test_csv_parser.py | 4 +- 2 files changed, 47 insertions(+), 46 deletions(-) diff --git a/airbyte_cdk/sources/file_based/file_types/csv_parser.py b/airbyte_cdk/sources/file_based/file_types/csv_parser.py index 3c180bd0f4..f92c2f6354 100644 --- a/airbyte_cdk/sources/file_based/file_types/csv_parser.py +++ b/airbyte_cdk/sources/file_based/file_types/csv_parser.py @@ -70,25 +70,14 @@ def read_data( file, file_read_mode, config_format.encoding, logger ) as fp: try: - headers = self._get_headers(fp, config_format, dialect_name) + headers, trailing_empty_header_count = self._read_and_validate_headers( + fp, config_format, dialect_name + ) except UnicodeError: raise AirbyteTracedException( message=f"{FileBasedSourceError.ENCODING_ERROR.value} Expected encoding: {config_format.encoding}", ) - # Compute trailing empty header count for data row stripping. - # _get_headers already stripped trailing empties from the returned headers, - # but we need the count to strip corresponding values from data rows. - trailing_empty_header_count = 0 - if not isinstance( - config_format.header_definition, (CsvHeaderUserProvided, CsvHeaderAutogenerated) - ): - self._skip_rows(fp, config_format.skip_rows_before_header) - raw_reader = csv.reader(fp, dialect=dialect_name) # type: ignore - raw_headers = list(next(raw_reader)) - trailing_empty_header_count = self._get_trailing_empty_header_count(raw_headers) - fp.seek(0) - rows_to_skip = ( config_format.skip_rows_before_header + (1 if config_format.header_definition.has_header_row() else 0) @@ -140,47 +129,59 @@ def read_data( def _get_headers(self, fp: IOBase, config_format: CsvFormat, dialect_name: str) -> List[str]: """Assumes the fp is pointing to the beginning of the files and will reset it as such.""" - # Note that this method assumes the dialect has already been registered if we're parsing the headers + headers, _ = self._read_and_validate_headers(fp, config_format, dialect_name) + return headers + + def _read_and_validate_headers( + self, fp: IOBase, config_format: CsvFormat, dialect_name: str + ) -> Tuple[List[str], int]: + """Reads headers from file, validates, strips trailing empties, and resets fp to 0.""" if isinstance(config_format.header_definition, CsvHeaderUserProvided): fp.seek(0) - return config_format.header_definition.column_names + return config_format.header_definition.column_names, 0 if isinstance(config_format.header_definition, CsvHeaderAutogenerated): self._skip_rows( fp, config_format.skip_rows_before_header + config_format.skip_rows_after_header ) headers = self._auto_generate_headers(fp, dialect_name) - else: - # Then read the header - self._skip_rows(fp, config_format.skip_rows_before_header) - reader = csv.reader(fp, dialect=dialect_name) # type: ignore - headers = list(next(reader)) - - trailing_empty_header_count = self._get_trailing_empty_header_count(headers) - headers_without_trailing = ( - headers[: len(headers) - trailing_empty_header_count] - if trailing_empty_header_count - else headers + fp.seek(0) + return headers, 0 + + self._skip_rows(fp, config_format.skip_rows_before_header) + reader = csv.reader(fp, dialect=dialect_name) # type: ignore + raw_headers = list(next(reader)) + + headers, trailing_empty_header_count = self._validate_trailing_headers(raw_headers) + + fp.seek(0) + return headers, trailing_empty_header_count + + def _validate_trailing_headers(self, raw_headers: List[str]) -> Tuple[List[str], int]: + """Validates header names and strips trailing empty headers.""" + trailing_empty_header_count = self._get_trailing_empty_header_count(raw_headers) + headers = ( + raw_headers[: len(raw_headers) - trailing_empty_header_count] + if trailing_empty_header_count + else raw_headers + ) + + if trailing_empty_header_count == len(raw_headers): + raise AirbyteTracedException( + message="CSV header row contains only empty column name(s).", + internal_message=f"Found {trailing_empty_header_count} empty/whitespace-only column name(s) in header: {raw_headers}", + failure_type=FailureType.config_error, ) - interior_empty_count = sum( - 1 for h in headers_without_trailing if self._is_empty_csv_value(h) + + interior_empty_count = sum(1 for h in headers if self._is_empty_csv_value(h)) + if interior_empty_count: + raise AirbyteTracedException( + message="CSV header row contains interior empty column name(s).", + internal_message=f"Found {interior_empty_count} empty/whitespace-only column name(s) before trailing blank columns in header: {raw_headers}", + failure_type=FailureType.config_error, ) - if interior_empty_count: - raise AirbyteTracedException( - message="CSV header row contains interior empty column name(s).", - internal_message=f"Found {interior_empty_count} empty/whitespace-only column name(s) before trailing blank columns in header: {headers}", - failure_type=FailureType.config_error, - ) - if trailing_empty_header_count == len(headers): - raise AirbyteTracedException( - message="CSV header row contains only empty column name(s).", - internal_message=f"Found {trailing_empty_header_count} empty/whitespace-only column name(s) in header: {headers}", - failure_type=FailureType.config_error, - ) - headers = headers_without_trailing - fp.seek(0) - return headers + return headers, trailing_empty_header_count def _strip_trailing_empty_header_values( self, diff --git a/unit_tests/sources/file_based/file_types/test_csv_parser.py b/unit_tests/sources/file_based/file_types/test_csv_parser.py index cafd673ed7..06781c127a 100644 --- a/unit_tests/sources/file_based/file_types/test_csv_parser.py +++ b/unit_tests/sources/file_based/file_types/test_csv_parser.py @@ -648,7 +648,7 @@ def test_read_data_with_encoding_error(self) -> None: self._stream_reader.open_file.return_value = ( CsvFileBuilder().with_data(["something"]).build() ) - self._csv_reader._get_headers = Mock( + self._csv_reader._read_and_validate_headers = Mock( side_effect=UnicodeDecodeError("encoding", b"", 0, 1, "reason") ) @@ -657,7 +657,7 @@ def test_read_data_with_encoding_error(self) -> None: assert len(list(data_generator)) == 0 assert "encoding" in ate.value.message - assert self._csv_reader._get_headers.called + assert self._csv_reader._read_and_validate_headers.called def _read_data(self) -> Generator[Dict[str, str], None, None]: data_generator = self._csv_reader.read_data( From d79be6cb2ea922551f3526e748c864da9338764a Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 8 Jun 2026 14:54:49 +0000 Subject: [PATCH 4/5] refactor(csv-parser): move trailing count computation into _strip_trailing_empty_header_values _strip_trailing_empty_header_values now calls _get_trailing_empty_header_count internally on the raw_headers, removing the need for read_data to track trailing_empty_header_count separately. _read_and_validate_headers returns (clean_headers, raw_headers) instead of (clean_headers, trailing_count). _validate_trailing_headers returns just the clean headers list. Co-Authored-By: bot_apk --- .../file_based/file_types/csv_parser.py | 28 +++++++++++-------- 1 file changed, 17 insertions(+), 11 deletions(-) diff --git a/airbyte_cdk/sources/file_based/file_types/csv_parser.py b/airbyte_cdk/sources/file_based/file_types/csv_parser.py index f92c2f6354..2db87f80b6 100644 --- a/airbyte_cdk/sources/file_based/file_types/csv_parser.py +++ b/airbyte_cdk/sources/file_based/file_types/csv_parser.py @@ -70,7 +70,7 @@ def read_data( file, file_read_mode, config_format.encoding, logger ) as fp: try: - headers, trailing_empty_header_count = self._read_and_validate_headers( + headers, raw_headers = self._read_and_validate_headers( fp, config_format, dialect_name ) except UnicodeError: @@ -93,7 +93,7 @@ def read_data( values = self._strip_trailing_empty_header_values( values, headers, - trailing_empty_header_count, + raw_headers, file.uri, lineno, ) @@ -134,11 +134,16 @@ def _get_headers(self, fp: IOBase, config_format: CsvFormat, dialect_name: str) def _read_and_validate_headers( self, fp: IOBase, config_format: CsvFormat, dialect_name: str - ) -> Tuple[List[str], int]: - """Reads headers from file, validates, strips trailing empties, and resets fp to 0.""" + ) -> Tuple[List[str], List[str]]: + """Reads headers from file, validates, strips trailing empties, and resets fp to 0. + + Returns (clean_headers, raw_headers). For user-provided or autogenerated headers, + raw_headers == clean_headers (no trailing empties to strip). + """ if isinstance(config_format.header_definition, CsvHeaderUserProvided): fp.seek(0) - return config_format.header_definition.column_names, 0 + columns = config_format.header_definition.column_names + return columns, columns if isinstance(config_format.header_definition, CsvHeaderAutogenerated): self._skip_rows( @@ -146,18 +151,18 @@ def _read_and_validate_headers( ) headers = self._auto_generate_headers(fp, dialect_name) fp.seek(0) - return headers, 0 + return headers, headers self._skip_rows(fp, config_format.skip_rows_before_header) reader = csv.reader(fp, dialect=dialect_name) # type: ignore raw_headers = list(next(reader)) - headers, trailing_empty_header_count = self._validate_trailing_headers(raw_headers) + headers = self._validate_trailing_headers(raw_headers) fp.seek(0) - return headers, trailing_empty_header_count + return headers, raw_headers - def _validate_trailing_headers(self, raw_headers: List[str]) -> Tuple[List[str], int]: + def _validate_trailing_headers(self, raw_headers: List[str]) -> List[str]: """Validates header names and strips trailing empty headers.""" trailing_empty_header_count = self._get_trailing_empty_header_count(raw_headers) headers = ( @@ -181,16 +186,17 @@ def _validate_trailing_headers(self, raw_headers: List[str]) -> Tuple[List[str], failure_type=FailureType.config_error, ) - return headers, trailing_empty_header_count + return headers def _strip_trailing_empty_header_values( self, values: List[str], headers: List[str], - trailing_empty_header_count: int, + raw_headers: List[str], filename: str, lineno: int, ) -> List[str]: + trailing_empty_header_count = self._get_trailing_empty_header_count(raw_headers) if not trailing_empty_header_count: return values From 7cbeb498791842cac801737171a3b9b56eaf06b7 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 8 Jun 2026 15:27:14 +0000 Subject: [PATCH 5/5] fix: skip blank CSV rows and optimize trailing header count - Add blank-row skip (if not values: continue) to match DictReader behavior - Precompute trailing_empty_header_count once before the loop instead of per-row - Add test coverage for blank lines mid-file and trailing blank lines Co-Authored-By: bot_apk --- .../sources/file_based/file_types/csv_parser.py | 8 +++++--- .../file_based/file_types/test_csv_parser.py | 13 +++++++++++++ 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/airbyte_cdk/sources/file_based/file_types/csv_parser.py b/airbyte_cdk/sources/file_based/file_types/csv_parser.py index 2db87f80b6..b8c720642a 100644 --- a/airbyte_cdk/sources/file_based/file_types/csv_parser.py +++ b/airbyte_cdk/sources/file_based/file_types/csv_parser.py @@ -87,13 +87,16 @@ def read_data( lineno += rows_to_skip reader = csv.reader(fp, dialect=dialect_name) # type: ignore + trailing_empty_header_count = self._get_trailing_empty_header_count(raw_headers) for values in reader: lineno += 1 + if not values: + continue values = self._strip_trailing_empty_header_values( values, headers, - raw_headers, + trailing_empty_header_count, file.uri, lineno, ) @@ -192,11 +195,10 @@ def _strip_trailing_empty_header_values( self, values: List[str], headers: List[str], - raw_headers: List[str], + trailing_empty_header_count: int, filename: str, lineno: int, ) -> List[str]: - trailing_empty_header_count = self._get_trailing_empty_header_count(raw_headers) if not trailing_empty_header_count: return values diff --git a/unit_tests/sources/file_based/file_types/test_csv_parser.py b/unit_tests/sources/file_based/file_types/test_csv_parser.py index 06781c127a..f12f6613e8 100644 --- a/unit_tests/sources/file_based/file_types/test_csv_parser.py +++ b/unit_tests/sources/file_based/file_types/test_csv_parser.py @@ -906,6 +906,19 @@ def test_get_headers_accepts_valid_headers() -> None: [{"col1": "v1", "col2": "v2", "col3": "v3"}], id="valid_headers", ), + pytest.param( + ["col1,col2", "v1,v2", "", "v3,v4"], + [ + {"col1": "v1", "col2": "v2"}, + {"col1": "v3", "col2": "v4"}, + ], + id="blank_line_mid_file", + ), + pytest.param( + ["col1,col2", "v1,v2", ""], + [{"col1": "v1", "col2": "v2"}], + id="trailing_blank_line", + ), ], ) def test_read_data_strips_trailing_empty_headers_and_values(