Adds config_change_callback to Destinations and Sources

airbyte/_connector_base.py

@@ -56,13 +56,15 @@ def __init__(
         executor: Executor,
         name: str,
         config: dict[str, Any] | None = None,
+        config_change_callback: Callable[[dict[str, Any], int], None] | None = None,
         *,
         validate: bool = False,
     ) -> None:
         """Initialize the source.
 
         If config is provided, it will be validated against the spec if validate is True.
         """
+        self.config_change_callback = config_change_callback
         self.executor = executor
         self._name = name
         self._config_dict: dict[str, Any] | None = None
@@ -361,7 +363,8 @@ def _peek_airbyte_message(
 
         This method handles reading Airbyte messages and taking action, if needed, based on the
         message type. For instance, log messages are logged, records are tallied, and errors are
-        raised as exceptions if `raise_on_error` is True.
+        raised as exceptions if `raise_on_error` is True. If a config change message is received,
+        the config change callback is called.
 
         Raises:
             AirbyteConnectorFailedError: If a TRACE message of type ERROR is emitted.
@@ -380,6 +383,16 @@ def _peek_airbyte_message(
                 )
             return
 
+        if (
+            message.type == "CONTROL"
+            and message.control.type == "CONNECTOR_CONFIG"
+            and self.config_change_callback is not None
+        ):
+            self.config_change_callback(
+                message.control.config, message.control.emitted_at
+            )
+            return
+
     def _execute(
         self,
         args: list[str],

airbyte/destinations/base.py

@@ -48,6 +48,7 @@ def __init__(
         executor: Executor,
         name: str,
         config: dict[str, Any] | None = None,
+        config_change_callback: Callable[[dict[str, Any]], None] | None = None,
         *,
         validate: bool = False,
     ) -> None:
@@ -59,6 +60,7 @@ def __init__(
             executor=executor,
             name=name,
             config=config,
+            config_change_callback=config_change_callback,
             validate=validate,
         )
 

airbyte/sources/base.py

@@ -58,6 +58,7 @@ def __init__(
         executor: Executor,
         name: str,
         config: dict[str, Any] | None = None,
+        config_change_callback: Callable[[dict[str, Any]], None] | None = None,
         streams: str | list[str] | None = None,
         *,
         validate: bool = False,
@@ -73,6 +74,7 @@ def __init__(
             executor=executor,
             name=name,
             config=config,
+            config_change_callback=config_change_callback,
             validate=validate,
         )
         self._config_dict: dict[str, Any] | None = None