From 806cdffdd0157d97381286d5ece4a8b4b09c0c72 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Mon, 20 Oct 2025 16:30:05 +0300 Subject: [PATCH 1/5] handle schema inference error for empty file --- airbyte_cdk/sources/file_based/exceptions.py | 4 +++ .../file_based/file_types/csv_parser.py | 5 ++-- .../stream/default_file_based_stream.py | 25 +++++++++++-------- 3 files changed, 22 insertions(+), 12 deletions(-) diff --git a/airbyte_cdk/sources/file_based/exceptions.py b/airbyte_cdk/sources/file_based/exceptions.py index b0d38947f9..0113c77395 100644 --- a/airbyte_cdk/sources/file_based/exceptions.py +++ b/airbyte_cdk/sources/file_based/exceptions.py @@ -157,3 +157,7 @@ class CustomFileBasedException(AirbyteTracedException): class FileSizeLimitError(CustomFileBasedException): pass + + +class EmptyFileSchemaInferenceError(AirbyteTracedException): + pass \ No newline at end of file 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 e3010690e6..7e4ae91c14 100644 --- a/airbyte_cdk/sources/file_based/file_types/csv_parser.py +++ b/airbyte_cdk/sources/file_based/file_types/csv_parser.py @@ -22,7 +22,8 @@ InferenceType, ) from airbyte_cdk.sources.file_based.config.file_based_stream_config import FileBasedStreamConfig -from airbyte_cdk.sources.file_based.exceptions import FileBasedSourceError, RecordParseError +from airbyte_cdk.sources.file_based.exceptions import FileBasedSourceError, RecordParseError, \ + EmptyFileSchemaInferenceError from airbyte_cdk.sources.file_based.file_based_stream_reader import ( AbstractFileBasedStreamReader, FileReadMode, @@ -203,7 +204,7 @@ async def infer_schema( break if not type_inferrer_by_field: - raise AirbyteTracedException( + 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, 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 0e7121325c..d49f08a778 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 @@ -21,7 +21,7 @@ MissingSchemaError, RecordParseError, SchemaInferenceError, - StopSyncPerValidationPolicy, + StopSyncPerValidationPolicy, EmptyFileSchemaInferenceError, ) from airbyte_cdk.sources.file_based.file_types import FileTransfer from airbyte_cdk.sources.file_based.remote_file import RemoteFile @@ -246,12 +246,12 @@ def get_json_schema(self) -> JsonSchema: exception=AirbyteTracedException(exception=config_exception), failure_type=FailureType.config_error, ) + except EmptyFileSchemaInferenceError as exc: + self._raise_schema_inference_error(exc) except AirbyteTracedException as ate: raise ate except Exception as exc: - raise SchemaInferenceError( - FileBasedSourceError.SCHEMA_INFERENCE_ERROR, stream=self.name - ) from exc + self._raise_schema_inference_error(exc) else: return {"type": "object", "properties": {**extra_fields, **schema["properties"]}} @@ -385,12 +385,17 @@ async def _infer_file_schema(self, file: RemoteFile) -> SchemaType: return await self.get_parser().infer_schema( self.config, file, self.stream_reader, self.logger ) + except EmptyFileSchemaInferenceError as exc: + self._raise_schema_inference_error(exc, file) except AirbyteTracedException as ate: raise ate except Exception as exc: - raise SchemaInferenceError( - FileBasedSourceError.SCHEMA_INFERENCE_ERROR, - file=file.uri, - format=str(self.config.format), - stream=self.name, - ) from exc + self._raise_schema_inference_error(exc, file) + + def _raise_schema_inference_error(self, exc: Exception, file: Optional[RemoteFile] = None) -> None: + raise SchemaInferenceError( + FileBasedSourceError.SCHEMA_INFERENCE_ERROR, + file=file.uri if file else None, + format=str(self.config.format) if self.config.format else None, + stream=self.name, + ) from exc \ No newline at end of file From e00e10cf14baa93a61614a2079e4e55254eb96f2 Mon Sep 17 00:00:00 2001 From: octavia-squidington-iii Date: Mon, 20 Oct 2025 14:23:32 +0000 Subject: [PATCH 2/5] Auto-fix lint and format issues --- airbyte_cdk/sources/file_based/exceptions.py | 2 +- airbyte_cdk/sources/file_based/file_types/csv_parser.py | 7 +++++-- .../file_based/stream/default_file_based_stream.py | 9 ++++++--- 3 files changed, 12 insertions(+), 6 deletions(-) diff --git a/airbyte_cdk/sources/file_based/exceptions.py b/airbyte_cdk/sources/file_based/exceptions.py index 0113c77395..75f7d3f837 100644 --- a/airbyte_cdk/sources/file_based/exceptions.py +++ b/airbyte_cdk/sources/file_based/exceptions.py @@ -160,4 +160,4 @@ class FileSizeLimitError(CustomFileBasedException): class EmptyFileSchemaInferenceError(AirbyteTracedException): - pass \ No newline at end of file + pass 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 7e4ae91c14..edab346fe0 100644 --- a/airbyte_cdk/sources/file_based/file_types/csv_parser.py +++ b/airbyte_cdk/sources/file_based/file_types/csv_parser.py @@ -22,8 +22,11 @@ InferenceType, ) from airbyte_cdk.sources.file_based.config.file_based_stream_config import FileBasedStreamConfig -from airbyte_cdk.sources.file_based.exceptions import FileBasedSourceError, RecordParseError, \ - EmptyFileSchemaInferenceError +from airbyte_cdk.sources.file_based.exceptions import ( + EmptyFileSchemaInferenceError, + FileBasedSourceError, + RecordParseError, +) from airbyte_cdk.sources.file_based.file_based_stream_reader import ( AbstractFileBasedStreamReader, FileReadMode, 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 d49f08a778..2c7ecd391e 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 @@ -16,12 +16,13 @@ from airbyte_cdk.sources.file_based.config.file_based_stream_config import PrimaryKeyType from airbyte_cdk.sources.file_based.exceptions import ( DuplicatedFilesError, + EmptyFileSchemaInferenceError, FileBasedSourceError, InvalidSchemaError, MissingSchemaError, RecordParseError, SchemaInferenceError, - StopSyncPerValidationPolicy, EmptyFileSchemaInferenceError, + StopSyncPerValidationPolicy, ) from airbyte_cdk.sources.file_based.file_types import FileTransfer from airbyte_cdk.sources.file_based.remote_file import RemoteFile @@ -392,10 +393,12 @@ async def _infer_file_schema(self, file: RemoteFile) -> SchemaType: except Exception as exc: self._raise_schema_inference_error(exc, file) - def _raise_schema_inference_error(self, exc: Exception, file: Optional[RemoteFile] = None) -> None: + def _raise_schema_inference_error( + self, exc: Exception, file: Optional[RemoteFile] = None + ) -> None: raise SchemaInferenceError( FileBasedSourceError.SCHEMA_INFERENCE_ERROR, file=file.uri if file else None, format=str(self.config.format) if self.config.format else None, stream=self.name, - ) from exc \ No newline at end of file + ) from exc From 699b40b91671331d8d1e5f2ba00cde0bac54f835 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Mon, 20 Oct 2025 17:34:33 +0300 Subject: [PATCH 3/5] fix mypy --- .../sources/file_based/stream/default_file_based_stream.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 2c7ecd391e..c19bdc075e 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 @@ -231,7 +231,7 @@ def cursor_field(self) -> Union[str, List[str]]: return self.ab_last_mod_col @cache - def get_json_schema(self) -> JsonSchema: + def get_json_schema(self) -> JsonSchema: # type: ignore if self.use_file_transfer: return file_transfer_schema extra_fields = { @@ -381,7 +381,7 @@ async def _infer_schema(self, files: List[RemoteFile]) -> Mapping[str, Any]: return base_schema - async def _infer_file_schema(self, file: RemoteFile) -> SchemaType: + async def _infer_file_schema(self, file: RemoteFile) -> SchemaType: # type: ignore try: return await self.get_parser().infer_schema( self.config, file, self.stream_reader, self.logger From 5fae6e354eed6c305efbf22c16db282126b6fff9 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Mon, 20 Oct 2025 17:41:03 +0300 Subject: [PATCH 4/5] updated typing --- .../stream/default_file_based_stream.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) 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 c19bdc075e..3b076fe0b3 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 @@ -9,7 +9,19 @@ from copy import deepcopy from functools import cache from os import path -from typing import Any, Dict, Iterable, List, Mapping, MutableMapping, Optional, Set, Tuple, Union +from typing import ( + Any, + Dict, + Iterable, + List, + Mapping, + MutableMapping, + Optional, + Set, + Tuple, + Union, + NoReturn, +) from airbyte_cdk.models import AirbyteLogMessage, AirbyteMessage, AirbyteStream, FailureType, Level from airbyte_cdk.models import Type as MessageType @@ -395,7 +407,7 @@ async def _infer_file_schema(self, file: RemoteFile) -> SchemaType: # type: ign def _raise_schema_inference_error( self, exc: Exception, file: Optional[RemoteFile] = None - ) -> None: + ) -> NoReturn: raise SchemaInferenceError( FileBasedSourceError.SCHEMA_INFERENCE_ERROR, file=file.uri if file else None, From 8c206510a427c835249714c0e7d05f2d2f64719e Mon Sep 17 00:00:00 2001 From: octavia-squidington-iii Date: Mon, 20 Oct 2025 14:48:28 +0000 Subject: [PATCH 5/5] Auto-fix lint and format issues --- .../sources/file_based/stream/default_file_based_stream.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 3b076fe0b3..3053a74d2a 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 @@ -16,11 +16,11 @@ List, Mapping, MutableMapping, + NoReturn, Optional, Set, Tuple, Union, - NoReturn, ) from airbyte_cdk.models import AirbyteLogMessage, AirbyteMessage, AirbyteStream, FailureType, Level