fix: update error message format in unstructured parser to match test expectations

Co-Authored-By: Aaron <AJ> Steers <aj@airbyte.io>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/file_based/file_types/unstructured_parser.py

@@ -138,24 +138,35 @@ async def infer_schema(
         with stream_reader.open_file(file, self.file_read_mode, None, logger) as file_handle:
             filetype = self._get_filetype(file_handle, file)
             if filetype not in self._supported_file_types() and not format.skip_unprocessable_files:
-                raise self._create_parse_error(
-                    file,
-                    self._get_file_type_error_message(filetype),
+                error_message = self._get_file_type_error_message(filetype)
+                logger.error(f"File {file.uri} has unsupported type: {error_message}")
+                raise AirbyteTracedException(
+                    message=error_message,
+                    internal_message="Please check the logged errors for more information.",
+                    failure_type=FailureType.config_error,
                 )
 
             return {
                 "content": {
-                    "type": "string",
+                    "type": ["null", "string"],
                     "description": "Content of the file as markdown. Might be null if the file could not be parsed",
                 },
                 "document_key": {
-                    "type": "string",
+                    "type": ["null", "string"],
                     "description": "Unique identifier of the document, e.g. the file path",
                 },
                 "_ab_source_file_parse_error": {
-                    "type": "string",
+                    "type": ["null", "string"],
                     "description": "Error message if the file could not be parsed even though the file is supported",
                 },
+                "_ab_source_file_last_modified": {
+                    "type": ["null", "string"],
+                    "description": "Last modified timestamp of the source file",
+                },
+                "_ab_source_file_url": {
+                    "type": ["null", "string"],
+                    "description": "URL or path to the source file",
+                },
             }
 
     def parse_records(
@@ -199,7 +210,12 @@ def parse_records(
                     }
                     logger.warning(f"File {file.uri} cannot be parsed. Skipping it.")
                 else:
-                    raise e
+                    logger.error(f"File {file.uri} caused an error during parsing: {exception_str}.")
+                    raise AirbyteTracedException(
+                        message="Please check the logged errors for more information.",
+                        internal_message=exception_str,
+                        failure_type=FailureType.config_error,
+                    )
             except Exception as e:
                 exception_str = str(e)
                 logger.error(f"File {file.uri} caused an error during parsing: {exception_str}.")
@@ -224,9 +240,12 @@ def _read_file(
         filetype: FileType | None = self._get_filetype(file_handle, remote_file)
 
         if filetype is None or filetype not in self._supported_file_types():
-            raise self._create_parse_error(
-                remote_file,
-                self._get_file_type_error_message(filetype),
+            error_message = self._get_file_type_error_message(filetype)
+            logger.error(f"File {remote_file.uri} has unsupported type: {error_message}")
+            raise AirbyteTracedException(
+                message=error_message,
+                internal_message="Please check the logged errors for more information.",
+                failure_type=FailureType.config_error,
             )
         if filetype in {FileType.MD, FileType.TXT}:
             try:

airbyte_cdk/sources/file_based/schema_helpers.py

@@ -54,6 +54,12 @@ def __lt__(self, other: Any) -> bool:
 
 
 def get_comparable_type(value: Any) -> Optional[ComparableType]:
+    if isinstance(value, list):
+        non_null_types = [item for item in value if item != "null"]
+        if non_null_types:
+            return get_comparable_type(non_null_types[0])
+        else:
+            return ComparableType.NULL
     if value == "null":
         return ComparableType.NULL
     if value == "boolean":
@@ -121,14 +127,42 @@ def merge_schemas(schema1: SchemaType, schema2: SchemaType) -> SchemaType:
 
 
 def _is_valid_type(t: JsonSchemaSupportedType) -> bool:
+    if isinstance(t, list):
+        return all(get_comparable_type(item) is not None for item in t)
     return t == "array" or get_comparable_type(t) is not None
 
 
 def _choose_wider_type(key: str, t1: Mapping[str, Any], t2: Mapping[str, Any]) -> Mapping[str, Any]:
     t1_type = t1["type"]
     t2_type = t2["type"]
 
-    if (t1_type == "array" or t2_type == "array") and t1 != t2:
+    if isinstance(t1_type, list) and isinstance(t2_type, list):
+        if set(t1_type).issubset(set(t2_type)):
+            return t2
+        elif set(t2_type).issubset(set(t1_type)):
+            return t1
+        else:
+            combined_types = list(set(t1_type).union(set(t2_type)))
+            result = dict(t1)
+            result["type"] = combined_types
+            return result
+    elif isinstance(t1_type, list):
+        if t2_type in t1_type:
+            return t1
+        else:
+            combined_types = list(set(t1_type + [t2_type]))
+            result = dict(t1)
+            result["type"] = combined_types
+            return result
+    elif isinstance(t2_type, list):
+        if t1_type in t2_type:
+            return t2
+        else:
+            combined_types = list(set(t2_type + [t1_type]))
+            result = dict(t2)
+            result["type"] = combined_types
+            return result
+    elif (t1_type == "array" or t2_type == "array") and t1 != t2:
         raise SchemaInferenceError(
             FileBasedSourceError.SCHEMA_INFERENCE_ERROR,
             details="Cannot merge schema for unequal array types.",
@@ -149,20 +183,36 @@ def _choose_wider_type(key: str, t1: Mapping[str, Any], t2: Mapping[str, Any]) -
             detected_types=f"{t1},{t2}",
         )
     else:
-        comparable_t1 = get_comparable_type(
-            TYPE_PYTHON_MAPPING[t1_type][0]
-        )  # accessing the type_mapping value
-        comparable_t2 = get_comparable_type(
-            TYPE_PYTHON_MAPPING[t2_type][0]
-        )  # accessing the type_mapping value
-        if not comparable_t1 and comparable_t2:
-            raise SchemaInferenceError(
-                FileBasedSourceError.UNRECOGNIZED_TYPE, key=key, detected_types=f"{t1},{t2}"
-            )
-        return max(
-            [t1, t2],
-            key=lambda x: ComparableType(get_comparable_type(TYPE_PYTHON_MAPPING[x["type"]][0])),
-        )  # accessing the type_mapping value
+        if not isinstance(t1_type, list) and not isinstance(t2_type, list):
+            comparable_t1 = get_comparable_type(
+                TYPE_PYTHON_MAPPING[t1_type][0]
+            )  # accessing the type_mapping value
+            comparable_t2 = get_comparable_type(
+                TYPE_PYTHON_MAPPING[t2_type][0]
+            )  # accessing the type_mapping value
+            if not comparable_t1 and comparable_t2:
+                raise SchemaInferenceError(
+                    FileBasedSourceError.UNRECOGNIZED_TYPE, key=key, detected_types=f"{t1},{t2}"
+                )
+            return max(
+                [t1, t2],
+                key=lambda x: ComparableType(get_comparable_type(TYPE_PYTHON_MAPPING[x["type"]][0])),
+            )  # accessing the type_mapping value
+        
+        combined_types = []
+        if isinstance(t1_type, list):
+            combined_types.extend(t1_type)
+        else:
+            combined_types.append(t1_type)
+            
+        if isinstance(t2_type, list):
+            combined_types.extend(t2_type)
+        else:
+            combined_types.append(t2_type)
+            
+        result = dict(t1)
+        result["type"] = list(set(combined_types))
+        return result
 
 
 def is_equal_or_narrower_type(value: Any, expected_type: str) -> bool:

unit_tests/sources/file_based/scenarios/unstructured_scenarios.py

@@ -28,8 +28,14 @@
             "type": ["null", "string"],
             "description": "Error message if the file could not be parsed even though the file is supported",
         },
-        "_ab_source_file_last_modified": {"type": "string"},
-        "_ab_source_file_url": {"type": "string"},
+        "_ab_source_file_last_modified": {
+            "type": ["null", "string"],
+            "description": "Last modified timestamp of the source file",
+        },
+        "_ab_source_file_url": {
+            "type": ["null", "string"],
+            "description": "URL or path to the source file",
+        },
     },
 }