add ConfigMigration class and new spec tests

Author: pnilan

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -3810,26 +3810,13 @@ definitions:
         properties:
           config_migrations:
             title: Config Migrations
-            description: The config will be migrated according to these transformations and updated within the platform for subsequent syncs.
-            type: object
-            required:
-              - description
-              - transformations
-            properties:
-              description:
-                type: string
-                description: The description/purpose of the config migration.
-              transformations:
-                type: array
-                items:
-                  anyOf:
-                    - "$ref": "#/definitions/ConfigRemapField"
-                    - "$ref": "#/definitions/ConfigAddFields"
-                    - "$ref": "#/definitions/ConfigRemoveFields"
-                default: []
+            description: The discrete migrations that will be applied on the incoming config. Each migration will be applied in the order they are defined.
+            type: array
+            items:
+              "$ref": "#/definitions/ConfigMigration"
           transformations:
             title: Transformations
-            description: The list of transformations that will be applied on the incoming config at the start of a sync.
+            description: The list of transformations that will be applied on the incoming config at the start of each sync. The transformations will be applied in the order they are defined.
             type: array
             items:
               anyOf:
@@ -3838,12 +3825,32 @@ definitions:
                 - "$ref": "#/definitions/ConfigRemoveFields"
           validations:
             title: Validations
-            description: The list of validations that will be performed on the incoming config before starting a sync
+            description: The list of validations that will be performed on the incoming config at the start of each sync.
             type: array
             items:
               anyOf:
                 - "$ref": "#/definitions/DpathValidator"
                 - "$ref": "#/definitions/PredicateValidator"
+  ConfigMigration:
+    title: Config Migration
+    description: A config migration that will be applied on the incoming config at the start of a sync.
+    type: object
+    required:
+      - transformations
+    properties:
+      description:
+        type: string
+        description: The description/purpose of the config migration.
+      transformations:
+        title: Transformations
+        description: The list of transformations that will attempt to be applied on an incoming unmigrated config. The transformations will be applied in the order they are defined.
+        type: array
+        items:
+          anyOf:
+            - "$ref": "#/definitions/ConfigRemapField"
+            - "$ref": "#/definitions/ConfigAddFields"
+            - "$ref": "#/definitions/ConfigRemoveFields"
+        default: []
   SubstreamPartitionRouter:
     title: Substream Partition Router
     description: Partition router that is used to retrieve records that have been partitioned according to records from the specified parent streams. An example of a parent stream is automobile brands and the substream would be the various car models associated with each branch.
@@ -4275,14 +4282,14 @@ definitions:
     type: object
     required:
       - type
-      - schema
+      - base_schema
     properties:
       type:
         type: string
         enum: [ValidateAdheresToSchema]
-      schema:
-        title: JSON Schema
-        description: The JSON schema used for validation.
+      base_schema:
+        title: Base JSON Schema
+        description: The base JSON schema against which the user-provided schema will be validated.
         type:
           - string
           - object
@@ -4406,9 +4413,10 @@ definitions:
         description: A list of field pointers to be removed from the config.
         type: array
         items:
-          type: string
+          items:
+            type: string
         examples:
-          - ["marketplace"]
+          - ["tags"]
           - [["content", "html"], ["content", "plain_text"]]
       condition:
         description: Fields will be removed if expression is evaluated to True.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1531,10 +1531,9 @@ class Config:
 
 class ValidateAdheresToSchema(BaseModel):
     type: Literal["ValidateAdheresToSchema"]
-    schema_: Union[str, Dict[str, Any]] = Field(
+    base_schema: Union[str, Dict[str, Any]] = Field(
         ...,
-        alias="schema",
-        description="The JSON schema used for validation.",
+        description="The base JSON schema against which the user-provided schema will be validated.",
         examples=[
             "{{ config['report_validation_schema'] }}",
             '\'{\n  "$schema": "http://json-schema.org/draft-07/schema#",\n  "title": "Person",\n  "type": "object",\n  "properties": {\n    "name": {\n      "type": "string",\n      "description": "The person\'s name"\n    },\n    "age": {\n      "type": "integer",\n      "minimum": 0,\n      "description": "The person\'s age"\n    }\n  },\n  "required": ["name", "age"]\n}\'\n',
@@ -1553,7 +1552,7 @@ class ValidateAdheresToSchema(BaseModel):
                 "required": ["name", "age"],
             },
         ],
-        title="JSON Schema",
+        title="Base JSON Schema",
     )
 
 
@@ -1583,10 +1582,10 @@ class ConfigRemapField(BaseModel):
 
 class ConfigRemoveFields(BaseModel):
     type: Literal["ConfigRemoveFields"]
