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