Feat: Adds config_change_callback to Destinations and Sources

airbyte/_connector_base.py

@@ -35,7 +35,7 @@
 
 if TYPE_CHECKING:
     import logging
-    from collections.abc import Generator
+    from collections.abc import Callable, Generator
     from typing import IO
 
     from airbyte._executors.base import Executor
@@ -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,14 @@ 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

@@ -33,6 +33,8 @@
 
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from airbyte._executors.base import Executor
     from airbyte.caches.base import CacheBase
     from airbyte.shared.state_writers import StateWriterBase
@@ -48,6 +50,7 @@ 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:
@@ -59,6 +62,7 @@ def __init__(
             executor=executor,
             name=name,
             config=config,
+            config_change_callback=config_change_callback,
             validate=validate,
         )
 

airbyte/destinations/util.py

@@ -13,12 +13,14 @@
 
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
     from pathlib import Path
 
 
 def get_destination(
     name: str,
     config: dict[str, Any] | None = None,
+    config_change_callback: Callable[[dict[str, Any], int], None] | None = None,
     *,
     version: str | None = None,
     pip_url: str | None = None,
@@ -58,6 +60,7 @@ def get_destination(
     return Destination(
         name=name,
         config=config,
+        config_change_callback=config_change_callback,
         executor=get_connector_executor(
             name=name,
             version=version,

airbyte/sources/base.py

@@ -36,7 +36,7 @@
 
 
 if TYPE_CHECKING:
-    from collections.abc import Generator, Iterable, Iterator
+    from collections.abc import Callable, Generator, Iterable, Iterator
 
     from airbyte_cdk import ConnectorSpecification
     from airbyte_protocol.models import AirbyteStream
@@ -58,6 +58,7 @@ def __init__(
         executor: Executor,
         name: str,
         config: dict[str, Any] | None = None,
+        config_change_callback: Callable[[dict[str, Any], int], 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

airbyte/sources/util.py

@@ -13,6 +13,7 @@
 
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
     from pathlib import Path
 
 
@@ -45,6 +46,7 @@ def get_connector(
 def get_source(  # noqa: PLR0913 # Too many arguments
     name: str,
     config: dict[str, Any] | None = None,
+    config_change_callback: Callable[[dict[str, Any], int], None] | None = None,
     *,
     streams: str | list[str] | None = None,
     version: str | None = None,
@@ -103,6 +105,7 @@ def get_source(  # noqa: PLR0913 # Too many arguments
     return Source(
         name=name,
         config=config,
+        config_change_callback=config_change_callback,
         streams=streams,
         executor=get_connector_executor(
             name=name,