-    field_pointers: List[str] = Field(
+    field_pointers: List[List[str]] = Field(
         ...,
         description="A list of field pointers to be removed from the config.",
-        examples=[["marketplace"], [["content", "html"], ["content", "plain_text"]]],
+        examples=[["tags"], [["content", "html"], ["content", "plain_text"]]],
         title="Field Pointers",
     )
     condition: Optional[str] = Field(
@@ -2104,29 +2103,35 @@ class Config:
     )
 
 
-class ConfigMigrations(BaseModel):
-    description: str = Field(
-        ..., description="The description/purpose of the config migration."
+class ConfigMigration(BaseModel):
+    description: Optional[str] = Field(
+        None, description="The description/purpose of the config migration."
+    )
+    transformations: List[
+        Union[ConfigRemapField, ConfigAddFields, ConfigRemoveFields]
+    ] = Field(
+        ...,
+        description="The list of transformations that will attempt to be applied on an incoming unmigrated config. The transformations will be applied in the order they are defined.",
+        title="Transformations",
     )
-    transformations: List[Union[ConfigRemapField, ConfigAddFields, ConfigRemoveFields]]
 
 
 class ConfigNormalizationRules(BaseModel):
-    config_migrations: Optional[ConfigMigrations] = Field(
+    config_migrations: Optional[List[ConfigMigration]] = Field(
         None,
-        description="The config will be migrated according to these transformations and updated within the platform for subsequent syncs.",
+        description="The discrete migrations that will be applied on the incoming config. Each migration will be applied in the order they are defined.",
         title="Config Migrations",
     )
     transformations: Optional[
         List[Union[ConfigRemapField, ConfigAddFields, ConfigRemoveFields]]
     ] = Field(
         None,
-        description="The list of transformations that will be applied on the incoming config at the start of a sync.",
+        description="The list of transformations that will be applied on the incoming config at the start of each sync. The transformations will be applied in the order they are defined.",
         title="Transformations",
     )
     validations: Optional[List[Union[DpathValidator, PredicateValidator]]] = Field(
         None,
-        description="The list of validations that will be performed on the incoming config before starting a sync",
+        description="The list of validations that will be performed on the incoming config at the start of each sync.",
         title="Validations",
     )
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -19,6 +19,7 @@
     Optional,
     Type,
     Union,
+    cast,
     get_args,
     get_origin,
     get_type_hints,
@@ -154,9 +155,21 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ConcurrencyLevel as ConcurrencyLevelModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    ConfigAddFields as ConfigAddFieldsModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ConfigComponentsResolver as ConfigComponentsResolverModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    ConfigMigration as ConfigMigrationModel,
+)
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    ConfigRemapField as ConfigRemapFieldModel,
+)
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    ConfigRemoveFields as ConfigRemoveFieldsModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ConstantBackoffStrategy as ConstantBackoffStrategyModel,
 )
@@ -226,6 +239,9 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     DpathFlattenFields as DpathFlattenFieldsModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    DpathValidator as DpathValidatorModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     DynamicSchemaLoader as DynamicSchemaLoaderModel,
 )
@@ -337,6 +353,9 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     ParentStreamConfig as ParentStreamConfigModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    PredicateValidator as PredicateValidatorModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     PropertiesFromEndpoint as PropertiesFromEndpointModel,
 )
@@ -401,6 +420,9 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     UnlimitedCallRatePolicy as UnlimitedCallRatePolicyModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    ValidateAdheresToSchema as ValidateAdheresToSchemaModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import ValueType
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     WaitTimeFromHeader as WaitTimeFromHeaderModel,
@@ -506,14 +528,19 @@
     TypesMap,
 )
 from airbyte_cdk.sources.declarative.schema.composite_schema_loader import CompositeSchemaLoader
-from airbyte_cdk.sources.declarative.spec import Spec
+from airbyte_cdk.sources.declarative.spec import ConfigMigration, Spec
 from airbyte_cdk.sources.declarative.stream_slicers import StreamSlicer
 from airbyte_cdk.sources.declarative.transformations import (
     AddFields,
     RecordTransformation,
     RemoveFields,
 )
 from airbyte_cdk.sources.declarative.transformations.add_fields import AddedFieldDefinition
