airbyte.cloud.connections

  1# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
  2"""Cloud Connections."""
  3
  4from __future__ import annotations
  5
  6import logging
  7from typing import TYPE_CHECKING, Any
  8
  9from typing_extensions import deprecated
 10
 11from airbyte._util import api_util
 12from airbyte.cloud._connection_state import (
 13    ConnectionStateResponse,
 14    _get_stream_list,
 15    _match_stream,
 16)
 17from airbyte.cloud.connectors import CloudDestination, CloudSource
 18from airbyte.cloud.sync_results import SyncResult
 19from airbyte.exceptions import AirbyteWorkspaceMismatchError, PyAirbyteInputError
 20
 21
 22logger = logging.getLogger(__name__)
 23
 24
 25if TYPE_CHECKING:
 26    from airbyte_api.models import ConnectionResponse, JobResponse, JobTypeEnum
 27
 28    from airbyte.cloud.workspaces import CloudWorkspace
 29
 30
 31class CloudConnection:  # noqa: PLR0904  # Too many public methods
 32    """A connection is an extract-load (EL) pairing of a source and destination in Airbyte Cloud.
 33
 34    You can use a connection object to run sync jobs, retrieve logs, and manage the connection.
 35    """
 36
 37    def __init__(
 38        self,
 39        workspace: CloudWorkspace,
 40        connection_id: str,
 41        source: str | None = None,
 42        destination: str | None = None,
 43    ) -> None:
 44        """It is not recommended to create a `CloudConnection` object directly.
 45
 46        Instead, use `CloudWorkspace.get_connection()` to create a connection object.
 47        """
 48        self.connection_id = connection_id
 49        """The ID of the connection."""
 50
 51        self.workspace = workspace
 52        """The workspace that the connection belongs to."""
 53
 54        self._source_id = source
 55        """The ID of the source."""
 56
 57        self._destination_id = destination
 58        """The ID of the destination."""
 59
 60        self._connection_info: ConnectionResponse | None = None
 61        """The connection info object. (Cached.)"""
 62
 63        self._cloud_source_object: CloudSource | None = None
 64        """The source object. (Cached.)"""
 65
 66        self._cloud_destination_object: CloudDestination | None = None
 67        """The destination object. (Cached.)"""
 68
 69    def _fetch_connection_info(
 70        self,
 71        *,
 72        force_refresh: bool = False,
 73        verify: bool = True,
 74    ) -> ConnectionResponse:
 75        """Fetch and cache connection info from the API.
 76
 77        By default, this method will only fetch from the API if connection info is not
 78        already cached. It also verifies that the connection belongs to the expected
 79        workspace unless verification is explicitly disabled.
 80
 81        Args:
 82            force_refresh: If True, always fetch from the API even if cached.
 83                If False (default), only fetch if not already cached.
 84            verify: If True (default), verify that the connection is valid (e.g., that
 85                the workspace_id matches this object's workspace). Raises an error if
 86                validation fails.
 87
 88        Returns:
 89            The ConnectionResponse from the API.
 90
 91        Raises:
 92            AirbyteWorkspaceMismatchError: If verify is True and the connection's
 93                workspace_id doesn't match the expected workspace.
 94            AirbyteMissingResourceError: If the connection doesn't exist.
 95        """
 96        if not force_refresh and self._connection_info is not None:
 97            # Use cached info, but still verify if requested
 98            if verify:
 99                self._verify_workspace_match(self._connection_info)
