add schema discovery options validation

Author: darynaishchenko

airbyte_cdk/sources/file_based/config/file_based_stream_config.py

@@ -5,7 +5,7 @@
 from enum import Enum
 from typing import Any, List, Mapping, Optional, Union
 
-from pydantic.v1 import BaseModel, Field, validator
+from pydantic.v1 import BaseModel, Field, validator, root_validator
 
 from airbyte_cdk.sources.file_based.config.avro_format import AvroFormat
 from airbyte_cdk.sources.file_based.config.csv_format import CsvFormat
@@ -75,7 +75,7 @@ class FileBasedStreamConfig(BaseModel):
         gt=0,
     )
     use_first_found_file_for_schema_discovery: bool = Field(
-        title="Use first found file for schema discovery",
+        title="Use First Found File For Schema Discover",
         description="When enabled, the source will use the first found file for schema discovery. Helps to avoid long discovery step",
         default=False,
     )
@@ -89,6 +89,35 @@ def validate_input_schema(cls, v: Optional[str]) -> Optional[str]:
                 raise ConfigValidationError(FileBasedSourceError.ERROR_PARSING_USER_PROVIDED_SCHEMA)
         return None
 
+    @root_validator
+    def validate_discovery_related_fields(cls, values):
+        """
+        Please update this validation when new related to schema discovery field is added.
+        Validates schema discovery options compatability.
+        Note, that initially the recent_n_files_to_read_for_schema_discovery was added without a validation if schemaless or input_schema were provided.
+        So this method doesn't check it to do not break already created connections.
+        If recent_n_files_to_read_for_schema_discovery and schemaless or recent_n_files_to_read_for_schema_discovery and input_schema were provided,
+        recent_n_files_to_read_for_schema_discovery will be ignored and second option will be used by default.
+        """
+        input_schema = values["input_schema"] is not None
+        schemaless = values["schemaless"]
+        recent_n_files_to_read_for_schema_discovery = (
+            values["recent_n_files_to_read_for_schema_discovery"] is not None
+        )
+        use_first_found_file_for_schema_discovery = values[
+            "use_first_found_file_for_schema_discovery"
+        ]
+
+        if (
+            recent_n_files_to_read_for_schema_discovery
+            and use_first_found_file_for_schema_discovery
+        ) or [schemaless, input_schema, use_first_found_file_for_schema_discovery].count(True) > 1:
+            raise ConfigValidationError(
+                FileBasedSourceError.ERROR_VALIDATION_STREAM_DISCOVERY_OPTIONS
+            )
+
+        return values
+
     def get_input_schema(self) -> Optional[Mapping[str, Any]]:
         """
         User defined input_schema is defined as a string in the config. This method takes the string representation

airbyte_cdk/sources/file_based/exceptions.py

@@ -23,6 +23,7 @@ class FileBasedSourceError(Enum):
         "The provided schema could not be transformed into valid JSON Schema."
     )
     ERROR_VALIDATING_RECORD = "One or more records do not pass the schema validation policy. Please modify your input schema, or select a more lenient validation policy."
+    ERROR_VALIDATION_STREAM_DISCOVERY_OPTIONS = "Only one of options 'Schemaless', 'Input Schema', 'Files To Read For Schema Discover' or 'Use First Found File For Schema Discover' can be provided at the same time."
     ERROR_PARSING_RECORD_MISMATCHED_COLUMNS = "A header field has resolved to `None`. This indicates that the CSV has more rows than the number of header fields. If you input your schema or headers, please verify that the number of columns corresponds to the number of columns in your CSV's rows."
     ERROR_PARSING_RECORD_MISMATCHED_ROWS = "A row's value has resolved to `None`. This indicates that the CSV has more columns in the header field than the number of columns in the row(s). If you input your schema or headers, please verify that the number of columns corresponds to the number of columns in your CSV's rows."
     STOP_SYNC_PER_SCHEMA_VALIDATION_POLICY = "Stopping sync in accordance with the configured validation policy. Records in file did not conform to the schema."

airbyte_cdk/sources/file_based/file_based_source.py

@@ -156,13 +156,20 @@ def check_connection(
         """
         try:
             streams = self.streams(config)
-        except Exception as config_exception:
+        except ConfigValidationError as config_exception:
             raise AirbyteTracedException(
                 internal_message="Please check the logged errors for more information.",
-                message=FileBasedSourceError.CONFIG_VALIDATION_ERROR.value,
+                message=str(config_exception),
                 exception=AirbyteTracedException(exception=config_exception),
                 failure_type=FailureType.config_error,
             )
+        except Exception as exp:
+            raise AirbyteTracedException(
+                internal_message="Please check the logged errors for more information.",
+                message=FileBasedSourceError.CONFIG_VALIDATION_ERROR.value,
+                exception=AirbyteTracedException(exception=exp),
+                failure_type=FailureType.config_error,
+            )
         if len(streams) == 0:
             return (
                 False,
@@ -250,7 +257,6 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
                     if (state_manager and catalog_stream)
                     else None
                 )
-                self._validate_input_schema(stream_config)
 
                 sync_mode = self._get_sync_mode_from_catalog(stream_config.name)
 
@@ -457,10 +463,3 @@ def _validate_and_get_validation_policy(
                 model=FileBasedStreamConfig,
             )
         return self.validation_policies[stream_config.validation_policy]
-
-    def _validate_input_schema(self, stream_config: FileBasedStreamConfig) -> None:
-        if stream_config.schemaless and stream_config.input_schema:
-            raise ValidationError(
-                "`input_schema` and `schemaless` options cannot both be set",
-                model=FileBasedStreamConfig,
-            )