+from airbyte_cdk.sources.declarative.transformations.config_transformations import (
+    ConfigAddFields,
+    ConfigRemapField,
+    ConfigRemoveFields,
+)
 from airbyte_cdk.sources.declarative.transformations.dpath_flatten_fields import (
     DpathFlattenFields,
     KeyTransformation,
@@ -530,6 +557,11 @@
 from airbyte_cdk.sources.declarative.transformations.keys_to_snake_transformation import (
     KeysToSnakeCaseTransformation,
 )
+from airbyte_cdk.sources.declarative.validators import (
+    DpathValidator,
+    PredicateValidator,
+    ValidateAdheresToSchema,
+)
 from airbyte_cdk.sources.message import (
     InMemoryMessageRepository,
     LogAppenderMessageRepositoryDecorator,
@@ -617,6 +649,10 @@ def _init_mappings(self) -> None:
             CheckDynamicStreamModel: self.create_check_dynamic_stream,
             CompositeErrorHandlerModel: self.create_composite_error_handler,
             ConcurrencyLevelModel: self.create_concurrency_level,
+            ConfigMigrationModel: self.create_config_migration,
+            ConfigAddFieldsModel: self.create_config_add_fields,
+            ConfigRemapFieldModel: self.create_config_remap_field,
+            ConfigRemoveFieldsModel: self.create_config_remove_fields,
             ConstantBackoffStrategyModel: self.create_constant_backoff_strategy,
             CsvDecoderModel: self.create_csv_decoder,
             CursorPaginationModel: self.create_cursor_pagination,
@@ -640,6 +676,7 @@ def _init_mappings(self) -> None:
             DefaultErrorHandlerModel: self.create_default_error_handler,
             DefaultPaginatorModel: self.create_default_paginator,
             DpathExtractorModel: self.create_dpath_extractor,
+            DpathValidatorModel: self.create_dpath_validator,
             ResponseToFileExtractorModel: self.create_response_to_file_extractor,
             ExponentialBackoffStrategyModel: self.create_exponential_backoff_strategy,
             SessionTokenAuthenticatorModel: self.create_session_token_authenticator,
@@ -673,6 +710,7 @@ def _init_mappings(self) -> None:
             OffsetIncrementModel: self.create_offset_increment,
             PageIncrementModel: self.create_page_increment,
             ParentStreamConfigModel: self.create_parent_stream_config,
+            PredicateValidatorModel: self.create_predicate_validator,
             PropertiesFromEndpointModel: self.create_properties_from_endpoint,
             PropertyChunkingModel: self.create_property_chunking,
             QueryPropertiesModel: self.create_query_properties,
@@ -687,6 +725,7 @@ def _init_mappings(self) -> None:
             StateDelegatingStreamModel: self.create_state_delegating_stream,
             SpecModel: self.create_spec,
             SubstreamPartitionRouterModel: self.create_substream_partition_router,
+            ValidateAdheresToSchemaModel: self.create_validate_adheres_to_schema,
             WaitTimeFromHeaderModel: self.create_wait_time_from_header,
             WaitUntilTimeFromHeaderModel: self.create_wait_until_time_from_header,
             AsyncRetrieverModel: self.create_async_retriever,
@@ -779,6 +818,65 @@ def _collect_model_deprecations(self, model: BaseModelWithDeprecations) -> None:
                 if log not in self._collected_deprecation_logs:
                     self._collected_deprecation_logs.append(log)
 
+
+    def create_config_migration(self, model: ConfigMigrationModel, config: Config) -> ConfigMigration:
+        transformations = []
+        for transformation in model.transformations:
+            transformations.append(self._create_component_from_model(transformation, config))
+
+        return ConfigMigration(
+            description=model.description,
+            transformations=transformations,
+        )
+
+    def create_config_add_fields(self, model: ConfigAddFieldsModel, config: Config) -> ConfigAddFields:
+        fields = [self._create_component_from_model(field, config) for field in model.fields]
+        return ConfigAddFields(
+            fields=fields,
+            condition=model.condition or "",
+        )
+
+    @staticmethod
+    def create_config_remove_fields(model: ConfigRemoveFieldsModel) -> ConfigRemoveFields:
+        return ConfigRemoveFields(
+            field_pointers=model.field_pointers,
+            condition=model.condition or "",
+        )
+
+    @staticmethod
+    def create_config_remap_field(model: ConfigRemapFieldModel, config: Config) -> ConfigRemapField:
+        mapping = cast(Mapping[str, Any], model.map)
+        return ConfigRemapField(
+            map=mapping,
+            field_path=model.field_path,
+            config=config,
+        )
+
+
+    def create_dpath_validator(self, model: DpathValidatorModel, config: Config) -> DpathValidator:
+
+        strategy = self._create_component_from_model(model.validation_strategy, config)
+
+        return DpathValidator(
+            field_path=model.field_path,
+            strategy=strategy,
+        )
+
+    def create_predicate_validator(self, model: PredicateValidatorModel, config: Config) -> PredicateValidator:
+        strategy = self._create_component_from_model(model.validation_strategy, config)
+
+        return PredicateValidator(
+            value=model.value,
+            strategy=strategy,
+        )
+
+    @staticmethod
+    def create_validate_adheres_to_schema(model: ValidateAdheresToSchemaModel) -> ValidateAdheresToSchema:
+        base_schema = cast(Mapping[str, Any], model.base_schema)
+        return ValidateAdheresToSchema(
+            schema=base_schema,
+        )
+
     @staticmethod
     def create_added_field_definition(
         model: AddedFieldDefinitionModel, config: Config, **kwargs: Any

airbyte_cdk/sources/declarative/spec/__init__.py

@@ -2,6 +2,6 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-from airbyte_cdk.sources.declarative.spec.spec import Spec
+from airbyte_cdk.sources.declarative.spec.spec import ConfigMigration, Spec
 
-__all__ = ["Spec"]
+__all__ = ["Spec", "ConfigMigration"]