feat: Add interfaces for editing cloud source, destination, and connection configs (#841)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

Author: aaronsteers

airbyte/_util/api_util.py

@@ -564,6 +564,57 @@ def delete_source(
         )
 
 
+def patch_source(
+    source_id: str,
+    *,
+    api_root: str,
+    client_id: SecretString,
+    client_secret: SecretString,
+    name: str | None = None,
+    config: models.SourceConfiguration | dict[str, Any] | None = None,
+) -> models.SourceResponse:
+    """Update/patch a source configuration.
+
+    This is a destructive operation that can break existing connections if the
+    configuration is changed incorrectly.
+
+    Args:
+        source_id: The ID of the source to update
+        api_root: The API root URL
+        client_id: Client ID for authentication
+        client_secret: Client secret for authentication
+        name: Optional new name for the source
+        config: Optional new configuration for the source
+
+    Returns:
+        Updated SourceResponse object
+    """
+    airbyte_instance = get_airbyte_server_instance(
+        client_id=client_id,
+        client_secret=client_secret,
+        api_root=api_root,
+    )
+    response = airbyte_instance.sources.patch_source(
+        api.PatchSourceRequest(
+            source_id=source_id,
+            source_patch_request=models.SourcePatchRequest(
+                name=name,
+                configuration=config,
+            ),
+        ),
+    )
+    if status_ok(response.status_code) and response.source_response:
+        return response.source_response
+
+    raise AirbyteError(
+        message="Could not update source.",
+        context={
+            "source_id": source_id,
+        },
+        response=response,
+    )
+
+
 # Utility function
 
 
@@ -701,9 +752,80 @@ def delete_destination(
         )
 
 
+def patch_destination(
+    destination_id: str,
+    *,
+    api_root: str,
+    client_id: SecretString,
+    client_secret: SecretString,
+    name: str | None = None,
+    config: DestinationConfiguration | dict[str, Any] | None = None,
+) -> models.DestinationResponse:
+    """Update/patch a destination configuration.
+
+    This is a destructive operation that can break existing connections if the
+    configuration is changed incorrectly.
+
+    Args:
+        destination_id: The ID of the destination to update
+        api_root: The API root URL
+        client_id: Client ID for authentication
+        client_secret: Client secret for authentication
+        name: Optional new name for the destination
+        config: Optional new configuration for the destination
+
+    Returns:
+        Updated DestinationResponse object
+    """
+    airbyte_instance = get_airbyte_server_instance(
+        client_id=client_id,
+        client_secret=client_secret,
+        api_root=api_root,
+    )
+    response = airbyte_instance.destinations.patch_destination(
+        api.PatchDestinationRequest(
+            destination_id=destination_id,
+            destination_patch_request=models.DestinationPatchRequest(
+                name=name,
+                configuration=config,
+            ),
+        ),
+    )
+    if status_ok(response.status_code) and response.destination_response:
+        return response.destination_response
+
+    raise AirbyteError(
+        message="Could not update destination.",
+        context={
+            "destination_id": destination_id,
+        },
+        response=response,
+    )
+
+
 # Create and delete connections
 
 
+def build_stream_configurations(
+    stream_names: list[str],
+) -> models.StreamConfigurations:
+    """Build a StreamConfigurations object from a list of stream names.
+
+    This helper creates the proper API model structure for stream configurations.
+    Used by both connection creation and updates.
+
+    Args:
+        stream_names: List of stream names to include in the configuration
+
+    Returns:
+        StreamConfigurations object ready for API submission
+    """
+    stream_configurations = [
+        models.StreamConfiguration(name=stream_name) for stream_name in stream_names
+    ]
+    return models.StreamConfigurations(streams=stream_configurations)
+
+
 def create_connection(  # noqa: PLR0913  # Too many arguments
     name: str,
     *,
@@ -722,15 +844,7 @@ def create_connection(  # noqa: PLR0913  # Too many arguments
         client_secret=client_secret,
         api_root=api_root,
     )
-    stream_configurations: list[models.StreamConfiguration] = []
-    if selected_stream_names:
-        for stream_name in selected_stream_names:
-            stream_configuration = models.StreamConfiguration(
-                name=stream_name,
-            )
-            stream_configurations.append(stream_configuration)
-
-    stream_configurations_obj = models.StreamConfigurations(stream_configurations)
+    stream_configurations_obj = build_stream_configurations(selected_stream_names)
     response = airbyte_instance.connections.create_connection(
         models.ConnectionCreateRequest(
             name=name,
@@ -815,6 +929,66 @@ def delete_connection(
         )
 
 
+def patch_connection(  # noqa: PLR0913  # Too many arguments
+    connection_id: str,
+    *,
+    api_root: str,
+    client_id: SecretString,
+    client_secret: SecretString,
+    name: str | None = None,
+    configurations: models.StreamConfigurationsInput | None = None,
+    schedule: models.AirbyteAPIConnectionSchedule | None = None,
+    prefix: str | None = None,
+    status: models.ConnectionStatusEnum | None = None,
+) -> models.ConnectionResponse:
+    """Update/patch a connection configuration.
+
+    This is a destructive operation that can break existing connections if the
+    configuration is changed incorrectly.
+
+    Args:
+        connection_id: The ID of the connection to update
+        api_root: The API root URL
+        client_id: Client ID for authentication
+        client_secret: Client secret for authentication
+        name: Optional new name for the connection
+        configurations: Optional new stream configurations
+        schedule: Optional new sync schedule
+        prefix: Optional new table prefix
+        status: Optional new connection status
+
+    Returns:
+        Updated ConnectionResponse object
+    """
+    airbyte_instance = get_airbyte_server_instance(
+        client_id=client_id,
+        client_secret=client_secret,
+        api_root=api_root,
+    )
+    response = airbyte_instance.connections.patch_connection(
+        api.PatchConnectionRequest(
+            connection_id=connection_id,
+            connection_patch_request=models.ConnectionPatchRequest(
+                name=name,
+                configurations=configurations,
+                schedule=schedule,
+                prefix=prefix,
+                status=status,
+            ),
+        ),
+    )
+    if status_ok(response.status_code) and response.connection_response:
+        return response.connection_response
+
+    raise AirbyteError(
+        message="Could not update connection.",
+        context={
+            "connection_id": connection_id,
+        },
+        response=response,
+    )
+
+
 # Functions for leveraging the Airbyte Config API (may not be supported or stable)
 
 

airbyte/cloud/connections.py

@@ -256,6 +256,68 @@ def get_sync_result(
             job_id=job_id,
         )
 
+    def rename(self, name: str) -> CloudConnection:
+        """Rename the connection.
+
+        Args:
+            name: New name for the connection
+
+        Returns:
+            Updated CloudConnection object with refreshed info
+        """
+        updated_response = api_util.patch_connection(
+            connection_id=self.connection_id,
+            api_root=self.workspace.api_root,
+            client_id=self.workspace.client_id,
+            client_secret=self.workspace.client_secret,
+            name=name,
+        )
+        self._connection_info = updated_response
+        return self
+
+    def set_table_prefix(self, prefix: str) -> CloudConnection:
+        """Set the table prefix for the connection.
+
+        Args:
+            prefix: New table prefix to use when syncing to the destination
+
+        Returns:
+            Updated CloudConnection object with refreshed info
+        """
+        updated_response = api_util.patch_connection(
+            connection_id=self.connection_id,
+            api_root=self.workspace.api_root,
+            client_id=self.workspace.client_id,
+            client_secret=self.workspace.client_secret,
+            prefix=prefix,
+        )
+        self._connection_info = updated_response
+        return self
+
+    def set_selected_streams(self, stream_names: list[str]) -> CloudConnection:
+        """Set the selected streams for the connection.
+
+        This is a destructive operation that can break existing connections if the
+        stream selection is changed incorrectly. Use with caution.
+
+        Args:
+            stream_names: List of stream names to sync
+
+        Returns:
+            Updated CloudConnection object with refreshed info
+        """
+        configurations = api_util.build_stream_configurations(stream_names)
+
+        updated_response = api_util.patch_connection(
+            connection_id=self.connection_id,
+            api_root=self.workspace.api_root,
+            client_id=self.workspace.client_id,
+            client_secret=self.workspace.client_secret,
+            configurations=configurations,
+        )
+        self._connection_info = updated_response
+        return self
+
     # Deletions
 
     def permanently_delete(

airbyte/cloud/connectors.py

@@ -199,6 +199,47 @@ def _fetch_connector_info(self) -> api_models.SourceResponse:
             client_secret=self.workspace.client_secret,
         )
 
+    def rename(self, name: str) -> CloudSource:
+        """Rename the source.
+
+        Args:
+            name: New name for the source
+
+        Returns:
+            Updated CloudSource object with refreshed info
+        """
+        updated_response = api_util.patch_source(
+            source_id=self.connector_id,
+            api_root=self.workspace.api_root,
+            client_id=self.workspace.client_id,
+            client_secret=self.workspace.client_secret,
+            name=name,
+        )
+        self._connector_info = updated_response
+        return self
+
+    def update_config(self, config: dict[str, Any]) -> CloudSource:
+        """Update the source configuration.
+
+        This is a destructive operation that can break existing connections if the
+        configuration is changed incorrectly. Use with caution.
+
+        Args:
+            config: New configuration for the source
+
+        Returns:
+            Updated CloudSource object with refreshed info
+        """
+        updated_response = api_util.patch_source(
+            source_id=self.connector_id,
+            api_root=self.workspace.api_root,
+            client_id=self.workspace.client_id,
+            client_secret=self.workspace.client_secret,
+            config=config,
+        )
+        self._connector_info = updated_response
+        return self
+
     @classmethod
     def _from_source_response(
         cls,
@@ -240,6 +281,47 @@ def _fetch_connector_info(self) -> api_models.DestinationResponse:
             client_secret=self.workspace.client_secret,
         )
 
+    def rename(self, name: str) -> CloudDestination:
+        """Rename the destination.
+
+        Args:
+            name: New name for the destination
+
+        Returns:
+            Updated CloudDestination object with refreshed info
+        """
+        updated_response = api_util.patch_destination(
+            destination_id=self.connector_id,
+            api_root=self.workspace.api_root,
+            client_id=self.workspace.client_id,
+            client_secret=self.workspace.client_secret,
+            name=name,
+        )
+        self._connector_info = updated_response
+        return self
+
+    def update_config(self, config: dict[str, Any]) -> CloudDestination:
+        """Update the destination configuration.
+
+        This is a destructive operation that can break existing connections if the
+        configuration is changed incorrectly. Use with caution.
+
+        Args:
+            config: New configuration for the destination
+
+        Returns:
+            Updated CloudDestination object with refreshed info
+        """
+        updated_response = api_util.patch_destination(
+            destination_id=self.connector_id,
+            api_root=self.workspace.api_root,
+            client_id=self.workspace.client_id,
+            client_secret=self.workspace.client_secret,
+            config=config,
+        )
+        self._connector_info = updated_response
+        return self
+
     @classmethod
     def _from_destination_response(
         cls,