100            return self._connection_info
101
102        # Fetch from API
103        connection_info = api_util.get_connection(
104            workspace_id=self.workspace.workspace_id,
105            connection_id=self.connection_id,
106            api_root=self.workspace.api_root,
107            client_id=self.workspace.client_id,
108            client_secret=self.workspace.client_secret,
109            bearer_token=self.workspace.bearer_token,
110        )
111
112        # Cache the result first (before verification may raise)
113        self._connection_info = connection_info
114
115        # Verify if requested
116        if verify:
117            self._verify_workspace_match(connection_info)
118
119        return connection_info
120
121    def _verify_workspace_match(self, connection_info: ConnectionResponse) -> None:
122        """Verify that the connection belongs to the expected workspace.
123
124        Raises:
125            AirbyteWorkspaceMismatchError: If the workspace IDs don't match.
126        """
127        if connection_info.workspace_id != self.workspace.workspace_id:
128            raise AirbyteWorkspaceMismatchError(
129                resource_type="connection",
130                resource_id=self.connection_id,
131                workspace=self.workspace,
132                expected_workspace_id=self.workspace.workspace_id,
133                actual_workspace_id=connection_info.workspace_id,
134                message=(
135                    f"Connection '{self.connection_id}' belongs to workspace "
136                    f"'{connection_info.workspace_id}', not '{self.workspace.workspace_id}'."
137                ),
138            )
139
140    def check_is_valid(self) -> bool:
141        """Check if this connection exists and belongs to the expected workspace.
142
143        This method fetches connection info from the API (if not already cached) and
144        verifies that the connection's workspace_id matches the workspace associated
145        with this CloudConnection object.
146
147        Returns:
148            True if the connection exists and belongs to the expected workspace.
149
150        Raises:
151            AirbyteWorkspaceMismatchError: If the connection belongs to a different workspace.
152            AirbyteMissingResourceError: If the connection doesn't exist.
153        """
154        self._fetch_connection_info(force_refresh=False, verify=True)
155        return True
156
157    @classmethod
158    def _from_connection_response(
159        cls,
160        workspace: CloudWorkspace,
161        connection_response: ConnectionResponse,
162    ) -> CloudConnection:
163        """Create a CloudConnection from a ConnectionResponse."""
164        result = cls(
165            workspace=workspace,
166            connection_id=connection_response.connection_id,
167            source=connection_response.source_id,
168            destination=connection_response.destination_id,
169        )
170        result._connection_info = connection_response  # noqa: SLF001 # Accessing Non-Public API
171        return result
172
173    # Properties
174
175    @property
176    def name(self) -> str | None:
177        """Get the display name of the connection, if available.
178
179        E.g. "My Postgres to Snowflake", not the connection ID.
180        """
181        if not self._connection_info:
182            self._connection_info = self._fetch_connection_info()
183
184        return self._connection_info.name
185
186    @property
187    def source_id(self) -> str:
188        """The ID of the source."""
189        if not self._source_id:
190            if not self._connection_info:
191                self._connection_info = self._fetch_connection_info()
192
193            self._source_id = self._connection_info.source_id
194
195        return self._source_id
196
197    @property
198    def source(self) -> CloudSource:
199        """Get the source object."""
200        if self._cloud_source_object:
201            return self._cloud_source_object
202
203        self._cloud_source_object = CloudSource(
204            workspace=self.workspace,
205            connector_id=self.source_id,
206        )
207        return self._cloud_source_object
208
209    @property
210    def destination_id(self) -> str:
211        """The ID of the destination."""
212        if not self._destination_id:
213            if not self._connection_info:
214                self._connection_info = self._fetch_connection_info()
215
216            self._destination_id = self._connection_info.destination_id
217
218        return self._destination_id
219
220    @property
221    def destination(self) -> CloudDestination:
222        """Get the destination object."""
223        if self._cloud_destination_object:
224            return self._cloud_destination_object
225
226        self._cloud_destination_object = CloudDestination(
227            workspace=self.workspace,
228            connector_id=self.destination_id,
229        )
230        return self._cloud_destination_object
231
232    @property
233    def stream_names(self) -> list[str]:
234        """The stream names."""
235        if not self._connection_info:
236            self._connection_info = self._fetch_connection_info()
237
238        return [stream.name for stream in self._connection_info.configurations.streams or []]
239
240    @property
241    def table_prefix(self) -> str:
242        """The table prefix."""
243        if not self._connection_info:
244            self._connection_info = self._fetch_connection_info()
245
246        return self._connection_info.prefix or ""
247
248    @property
249    def connection_url(self) -> str | None:
250        """The web URL to the connection."""
251        return f"{self.workspace.workspace_url}/connections/{self.connection_id}"
252
253    @property
254    def job_history_url(self) -> str | None:
255        """The URL to the job history for the connection."""
256        return f"{self.connection_url}/timeline"
257
258    # Run Sync
259
260    def run_sync(
261        self,
262        *,
263        wait: bool = True,
264        wait_timeout: int = 300,
265    ) -> SyncResult:
266        """Run a sync."""
267        connection_response = api_util.run_connection(
268            connection_id=self.connection_id,
269            api_root=self.workspace.api_root,
270            workspace_id=self.workspace.workspace_id,
271            client_id=self.workspace.client_id,
272            client_secret=self.workspace.client_secret,
273            bearer_token=self.workspace.bearer_token,
274        )
275        sync_result = SyncResult(
276            workspace=self.workspace,
277            connection=self,
278            job_id=connection_response.job_id,
279        )
280
281        if wait:
282            sync_result.wait_for_completion(
283                wait_timeout=wait_timeout,
284                raise_failure=True,
285                raise_timeout=True,
286            )
287
288        return sync_result
289
290    def __repr__(self) -> str:
291        """String representation of the connection."""
292        return (
293            f"CloudConnection(connection_id={self.connection_id}, source_id={self.source_id}, "
294            f"destination_id={self.destination_id}, connection_url={self.connection_url})"
295        )
296
297    # Logs
298
299    def get_previous_sync_logs(
300        self,
301        *,
302        limit: int = 20,
303        offset: int | None = None,
304        from_tail: bool = True,
305        job_type: JobTypeEnum | None = None,
306    ) -> list[SyncResult]:
307        """Get previous sync jobs for a connection with pagination support.
308
309        Returns SyncResult objects containing job metadata (job_id, status, bytes_synced,
310        rows_synced, start_time). Full log text can be fetched lazily via
311        `SyncResult.get_full_log_text()`.
312
313        Args:
314            limit: Maximum number of jobs to return. Defaults to 20.
315            offset: Number of jobs to skip from the beginning. Defaults to None (0).
316            from_tail: If True, returns jobs ordered newest-first (createdAt DESC).
317                If False, returns jobs ordered oldest-first (createdAt ASC).
318                Defaults to True.
319            job_type: Filter by job type (e.g., JobTypeEnum.SYNC, JobTypeEnum.REFRESH).
320                If not specified, defaults to sync and reset jobs only (API default behavior).
321
322        Returns:
323            A list of SyncResult objects representing the sync jobs.
324        """
325        order_by = (
326            api_util.JOB_ORDER_BY_CREATED_AT_DESC
327            if from_tail
328            else api_util.JOB_ORDER_BY_CREATED_AT_ASC
329        )
330        sync_logs: list[JobResponse] = api_util.get_job_logs(
331            connection_id=self.connection_id,
332            api_root=self.workspace.api_root,
333            workspace_id=self.workspace.workspace_id,
334            limit=limit,
335            offset=offset,
336            order_by=order_by,
337            job_type=job_type,
338            client_id=self.workspace.client_id,
339            client_secret=self.workspace.client_secret,
340            bearer_token=self.workspace.bearer_token,
341        )
342        return [
343            SyncResult(
344                workspace=self.workspace,
345                connection=self,
346                job_id=sync_log.job_id,
347                _latest_job_info=sync_log,
348            )
349            for sync_log in sync_logs
350        ]
351
352    def get_sync_result(
353        self,
354        job_id: int | None = None,
355    ) -> SyncResult | None:
356        """Get the sync result for the connection.
357
358        If `job_id` is not provided, the most recent sync job will be used.
359
360        Returns `None` if job_id is omitted and no previous jobs are found.
361        """
362        if job_id is None:
363            # Get the most recent sync job
364            results = self.get_previous_sync_logs(
365                limit=1,
366            )
367            if results:
368                return results[0]
369
370            return None
371
372        # Get the sync job by ID (lazy loaded)
373        return SyncResult(
374            workspace=self.workspace,
375            connection=self,
376            job_id=job_id,
377        )
378
379    # Artifacts
380
381    @deprecated("Use 'dump_raw_state()' instead.")
382    def get_state_artifacts(self) -> list[dict[str, Any]] | None:
383        """Deprecated. Use `dump_raw_state()` instead."""
384        state_response = api_util.get_connection_state(
385            connection_id=self.connection_id,
386            api_root=self.workspace.api_root,
387            client_id=self.workspace.client_id,
388            client_secret=self.workspace.client_secret,
389            bearer_token=self.workspace.bearer_token,
390        )
391        if state_response.get("stateType") == "not_set":
392            return None
393        return state_response.get("streamState", [])
394
395    def dump_raw_state(self) -> dict[str, Any]:
396        """Dump the full raw state for this connection.
397
398        Returns the connection's sync state as a raw dictionary from the API.
399        The result includes stateType, connectionId, and all state data.
400
401        The output of this method can be passed directly to `import_raw_state()`
402        on the same or a different connection (connectionId is overridden on import).
403
404        Returns:
405            The full connection state as a dictionary.
406        """
407        return api_util.get_connection_state(
408            connection_id=self.connection_id,
409            api_root=self.workspace.api_root,
410            client_id=self.workspace.client_id,
411            client_secret=self.workspace.client_secret,
412            bearer_token=self.workspace.bearer_token,
413        )
414
415    def import_raw_state(
416        self,
417        connection_state_dict: dict[str, Any],
418    ) -> dict[str, Any]:
419        """Import (restore) the full raw state for this connection.
420
421        > ⚠️ **WARNING:** Modifying the state directly is not recommended and
422        > could result in broken connections, and/or incorrect sync behavior.
423
424        Replaces the entire connection state with the provided state blob.
425        Uses the safe variant that prevents updates while a sync is running (HTTP 423).
426
427        This is the counterpart to `dump_raw_state()` for backup/restore workflows.
428        The `connectionId` in the blob is always overridden with this connection's
429        ID, making state blobs portable across connections.
430
431        Args:
432            connection_state_dict: The full connection state dict to import. Must include:
433                - stateType: "global", "stream", or "legacy"
434                - One of: state (legacy), streamState (stream), globalState (global)
435
436        Returns:
437            The updated connection state as a dictionary.
438
439        Raises:
440            AirbyteConnectionSyncActiveError: If a sync is currently running on this
441                connection (HTTP 423). Wait for the sync to complete before retrying.
442        """
443        return api_util.replace_connection_state(
444            connection_id=self.connection_id,
445            connection_state_dict=connection_state_dict,
446            api_root=self.workspace.api_root,
447            client_id=self.workspace.client_id,
448            client_secret=self.workspace.client_secret,
449            bearer_token=self.workspace.bearer_token,
450        )
451
452    def get_stream_state(
453        self,
454        stream_name: str,
455        stream_namespace: str | None = None,
456    ) -> dict[str, Any] | None:
457        """Get the state blob for a single stream within this connection.
458
459        Returns just the stream's state dictionary (e.g., {"cursor": "2024-01-01"}),
460        not the full connection state envelope.
461
462        This is compatible with `stream`-type state and stream-level entries
463        within a `global`-type state. It is not compatible with `legacy` state.
464        To get or set the entire connection-level state artifact, use
465        `dump_raw_state` and `import_raw_state` instead.
466
467        Args:
468            stream_name: The name of the stream to get state for.
469            stream_namespace: The source-side stream namespace. This refers to the
470                namespace from the source (e.g., database schema), not any destination
471                namespace override set in connection advanced settings.
472
473        Returns:
474            The stream's state blob as a dictionary, or None if the stream is not found.
475        """
476        state_data = self.dump_raw_state()
477        result = ConnectionStateResponse(**state_data)
478
479        streams = _get_stream_list(result)
480        matching = [s for s in streams if _match_stream(s, stream_name, stream_namespace)]
481
482        if not matching:
483            available = [s.stream_descriptor.name for s in streams]
484            logger.warning(
485                "Stream '%s' not found in connection state for connection '%s'. "
486                "Available streams: %s",
487                stream_name,
488                self.connection_id,
489                available,
490            )
491            return None
492
493        return matching[0].stream_state
494
495    def set_stream_state(
496        self,
497        stream_name: str,
498        state_blob_dict: dict[str, Any],
499        stream_namespace: str | None = None,
500    ) -> None:
501        """Set the state for a single stream within this connection.
502
503        Fetches the current full state, replaces only the specified stream's state,
504        then sends the full updated state back to the API. If the stream does not
505        exist in the current state, it is appended.
506
507        This is compatible with `stream`-type state and stream-level entries
508        within a `global`-type state. It is not compatible with `legacy` state.
509        To get or set the entire connection-level state artifact, use
510        `dump_raw_state` and `import_raw_state` instead.
511
512        Uses the safe variant that prevents updates while a sync is running (HTTP 423).
513
514        Args:
515            stream_name: The name of the stream to update state for.
516            state_blob_dict: The state blob dict for this stream (e.g., {"cursor": "2024-01-01"}).
517            stream_namespace: The source-side stream namespace. This refers to the
518                namespace from the source (e.g., database schema), not any destination
519                namespace override set in connection advanced settings.
520
521        Raises:
522            PyAirbyteInputError: If the connection state type is not supported for
523                stream-level operations (not_set, legacy).
524            AirbyteConnectionSyncActiveError: If a sync is currently running on this
525                connection (HTTP 423). Wait for the sync to complete before retrying.
526        """
527        state_data = self.dump_raw_state()
528        current = ConnectionStateResponse(**state_data)
529
530        if current.state_type == "not_set":
531            raise PyAirbyteInputError(
532                message="Cannot set stream state: connection has no existing state.",
533                context={"connection_id": self.connection_id},
534            )
535
536        if current.state_type == "legacy":
537            raise PyAirbyteInputError(
538                message="Cannot set stream state on a legacy-type connection state.",
539                context={"connection_id": self.connection_id},
540            )
541
542        new_stream_entry = {
543            "streamDescriptor": {
544                "name": stream_name,
545                **(
546                    {
547                        "namespace": stream_namespace,
548                    }
549                    if stream_namespace
550                    else {}
551                ),
552            },
553            "streamState": state_blob_dict,
554        }
555
556        raw_streams: list[dict[str, Any]]
557        if current.state_type == "stream":
558            raw_streams = state_data.get("streamState", [])
559        elif current.state_type == "global":
560            raw_streams = state_data.get("globalState", {}).get("streamStates", [])
561        else:
562            raw_streams = []
563
564        streams = _get_stream_list(current)
565        found = False
566        updated_streams_raw: list[dict[str, Any]] = []
567        for raw_s, parsed_s in zip(raw_streams, streams, strict=False):
568            if _match_stream(parsed_s, stream_name, stream_namespace):
569                updated_streams_raw.append(new_stream_entry)
570                found = True
571            else:
572                updated_streams_raw.append(raw_s)
573
574        if not found:
575            updated_streams_raw.append(new_stream_entry)
576
577        full_state: dict[str, Any] = {
578            **state_data,
579        }
580
581        if current.state_type == "stream":
582            full_state["streamState"] = updated_streams_raw
583        elif current.state_type == "global":
584            original_global = state_data.get("globalState", {})
585            full_state["globalState"] = {
586                **original_global,
587                "streamStates": updated_streams_raw,
588            }
589
590        self.import_raw_state(full_state)
591
592    @deprecated("Use 'dump_raw_catalog()' instead.")
593    def get_catalog_artifact(self) -> dict[str, Any] | None:
594        """Get the configured catalog for this connection.
595
596        Returns the full configured catalog (syncCatalog) for this connection,
597        including stream schemas, sync modes, cursor fields, and primary keys.
598
599        Uses the Config API endpoint: POST /v1/web_backend/connections/get
600
601        Returns:
602            Dictionary containing the configured catalog, or `None` if not found.
603        """
604        return self.dump_raw_catalog()
605
606    def dump_raw_catalog(self) -> dict[str, Any] | None:
607        """Dump the full configured catalog (syncCatalog) for this connection.
608
609        Returns the raw catalog dict as returned by the API, including stream
610        schemas, sync modes, cursor fields, and primary keys.
611
612        The returned dict can be passed to `import_raw_catalog()` on this or
613        another connection to restore or clone the catalog configuration.
614
615        Returns:
616            Dictionary containing the configured catalog, or `None` if not found.
617        """
618        connection_response = api_util.get_connection_catalog(
619            connection_id=self.connection_id,
620            api_root=self.workspace.api_root,
621            client_id=self.workspace.client_id,
622            client_secret=self.workspace.client_secret,
623            bearer_token=self.workspace.bearer_token,
624        )
625        return connection_response.get("syncCatalog")
626
627    def import_raw_catalog(self, catalog: dict[str, Any]) -> None:
628        """Replace the configured catalog for this connection.
629
630        > ⚠️ **WARNING:** Modifying the catalog directly is not recommended and
631        > could result in broken connections, and/or incorrect sync behavior.
632
633        Accepts a configured catalog dict and replaces the connection's entire
634        catalog with it. All other connection settings remain unchanged.
635
636        The catalog shape should match the output of `dump_raw_catalog()`:
637        `{"streams": [{"stream": {...}, "config": {...}}, ...]}`.
638
639        Args:
640            catalog: The configured catalog dict to set.
641        """
642        api_util.replace_connection_catalog(
643            connection_id=self.connection_id,
644            configured_catalog_dict=catalog,
645            api_root=self.workspace.api_root,
646            client_id=self.workspace.client_id,
647            client_secret=self.workspace.client_secret,
648            bearer_token=self.workspace.bearer_token,
649        )
650
651    def rename(self, name: str) -> CloudConnection:
652        """Rename the connection.
653
654        Args:
655            name: New name for the connection
656
657        Returns:
658            Updated CloudConnection object with refreshed info
659        """
660        updated_response = api_util.patch_connection(
661            connection_id=self.connection_id,
662            api_root=self.workspace.api_root,
663            client_id=self.workspace.client_id,
664            client_secret=self.workspace.client_secret,
665            bearer_token=self.workspace.bearer_token,
666            name=name,
667        )
668        self._connection_info = updated_response
669        return self
670
671    def set_table_prefix(self, prefix: str) -> CloudConnection:
672        """Set the table prefix for the connection.
673
674        Args:
675            prefix: New table prefix to use when syncing to the destination
676
677        Returns:
678            Updated CloudConnection object with refreshed info
679        """
680        updated_response = api_util.patch_connection(
681            connection_id=self.connection_id,
682            api_root=self.workspace.api_root,
683            client_id=self.workspace.client_id,
684            client_secret=self.workspace.client_secret,
685            bearer_token=self.workspace.bearer_token,
686            prefix=prefix,
687        )
688        self._connection_info = updated_response
689        return self
690
691    def set_selected_streams(self, stream_names: list[str]) -> CloudConnection:
692        """Set the selected streams for the connection.
693
694        This is a destructive operation that can break existing connections if the
695        stream selection is changed incorrectly. Use with caution.
696
697        Args:
698            stream_names: List of stream names to sync
699
700        Returns:
701            Updated CloudConnection object with refreshed info
702        """
703        configurations = api_util.build_stream_configurations(stream_names)
704
705        updated_response = api_util.patch_connection(
706            connection_id=self.connection_id,
707            api_root=self.workspace.api_root,
708            client_id=self.workspace.client_id,
709            client_secret=self.workspace.client_secret,
710            bearer_token=self.workspace.bearer_token,
711            configurations=configurations,
712        )
713        self._connection_info = updated_response
714        return self
715
716    # Enable/Disable
717
718    @property
719    def enabled(self) -> bool:
720        """Get the current enabled status of the connection.
721
722        This property always fetches fresh data from the API to ensure accuracy,
723        as another process or user may have toggled the setting.
724
725        Returns:
726            True if the connection status is 'active', False otherwise.
727        """
728        connection_info = self._fetch_connection_info(force_refresh=True)
729        return connection_info.status == api_util.models.ConnectionStatusEnum.ACTIVE
730
731    @enabled.setter
732    def enabled(self, value: bool) -> None:
733        """Set the enabled status of the connection.
734
735        Args:
736            value: True to enable (set status to 'active'), False to disable
737                (set status to 'inactive').
738        """
739        self.set_enabled(enabled=value)
740
741    def set_enabled(
742        self,
743        *,
744        enabled: bool,
745        ignore_noop: bool = True,
746    ) -> None:
747        """Set the enabled status of the connection.
748
749        Args:
750            enabled: True to enable (set status to 'active'), False to disable
751                (set status to 'inactive').
752            ignore_noop: If True (default), silently return if the connection is already
753                in the requested state. If False, raise ValueError when the requested
754                state matches the current state.
755
756        Raises:
757            ValueError: If ignore_noop is False and the connection is already in the
758                requested state.
759        """
760        # Always fetch fresh data to check current status
761        connection_info = self._fetch_connection_info(force_refresh=True)
762        current_status = connection_info.status
763        desired_status = (
764            api_util.models.ConnectionStatusEnum.ACTIVE
765            if enabled
766            else api_util.models.ConnectionStatusEnum.INACTIVE
767        )
768
769        if current_status == desired_status:
770            if ignore_noop:
771                return
772            raise ValueError(
773                f"Connection is already {'enabled' if enabled else 'disabled'}. "
774                f"Current status: {current_status}"
775            )
776
777        updated_response = api_util.patch_connection(
778            connection_id=self.connection_id,
779            api_root=self.workspace.api_root,
780            client_id=self.workspace.client_id,
781            client_secret=self.workspace.client_secret,
782            bearer_token=self.workspace.bearer_token,
783            status=desired_status,
784        )
785        self._connection_info = updated_response
786
787    # Scheduling
788
789    def set_schedule(
790        self,
791        cron_expression: str,
792    ) -> None:
793        """Set a cron schedule for the connection.
794
795        Args:
796            cron_expression: A cron expression defining when syncs should run.
797
798        Examples:
799                - "0 0 * * *" - Daily at midnight UTC
800                - "0 */6 * * *" - Every 6 hours
801                - "0 0 * * 0" - Weekly on Sunday at midnight UTC
802        """
803        schedule = api_util.models.AirbyteAPIConnectionSchedule(
804            schedule_type=api_util.models.ScheduleTypeEnum.CRON,
805            cron_expression=cron_expression,
806        )
807        updated_response = api_util.patch_connection(
808            connection_id=self.connection_id,
809            api_root=self.workspace.api_root,
810            client_id=self.workspace.client_id,
811            client_secret=self.workspace.client_secret,
812            bearer_token=self.workspace.bearer_token,
813            schedule=schedule,
814        )
815        self._connection_info = updated_response
816
817    def set_manual_schedule(self) -> None:
818        """Set the connection to manual scheduling.
819
820        Disables automatic syncs. Syncs will only run when manually triggered.
821        """
822        schedule = api_util.models.AirbyteAPIConnectionSchedule(
823            schedule_type=api_util.models.ScheduleTypeEnum.MANUAL,
824        )
825        updated_response = api_util.patch_connection(
826            connection_id=self.connection_id,
827            api_root=self.workspace.api_root,
828            client_id=self.workspace.client_id,
829            client_secret=self.workspace.client_secret,
830            bearer_token=self.workspace.bearer_token,
831            schedule=schedule,
832        )
833        self._connection_info = updated_response
834
835    # Deletions
836
837    def permanently_delete(
838        self,
839        *,
840        cascade_delete_source: bool = False,
841        cascade_delete_destination: bool = False,
842    ) -> None:
843        """Delete the connection.
844
845        Args:
846            cascade_delete_source: Whether to also delete the source.
847            cascade_delete_destination: Whether to also delete the destination.
848        """
849        self.workspace.permanently_delete_connection(self)
850
851        if cascade_delete_source:
852            self.workspace.permanently_delete_source(self.source_id)
853
854        if cascade_delete_destination:
855            self.workspace.permanently_delete_destination(self.destination_id)