diff --git a/airbyte_cdk/sources/file_based/config/file_based_stream_config.py b/airbyte_cdk/sources/file_based/config/file_based_stream_config.py index 4946fb6266..7f4cb60f4b 100644 --- a/airbyte_cdk/sources/file_based/config/file_based_stream_config.py +++ b/airbyte_cdk/sources/file_based/config/file_based_stream_config.py @@ -30,7 +30,7 @@ class FileBasedStreamConfig(BaseModel): globs: Optional[List[str]] = Field( default=["**"], title="Globs", - description='The pattern used to specify which files should be selected from the file system. For more information on glob pattern matching look here.', + description='The pattern used to specify which files should be selected from the file system. Patterns prefixed with `!` exclude matching files. For more information on glob pattern matching look here.', order=1, ) legacy_prefix: Optional[str] = Field( diff --git a/airbyte_cdk/sources/file_based/file_based_stream_reader.py b/airbyte_cdk/sources/file_based/file_based_stream_reader.py index 30dc588e73..62ce8cb0d4 100644 --- a/airbyte_cdk/sources/file_based/file_based_stream_reader.py +++ b/airbyte_cdk/sources/file_based/file_based_stream_reader.py @@ -13,7 +13,7 @@ from typing import Any, Iterable, List, MutableMapping, Optional, Set, Tuple from airbyte_protocol_dataclasses.models import FailureType -from wcmatch.glob import GLOBSTAR, globmatch +from wcmatch.glob import GLOBSTAR, NEGATE, globmatch from airbyte_cdk.models import AirbyteRecordMessageFileReference from airbyte_cdk.sources.file_based.config.abstract_file_based_spec import AbstractFileBasedSpec @@ -150,14 +150,15 @@ def filter_files_by_globs_and_start_date( def file_matches_globs(file: RemoteFile, globs: List[str]) -> bool: # Use the GLOBSTAR flag to enable recursive ** matching # (https://facelessuser.github.io/wcmatch/wcmatch/#globstar) - return any(globmatch(file.uri, g, flags=GLOBSTAR) for g in globs) + # and the NEGATE flag so that patterns starting with `!` exclude matching files + return globmatch(file.uri, globs, flags=GLOBSTAR | NEGATE) @staticmethod def get_prefixes_from_globs(globs: List[str]) -> Set[str]: """ Utility method for extracting prefixes from the globs. """ - prefixes = {glob.split("*")[0] for glob in globs} + prefixes = {glob.split("*")[0] for glob in globs if not glob.startswith("!")} return set(filter(lambda x: bool(x), prefixes)) def use_file_transfer(self) -> bool: 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 b8c720642a..3a1bbd4c1f 100644 --- a/airbyte_cdk/sources/file_based/file_types/csv_parser.py +++ b/airbyte_cdk/sources/file_based/file_types/csv_parser.py @@ -22,11 +22,7 @@ InferenceType, ) from airbyte_cdk.sources.file_based.config.file_based_stream_config import FileBasedStreamConfig -from airbyte_cdk.sources.file_based.exceptions import ( - EmptyFileSchemaInferenceError, - FileBasedSourceError, - RecordParseError, -) +from airbyte_cdk.sources.file_based.exceptions import FileBasedSourceError, RecordParseError from airbyte_cdk.sources.file_based.file_based_stream_reader import ( AbstractFileBasedStreamReader, FileReadMode, @@ -158,7 +154,11 @@ def _read_and_validate_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)) + try: + raw_headers = list(next(reader)) + except StopIteration: + # Empty file (no bytes at all): no header row to parse + return [], [] headers = self._validate_trailing_headers(raw_headers) @@ -246,7 +246,11 @@ def _auto_generate_headers(self, fp: IOBase, dialect_name: str) -> List[str]: See https://arrow.apache.org/docs/python/generated/pyarrow.csv.ReadOptions.html """ reader = csv.reader(fp, dialect=dialect_name) # type: ignore - number_of_columns = len(next(reader)) # type: ignore + try: + number_of_columns = len(next(reader)) # type: ignore + except StopIteration: + # Empty file (no bytes at all): no row to autogenerate headers from + return [] return [f"f{i}" for i in range(number_of_columns)] @staticmethod @@ -310,11 +314,11 @@ async def infer_schema( break if not type_inferrer_by_field: - raise EmptyFileSchemaInferenceError( - message=f"Could not infer schema as there are no rows in {file.uri}. If having an empty CSV file is expected, ignore this. " - f"Else, please contact Airbyte.", - failure_type=FailureType.config_error, + logger.warning( + f"Could not infer schema from {file.uri} as it has no rows; skipping it for schema inference." ) + data_generator.close() + return {} schema = { header.strip(): {"type": type_inferred.infer()} for header, type_inferred in type_inferrer_by_field.items() diff --git a/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py b/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py index 588c4a18e4..8f06748649 100644 --- a/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py +++ b/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py @@ -316,7 +316,8 @@ def _get_raw_json_schema(self) -> JsonSchema: if not inferred_schema: raise InvalidSchemaError( FileBasedSourceError.INVALID_SCHEMA_ERROR, - details=f"Empty schema. Please check that the files are valid for format {self.config.format}", + details=f"Empty schema. Please check that the files are valid for format {self.config.format}." + f" This can happen when all files sampled for schema inference are empty.", stream=self.name, ) 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 f12f6613e8..adf92c4f9b 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 @@ -316,11 +316,32 @@ def test_given_big_file_when_infer_schema_then_stop_early(self) -> None: # since the type is number, we know the string at the end was not considered assert inferred_schema == {self._HEADER_NAME: {"type": "number"}} - def test_given_empty_csv_file_when_infer_schema_then_raise_config_error(self) -> None: - self._csv_reader.read_data.return_value = [] - with pytest.raises(AirbyteTracedException) as exception: - self._infer_schema() - assert exception.value.failure_type == FailureType.config_error + def test_given_empty_csv_file_when_infer_schema_then_return_empty_schema(self) -> None: + self._csv_reader.read_data.return_value = (row for row in []) + inferred_schema = self._infer_schema() + assert inferred_schema == {} + + def test_given_header_only_csv_file_when_infer_schema_then_return_empty_schema(self) -> None: + self._stream_reader.open_file.return_value = ( + CsvFileBuilder().with_data(["col1,col2,col3"]).build() + ) + parser = CsvParser() + loop = asyncio.new_event_loop() + task = loop.create_task( + parser.infer_schema(self._config, self._file, self._stream_reader, self._logger) + ) + loop.run_until_complete(task) + assert task.result() == {} + + def test_given_zero_byte_csv_file_when_infer_schema_then_return_empty_schema(self) -> None: + self._stream_reader.open_file.return_value = io.StringIO("") + parser = CsvParser() + loop = asyncio.new_event_loop() + task = loop.create_task( + parser.infer_schema(self._config, self._file, self._stream_reader, self._logger) + ) + loop.run_until_complete(task) + assert task.result() == {} def _test_infer_schema(self, rows: List[str], expected_type: str) -> None: self._csv_reader.read_data.return_value = ({self._HEADER_NAME: row} for row in rows) diff --git a/unit_tests/sources/file_based/scenarios/csv_scenarios.py b/unit_tests/sources/file_based/scenarios/csv_scenarios.py index f16d83e200..9abb720674 100644 --- a/unit_tests/sources/file_based/scenarios/csv_scenarios.py +++ b/unit_tests/sources/file_based/scenarios/csv_scenarios.py @@ -89,7 +89,7 @@ }, "globs": { "title": "Globs", - "description": 'The pattern used to specify which files should be selected from the file system. For more information on glob pattern matching look here.', + "description": 'The pattern used to specify which files should be selected from the file system. Patterns prefixed with `!` exclude matching files. For more information on glob pattern matching look here.', "type": "array", "items": {"type": "string"}, "order": 1, diff --git a/unit_tests/sources/file_based/stream/test_default_file_based_stream.py b/unit_tests/sources/file_based/stream/test_default_file_based_stream.py index 54394a36d8..3fc0eb6779 100644 --- a/unit_tests/sources/file_based/stream/test_default_file_based_stream.py +++ b/unit_tests/sources/file_based/stream/test_default_file_based_stream.py @@ -227,6 +227,34 @@ def test_override_max_n_files_for_schema_inference_is_respected(self) -> None: } assert self._parser.infer_schema.call_count == 3 + def test_given_one_empty_and_one_non_empty_file_when_infer_schema_then_schema_from_non_empty_file( + self, + ) -> None: + self._discovery_policy.n_concurrent_requests = 1 + self._discovery_policy.get_max_n_files_for_schema_inference.return_value = 3 + self._stream.config.input_schema = None + self._stream.config.schemaless = None + self._stream.config.use_first_found_file_for_schema_discovery = False + self._stream.config.recent_n_files_to_read_for_schema_discovery = None + # Empty CSV files return an empty schema that must not prevent merging the schema of other files + self._parser.infer_schema.side_effect = [{}, {"data": {"type": "string"}}] + files = [ + RemoteFile(uri="empty.csv", last_modified=self._NOW), + RemoteFile(uri="non_empty.csv", last_modified=self._NOW), + ] + self._stream_reader.get_matching_files.return_value = files + + schema = self._stream.get_json_schema() + + assert schema == { + "type": "object", + "properties": { + "_ab_source_file_last_modified": {"type": "string"}, + "_ab_source_file_url": {"type": "string"}, + "data": {"type": ["null", "string"]}, + }, + } + def test_use_first_found_file_for_schema_discovery(self) -> None: self._stream.config.use_first_found_file_for_schema_discovery = True diff --git a/unit_tests/sources/file_based/test_file_based_stream_reader.py b/unit_tests/sources/file_based/test_file_based_stream_reader.py index fc473545cb..0ae9bc832d 100644 --- a/unit_tests/sources/file_based/test_file_based_stream_reader.py +++ b/unit_tests/sources/file_based/test_file_based_stream_reader.py @@ -408,6 +408,27 @@ def documentation_url(cls) -> AnyUrl: set(), id="all_csvs_start_date_without_microseconds", ), + pytest.param( + ["**/*.csv", "!a/b.csv"], + DEFAULT_CONFIG, + {"a.csv", "a/c.csv", "a/b/c.csv", "a/c/c.csv", "a/b/c/d.csv"}, + set(), + id="negation_excludes_matching_file", + ), + pytest.param( + ["a/*.csv", "a/*/*.csv", "!a/b/*.csv"], + DEFAULT_CONFIG, + {"a/b.csv", "a/c.csv", "a/c/c.csv"}, + {"a/"}, + id="negation_excluded_from_prefixes", + ), + pytest.param( + ["!a/*.csv"], + DEFAULT_CONFIG, + set(), + set(), + id="only_negation_matches_nothing", + ), pytest.param( ["**/*.csv"], {"start_date": "2023-06-05T03:54:07.000Z", "streams": []},