update per comments

Author: pnilan

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -8,7 +8,7 @@
 from copy import deepcopy
 from importlib import metadata
 from types import ModuleType
-from typing import Any, Dict, Iterator, List, Mapping, Optional, Set
+from typing import Any, Dict, Iterator, List, Mapping, MutableMapping, Optional, Set
 
 import orjson
 import yaml
@@ -145,33 +145,13 @@ def __init__(
         self._post_process_manifest()
 
         self._config: Mapping[str, Any]
-        self._spec_component: Optional[Spec] = None
-        spec = self._source_config.get("spec")
-        if 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 {}
+        self._spec_component: Spec
+        spec: Mapping[str, Any] = self._source_config["spec"]
+        self._spec_component = self._constructor.create_component(SpecModel, spec, dict())
+        mutable_config = dict(config) if config else {}
+        self._migrate_config(config_path, mutable_config, config)
+        self._spec_component.transform_config(mutable_config)
+        self._config = mutable_config
 
     @property
     def resolved_manifest(self) -> Mapping[str, Any]:
@@ -233,6 +213,25 @@ def _normalize_manifest(self) -> None:
             normalizer = ManifestNormalizer(self._source_config, self._declarative_component_schema)
             self._source_config = normalizer.normalize()
 
+    def _migrate_config(
+        self,
+        config_path: Optional[str],
+        mutable_config: MutableMapping[str, Any],
+        config: Optional[Mapping[str, Any]],
+    ) -> None:
+        if config_path and config:
+            self._spec_component.migrate_config(mutable_config)
+            if mutable_config != config:
+                if config_path:
+                    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())
+
     def _migrate_manifest(self) -> None:
         """
         This method is used to migrate the manifest. It should be called after the manifest has been validated.