update config migration/transformation/validation flow

Author: pnilan

airbyte_cdk/entrypoint.py

@@ -186,13 +186,6 @@ def run(self, parsed_args: argparse.Namespace) -> Iterable[str]:
                 else:
                     raw_config = self.source.read_config(parsed_args.config)
                     config = self.source.configure(raw_config, temp_dir)
-                    mutable_config = dict(config)
-                    config_path = parsed_args.config if parsed_args.config else None
-                    if config_path:
-                        self.source.migrate_config(config_path, mutable_config)
-                    self.source.transform_config(mutable_config)
-                    config = mutable_config
-
                     yield from [
                         self.airbyte_message_to_string(queued_message)
                         for queued_message in self._emit_queued_messages(self.source)

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -10,11 +10,13 @@
 from types import ModuleType
 from typing import Any, Dict, Iterator, List, Mapping, MutableMapping, Optional, Set
 
+import orjson
 import yaml
 from jsonschema.exceptions import ValidationError
 from jsonschema.validators import validate
 from packaging.version import InvalidVersion, Version
 
+from airbyte_cdk.config_observation import create_connector_config_control_message
 from airbyte_cdk.connector_builder.models import (
     LogMessage as ConnectorBuilderLogMessage,
 )
@@ -29,6 +31,7 @@
     ConnectorSpecification,
     FailureType,
 )
+from airbyte_cdk.models.airbyte_protocol_serializers import AirbyteMessageSerializer
 from airbyte_cdk.sources.declarative.checks import COMPONENTS_CHECKER_TYPE_MAPPING
 from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker
 from airbyte_cdk.sources.declarative.declarative_source import DeclarativeSource
@@ -100,6 +103,7 @@ def __init__(
         component_factory: Optional[ModelToComponentFactory] = None,
         migrate_manifest: Optional[bool] = False,
         normalize_manifest: Optional[bool] = False,
+        config_path: Optional[str] = None,
     ) -> None:
         """
         Args:
@@ -109,6 +113,7 @@ def __init__(
             emit_connector_builder_messages: True if messages should be emitted to the connector builder.
             component_factory: optional factory if ModelToComponentFactory's default behavior needs to be tweaked.
             normalize_manifest: Optional flag to indicate if the manifest should be normalized.
+            config_path: Optional path to the config file.
         """
         self.logger = logging.getLogger(f"airbyte.{self.name}")
         self._should_normalize = normalize_manifest
@@ -131,7 +136,6 @@ def __init__(
         self._slice_logger: SliceLogger = (
             AlwaysLogSliceLogger() if emit_connector_builder_messages else DebugSliceLogger()
         )
-        self._config = config or {}
 
         # resolve all components in the manifest
         self._source_config = self._pre_process_manifest(dict(source_config))
@@ -140,11 +144,33 @@ def __init__(
         # apply additional post-processing to the manifest
         self._post_process_manifest()
 
+        self._config: Mapping[str, Any]
         self._spec_component: Optional[Spec] = None
         if spec := self._source_config.get("spec"):
             if "type" not in spec:
                 spec["type"] = "Spec"
             self._spec_component = self._constructor.create_component(SpecModel, spec, dict())
+            mutable_config = dict(config) if config else {}
+
+            if config_path:
+                self._spec_component.migrate_config(mutable_config)
+                try:
+                    if mutable_config != config:
+                        with open(config_path, "w") as f:
+                            json.dump(mutable_config, f)
+                        self.message_repository.emit_message(
+                            create_connector_config_control_message(mutable_config)
+                        )
+                        # We have no mechanism for consuming the queue, so we print the messages to stdout
+                        for message in self.message_repository.consume_queue():
+                            print(orjson.dumps(AirbyteMessageSerializer.dump(message)).decode())
+                except Exception as e:
+                    self.logger.error(f"Error migrating config: {str(e)}")
+                    mutable_config = dict(config) if config else {}
+            self._spec_component.transform_config(mutable_config)
+            self._config = mutable_config
+        else:
+            self._config = config or {}
 
     @property
     def resolved_manifest(self) -> Mapping[str, Any]:
@@ -296,12 +322,6 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
 
         return source_streams
 
-    def migrate_config(self, config_path: Optional[str], config: MutableMapping[str, Any]) -> None:
-        self._spec_component.migrate_config(config_path, config) if self._spec_component else None
-
-    def transform_config(self, config: MutableMapping[str, Any]) -> None:
-        self._spec_component.transform_config(config) if self._spec_component else None
-
     @staticmethod
     def _initialize_cache_for_parent_streams(
         stream_configs: List[Dict[str, Any]],

airbyte_cdk/sources/declarative/spec/spec.py

@@ -69,30 +69,16 @@ def generate_spec(self) -> ConnectorSpecification:
         # We remap these keys to camel case because that's the existing format expected by the rest of the platform
         return ConnectorSpecificationSerializer.load(obj)
 
-    def migrate_config(self, config_path: Optional[str], config: Mapping[str, Any]) -> None:
+    def migrate_config(self, config: MutableMapping[str, Any]) -> None:
         """
-        Apply all specified config transformations to the provided config and save the modified config to the given path and emit a control message.
+        Apply all specified config transformations to the provided config and emit a control message.
 
-        :param config_path: The path to the config file
         :param config: The user-provided config to migrate
         """
 
-        if not config_path:
-            return
-
-        mutable_config = dict(config)
         for migration in self.config_migrations:
             for transformation in migration.transformations:
-                transformation.transform(mutable_config)
-
-        if mutable_config != config:
-            with open(config_path, "w") as f:
-                json.dump(mutable_config, f)
-            self.message_repository.emit_message(
-                create_connector_config_control_message(mutable_config)
-            )
-            for message in self.message_repository.consume_queue():
-                print(orjson.dumps(AirbyteMessageSerializer.dump(message)).decode())
+                transformation.transform(config)
 
     def transform_config(self, config: MutableMapping[str, Any]) -> None:
         """

airbyte_cdk/sources/source.py

@@ -93,15 +93,3 @@ def read_catalog(cls, catalog_path: str) -> ConfiguredAirbyteCatalog:
     def name(self) -> str:
         """Source name"""
         return self.__class__.__name__
-
-    def migrate_config(self, config_path: Optional[str], config: MutableMapping[str, Any]) -> None:
-        """
-        Optional method to migrate config.
-        """
-        pass
-
-    def transform_config(self, config: MutableMapping[str, Any]) -> None:
-        """
-        Optional method to transform config.
-        """
-        pass