fix(python-cdk): add user friendly message for encoding errors

airbyte-cdk/python/airbyte_cdk/sources/file_based/availability_strategy/default_file_based_availability_strategy.py

@@ -8,7 +8,7 @@
 
 from airbyte_cdk.sources import Source
 from airbyte_cdk.sources.file_based.availability_strategy import AbstractFileBasedAvailabilityStrategy
-from airbyte_cdk.sources.file_based.exceptions import CheckAvailabilityError, CustomFileBasedException, FileBasedSourceError
+from airbyte_cdk.sources.file_based.exceptions import CheckAvailabilityError, CustomFileBasedException, EncodingError, FileBasedSourceError
 from airbyte_cdk.sources.file_based.file_based_stream_reader import AbstractFileBasedStreamReader
 from airbyte_cdk.sources.file_based.remote_file import RemoteFile
 from airbyte_cdk.sources.file_based.schema_helpers import conforms_to_schema
@@ -66,6 +66,8 @@ def check_availability_and_parsability(
                 # If the parser is set to not check parsability, we still want to check that we can open the file.
                 handle = stream.stream_reader.open_file(file, parser.file_read_mode, None, logger)
                 handle.close()
+        except EncodingError as encoding_error:
+            raise encoding_error
         except CheckAvailabilityError:
             return False, "".join(traceback.format_exc())
 
@@ -98,6 +100,8 @@ def _check_parse_record(self, stream: "AbstractFileBasedStream", file: RemoteFil
             # consider the connection check successful even though it means
             # we skip the schema validation check.
             return
+        except EncodingError as encoding_error:
+            raise encoding_error
         except Exception as exc:
             raise CheckAvailabilityError(FileBasedSourceError.ERROR_READING_FILE, stream=stream.name, file=file.uri) from exc
 

airbyte-cdk/python/airbyte_cdk/sources/file_based/exceptions.py

@@ -14,6 +14,9 @@ class FileBasedSourceError(Enum):
     GLOB_PARSE_ERROR = (
         "Error parsing glob pattern. Please refer to the glob pattern rules at https://facelessuser.github.io/wcmatch/glob/#split."
     )
+    ENCODING_ERROR = (
+        "Error decoding file. Please check that the configured encoding matches the encoding of the file. Default encoding is UTF-8."
+    )
     ERROR_CASTING_VALUE = "Could not cast the value to the expected type."
     ERROR_CASTING_VALUE_UNRECOGNIZED_TYPE = "Could not cast the value to the expected type because the type is not recognized. Valid types are null, array, boolean, integer, number, object, and string."
     ERROR_DECODING_VALUE = "Expected a JSON-decodeable value but could not decode record."
@@ -76,6 +79,10 @@ class ConfigValidationError(BaseFileBasedSourceError):
     pass
 
 
+class EncodingError(BaseFileBasedSourceError):
+    pass
+
+
 class InvalidSchemaError(BaseFileBasedSourceError):
     pass
 

airbyte-cdk/python/airbyte_cdk/sources/file_based/file_based_source.py

@@ -26,7 +26,7 @@
 from airbyte_cdk.sources.file_based.config.abstract_file_based_spec import AbstractFileBasedSpec
 from airbyte_cdk.sources.file_based.config.file_based_stream_config import FileBasedStreamConfig, ValidationPolicy
 from airbyte_cdk.sources.file_based.discovery_policy import AbstractDiscoveryPolicy, DefaultDiscoveryPolicy
-from airbyte_cdk.sources.file_based.exceptions import ConfigValidationError, FileBasedErrorsCollector, FileBasedSourceError
+from airbyte_cdk.sources.file_based.exceptions import ConfigValidationError, EncodingError, FileBasedErrorsCollector, FileBasedSourceError
 from airbyte_cdk.sources.file_based.file_based_stream_reader import AbstractFileBasedStreamReader
 from airbyte_cdk.sources.file_based.file_types import default_parsers
 from airbyte_cdk.sources.file_based.file_types.file_type_parser import FileTypeParser
@@ -130,6 +130,13 @@ def check_connection(self, logger: logging.Logger, config: Mapping[str, Any]) ->
                     stream_is_available,
                     reason,
                 ) = stream.availability_strategy.check_availability_and_parsability(stream, logger, self)
+            except EncodingError as encoding_error:
+                raise AirbyteTracedException(
+                    internal_message="File encoding does not match configuration.",
+                    message=FileBasedSourceError.ENCODING_ERROR.value,
+                    exception=encoding_error,
+                    failure_type=FailureType.config_error,
+                )
             except Exception:
                 errors.append(f"Unable to connect to stream {stream.name} - {''.join(traceback.format_exc())}")
             else:

airbyte-cdk/python/airbyte_cdk/sources/file_based/file_types/csv_parser.py

@@ -15,7 +15,7 @@
 from airbyte_cdk.models import FailureType
 from airbyte_cdk.sources.file_based.config.csv_format import CsvFormat, CsvHeaderAutogenerated, CsvHeaderUserProvided, 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 EncodingError, FileBasedSourceError, RecordParseError
 from airbyte_cdk.sources.file_based.file_based_stream_reader import AbstractFileBasedStreamReader, FileReadMode
 from airbyte_cdk.sources.file_based.file_types.file_type_parser import FileTypeParser
 from airbyte_cdk.sources.file_based.remote_file import RemoteFile
@@ -52,7 +52,14 @@ def read_data(
             quoting=csv.QUOTE_MINIMAL,
         )
         with stream_reader.open_file(file, file_read_mode, config_format.encoding, logger) as fp:
-            headers = self._get_headers(fp, config_format, dialect_name)
+            try:
+                headers = self._get_headers(fp, config_format, dialect_name)
+            except UnicodeError:
+                file_name = file.uri
+                # Reduce huge file names in error messages
+                if "?" in file.uri:
+                    file_name = file.uri[: file.uri.find("?")]
+                raise EncodingError(FileBasedSourceError.ENCODING_ERROR, filename=file_name) from None
 
             rows_to_skip = (
                 config_format.skip_rows_before_header
@@ -274,7 +281,7 @@ def _pre_propcess_property_types(property_types: Dict[str, Any]) -> Mapping[str,
 
     @staticmethod
     def _cast_types(
-        row: Dict[str, str], deduped_property_types: Dict[str, str], config_format: CsvFormat, logger: logging.Logger
+        row: Dict[str, str], deduped_property_types: Mapping[str, str], config_format: CsvFormat, logger: logging.Logger
     ) -> Dict[str, Any]:
         """
         Casts the values in the input 'row' dictionary according to the types defined in the JSON schema.

airbyte-cdk/python/airbyte_cdk/sources/file_based/stream/default_file_based_stream.py

@@ -13,6 +13,7 @@
 from airbyte_cdk.models import Type as MessageType
 from airbyte_cdk.sources.file_based.config.file_based_stream_config import PrimaryKeyType
 from airbyte_cdk.sources.file_based.exceptions import (
+    EncodingError,
     FileBasedSourceError,
     InvalidSchemaError,
     MissingSchemaError,
@@ -54,7 +55,7 @@ def state(self, value: MutableMapping[str, Any]) -> None:
         """State setter, accept state serialized by state getter."""
         self._cursor.set_initial_state(value)
 
-    @property
+    @property  # type: ignore # no suitable parent cursor type
     def cursor(self) -> Optional[AbstractFileBasedCursor]:
         return self._cursor
 
@@ -172,13 +173,19 @@ def get_json_schema(self) -> JsonSchema:
         try:
             schema = self._get_raw_json_schema()
         except InvalidSchemaError as config_exception:
-            self.logger.exception(FileBasedSourceError.SCHEMA_INFERENCE_ERROR.value, exc_info=config_exception)
             raise AirbyteTracedException(
                 internal_message="Please check the logged errors for more information.",
                 message=FileBasedSourceError.SCHEMA_INFERENCE_ERROR.value,
                 exception=AirbyteTracedException(exception=config_exception),
                 failure_type=FailureType.config_error,
             )
+        except EncodingError as encoding_error:
+            raise AirbyteTracedException(
+                internal_message="File encoding does not match configuration.",
+                message=FileBasedSourceError.ENCODING_ERROR.value,
+                exception=encoding_error,
+                failure_type=FailureType.config_error,
+            )
         except Exception as exc:
             raise SchemaInferenceError(FileBasedSourceError.SCHEMA_INFERENCE_ERROR, stream=self.name) from exc
         else:
@@ -279,6 +286,8 @@ async def _infer_schema(self, files: List[RemoteFile]) -> Mapping[str, Any]:
             for task in done:
                 try:
                     base_schema = merge_schemas(base_schema, task.result())
+                except EncodingError as encoding_error:
+                    raise encoding_error
                 except Exception as exc:
                     self.logger.error(f"An error occurred inferring the schema. \n {traceback.format_exc()}", exc_info=exc)
 
@@ -287,6 +296,8 @@ async def _infer_schema(self, files: List[RemoteFile]) -> Mapping[str, Any]:
     async def _infer_file_schema(self, file: RemoteFile) -> SchemaType:
         try:
             return await self.get_parser().infer_schema(self.config, file, self.stream_reader, self.logger)
+        except EncodingError as encoding_error:
+            raise encoding_error
         except Exception as exc:
             raise SchemaInferenceError(
                 FileBasedSourceError.SCHEMA_INFERENCE_ERROR,