devin-ai-integration · Sep 25, 2026
@@ -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 <a href="https://en.wikipedia.org/wiki/Glob_(programming)">here</a>.',
+        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 <a href="https://en.wikipedia.org/wiki/Glob_(programming)">here</a>.',
         order=1,
     )
     legacy_prefix: Optional[str] = Field(

@@ -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:

@@ -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()

@@ -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,
             )
 

@@ -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)

@@ -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 <a href="https://en.wikipedia.org/wiki/Glob_(programming)">here</a>.',
+                                    "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 <a href="https://en.wikipedia.org/wiki/Glob_(programming)">here</a>.',
                                     "type": "array",
                                     "items": {"type": "string"},
                                     "order": 1,

@@ -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
 

@@ -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": []},