airbyte.cloud

PyAirbyte classes and methods for interacting with the Airbyte Cloud API.

You can use this module to interact with Airbyte Cloud, OSS, and Enterprise.

Self-managed Airbyte instances

For self-managed Airbyte instances, set api_root to the Public API root for your deployment. For the default self-managed route, that usually ends in /api/public/v1. PyAirbyte uses the Public API for workspace and organization discovery.

Some Cloud module methods also call the Config API, including methods such as CloudConnection.dump_raw_catalog(), which reads the configured catalog directly from Airbyte. For documented self-managed deployments where the Public API root ends in /api/public/v1, PyAirbyte infers the Config API root by replacing that suffix with /api/v1.

If your deployment uses custom ingress or a nonstandard reverse proxy, pass config_api_root explicitly or set the AIRBYTE_CLOUD_CONFIG_API_URL environment variable.

from airbyte import cloud

workspace = cloud.CloudWorkspace(
    workspace_id="...",
    client_id="...",
    client_secret="...",
    api_root="https://airbyte.example.com/api/public/v1",
    config_api_root="https://airbyte.example.com/api/v1",
)

connection = workspace.get_connection(connection_id="...")
raw_catalog = connection.dump_raw_catalog()

Examples

Basic Sync Example:

import airbyte as ab
from airbyte import cloud

# Initialize an Airbyte Cloud workspace object
workspace = cloud.CloudWorkspace(
    workspace_id="123",
    api_key=ab.get_secret("AIRBYTE_CLOUD_API_KEY"),
)

# Run a sync job on Airbyte Cloud
connection = workspace.get_connection(connection_id="456")
sync_result = connection.run_sync()
print(sync_result.get_job_status())

Example Read From Cloud Destination:

If your destination is supported, you can read records directly from the SyncResult object. Currently this is supported in Snowflake and BigQuery only.

# Assuming we've already created a `connection` object...

# Get the latest job result and print the stream names
sync_result = connection.get_sync_result()
print(sync_result.stream_names)

# Get a dataset from the sync result
dataset: CachedDataset = sync_result.get_dataset("users")

# Get a SQLAlchemy table to use in SQL queries...
users_table = dataset.to_sql_table()
print(f"Table name: {users_table.name}")

# Or iterate over the dataset directly
for record in dataset:
    print(record)

View Source

  1# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
  2"""PyAirbyte classes and methods for interacting with the Airbyte Cloud API.
  3
  4You can use this module to interact with Airbyte Cloud, OSS, and Enterprise.
  5
  6## Self-managed Airbyte instances
  7
  8For self-managed Airbyte instances, set `api_root` to the Public API root for your
  9deployment. For the default self-managed route, that usually ends in `/api/public/v1`.
 10PyAirbyte uses the Public API for workspace and organization discovery.
 11
 12Some Cloud module methods also call the Config API, including methods such as
 13`CloudConnection.dump_raw_catalog()`, which reads the configured catalog directly
 14from Airbyte. For documented self-managed deployments where the Public API root ends in
 15`/api/public/v1`, PyAirbyte infers the Config API root by replacing that suffix with
 16`/api/v1`.
 17
 18If your deployment uses custom ingress or a nonstandard reverse proxy, pass
 19`config_api_root` explicitly or set the `AIRBYTE_CLOUD_CONFIG_API_URL` environment
 20variable.
 21
 22```python
 23from airbyte import cloud
 24
 25workspace = cloud.CloudWorkspace(
 26    workspace_id="...",
 27    client_id="...",
 28    client_secret="...",
 29    api_root="https://airbyte.example.com/api/public/v1",
 30    config_api_root="https://airbyte.example.com/api/v1",
 31)
 32
 33connection = workspace.get_connection(connection_id="...")
 34raw_catalog = connection.dump_raw_catalog()
 35```
 36
 37## Examples
 38
 39### Basic Sync Example:
 40
 41```python
 42import airbyte as ab
 43from airbyte import cloud
 44
 45# Initialize an Airbyte Cloud workspace object
 46workspace = cloud.CloudWorkspace(
 47    workspace_id="123",
 48    api_key=ab.get_secret("AIRBYTE_CLOUD_API_KEY"),
 49)
 50
 51# Run a sync job on Airbyte Cloud
 52connection = workspace.get_connection(connection_id="456")
 53sync_result = connection.run_sync()
 54print(sync_result.get_job_status())
 55```
 56
 57### Example Read From Cloud Destination:
 58
 59If your destination is supported, you can read records directly from the
 60`SyncResult` object. Currently this is supported in Snowflake and BigQuery only.
 61
 62
 63```python
 64# Assuming we've already created a `connection` object...
 65
 66# Get the latest job result and print the stream names
 67sync_result = connection.get_sync_result()
 68print(sync_result.stream_names)
 69
 70# Get a dataset from the sync result
 71dataset: CachedDataset = sync_result.get_dataset("users")
 72
 73# Get a SQLAlchemy table to use in SQL queries...
 74users_table = dataset.to_sql_table()
 75print(f"Table name: {users_table.name}")
 76
 77# Or iterate over the dataset directly
 78for record in dataset:
 79    print(record)
 80```
 81"""
 82
 83from __future__ import annotations
 84
 85from typing import TYPE_CHECKING
 86
 87from airbyte.cloud.client import CloudClient
 88from airbyte.cloud.client_config import CloudClientConfig
 89from airbyte.cloud.connections import CloudConnection
 90from airbyte.cloud.models import (
 91    CloudDefaultContextInfo,
 92    CloudWorkspaceInfo,
 93    JobStatusEnum,
 94    JobTypeEnum,
 95    WorkspacePrivilegeScope,
 96)
 97from airbyte.cloud.organizations import CloudOrganization
 98from airbyte.cloud.sync_results import SyncResult
 99from airbyte.cloud.workspaces import CloudWorkspace
100
101
102# Submodules imported here for documentation reasons: https://github.com/mitmproxy/pdoc/issues/757
103if TYPE_CHECKING:
104    # ruff: noqa: TC004
105    from airbyte.cloud import (
106        client,
107        client_config,
108        connections,
109        constants,
110        organizations,
111        sync_results,
112        workspaces,
113    )
114
115
116__all__ = [
117    # Submodules
118    "workspaces",
119    "client",
120    "organizations",
121    "connections",
122    "constants",
123    "client_config",
124    "sync_results",
125    # Classes
126    "CloudClient",
127    "CloudOrganization",
128    "CloudWorkspace",
129    "CloudConnection",
130    "CloudClientConfig",
131    "CloudDefaultContextInfo",
132    "CloudWorkspaceInfo",
133    "SyncResult",
134    # Enums
135    "JobStatusEnum",
136    "JobTypeEnum",
137    "WorkspacePrivilegeScope",
138]

class CloudOrganization: View Source

 22class CloudOrganization:
 23    """Information about an organization in Airbyte Cloud.
 24
 25    This class provides lazy loading of organization attributes including billing status.
 26    It is typically created via `CloudWorkspace.get_organization()`.
 27    """
 28
 29    def __init__(
 30        self,
 31        organization_id: str,
 32        organization_name: str | None = None,
 33        email: str | None = None,
 34        *,
 35        client_id: str | SecretString | None = None,
 36        client_secret: str | SecretString | None = None,
 37        bearer_token: str | SecretString | None = None,
 38        public_api_root: str | None = None,
 39        config_api_root: str | None = None,
 40    ) -> None:
 41        """Initialize a `CloudOrganization`."""
 42        self.organization_id = organization_id
 43        """The organization ID."""
 44
 45        self._organization_name = organization_name
 46        """Display name of the organization."""
 47
 48        self._email = email
 49        """Email associated with the organization."""
 50
 51        self._credentials = _AirbyteCredentials(
 52            client_id=SecretString(client_id) if client_id else None,
 53            client_secret=SecretString(client_secret) if client_secret else None,
 54            bearer_token=SecretString(bearer_token) if bearer_token else None,
 55            public_api_root=public_api_root or api_util.CLOUD_API_ROOT,
 56            config_api_root=config_api_root,
 57            organization_id=organization_id,
 58        )
 59        self._organization_info: dict[str, Any] | None = None
 60        self._organization_info_fetch_failed: bool = False
 61
 62    def _fetch_organization_info(self, *, force_refresh: bool = False) -> dict[str, Any]:
 63        """Fetch and cache organization info including billing status."""
 64        if force_refresh:
 65            self._organization_info_fetch_failed = False
 66
 67        if self._organization_info_fetch_failed and self._organization_info is None:
 68            return {}
 69
 70        if not force_refresh and self._organization_info is not None:
 71            return self._organization_info
 72
 73        try:
 74            self._organization_info = api_util.get_organization_info(
 75                organization_id=self.organization_id,
 76                api_root=self._credentials.public_api_root,
 77                config_api_root=self._credentials.config_api_root,
 78                client_id=self._credentials.client_id,
 79                client_secret=self._credentials.client_secret,
 80                bearer_token=self._credentials.bearer_token,
 81            )
 82        except Exception as ex:
 83            logger.debug("Failed to fetch organization info.", exc_info=ex)
 84            if self._organization_info is None:
 85                self._organization_info_fetch_failed = True
 86            return self._organization_info or {}
 87        else:
 88            return self._organization_info
 89
 90    @property
 91    def organization_name(self) -> str | None:
 92        """Display name of the organization."""
 93        if self._organization_name is not None:
 94            return self._organization_name
 95        info = self._fetch_organization_info()
 96        return info.get("organizationName")
 97
 98    @property
 99    def email(self) -> str | None:
100        """Email associated with the organization."""
101        if self._email is not None:
102            return self._email
103        info = self._fetch_organization_info()
104        return info.get("email")
105
106    def get_billing_status(self) -> CloudOrganizationBillingInfo:
107        """Fetch billing status for the organization or raise on failure."""
108        try:
109            info = api_util.get_organization_info(
110                organization_id=self.organization_id,
111                api_root=self._credentials.public_api_root,
112                config_api_root=self._credentials.config_api_root,
113                client_id=self._credentials.client_id,
114                client_secret=self._credentials.client_secret,
115                bearer_token=self._credentials.bearer_token,
116            )
117        except (requests.RequestException, ValueError) as ex:
118            raise AirbyteError(
119                message="Failed to retrieve organization billing information.",
120                context={"organization_id": self.organization_id},
121            ) from ex
122        billing = info.get("billing")
123        if not isinstance(billing, dict):
124            raise AirbyteError(
125                message="Organization info did not include billing details.",
126                context={"organization_id": self.organization_id},
127            )
128        payment_status = billing.get("paymentStatus")
129        subscription_status = billing.get("subscriptionStatus")
130        return CloudOrganizationBillingInfo(
131            payment_status=payment_status if isinstance(payment_status, str) else None,
132            subscription_status=(
133                subscription_status if isinstance(subscription_status, str) else None
134            ),
135            is_account_locked=api_util.is_account_locked(payment_status, subscription_status),
136        )
137
138    @property
139    def payment_status(self) -> str | None:
140        """Payment status of the organization."""
141        info = self._fetch_organization_info()
142        return (info.get("billing") or {}).get("paymentStatus")
143
144    @property
145    def subscription_status(self) -> str | None:
146        """Subscription status of the organization."""
147        info = self._fetch_organization_info()
148        return (info.get("billing") or {}).get("subscriptionStatus")
149
150    @property
151    def is_account_locked(self) -> bool:
152        """Whether the account is locked due to billing issues."""
153        return api_util.is_account_locked(self.payment_status, self.subscription_status)

Information about an organization in Airbyte Cloud.

This class provides lazy loading of organization attributes including billing status. It is typically created via CloudWorkspace.get_organization().

29    def __init__(
30        self,
31        organization_id: str,
32        organization_name: str | None = None,
33        email: str | None = None,
34        *,
35        client_id: str | SecretString | None = None,
36        client_secret: str | SecretString | None = None,
37        bearer_token: str | SecretString | None = None,
38        public_api_root: str | None = None,
39        config_api_root: str | None = None,
40    ) -> None:
41        """Initialize a `CloudOrganization`."""
42        self.organization_id = organization_id
43        """The organization ID."""
44
45        self._organization_name = organization_name
46        """Display name of the organization."""
47
48        self._email = email
49        """Email associated with the organization."""
50
51        self._credentials = _AirbyteCredentials(
52            client_id=SecretString(client_id) if client_id else None,
53            client_secret=SecretString(client_secret) if client_secret else None,
54            bearer_token=SecretString(bearer_token) if bearer_token else None,
55            public_api_root=public_api_root or api_util.CLOUD_API_ROOT,
56            config_api_root=config_api_root,
57            organization_id=organization_id,
58        )
59        self._organization_info: dict[str, Any] | None = None
60        self._organization_info_fetch_failed: bool = False

Initialize a CloudOrganization.

organization_id

The organization ID.

organization_name: str | None View Source

90    @property
91    def organization_name(self) -> str | None:
92        """Display name of the organization."""
93        if self._organization_name is not None:
94            return self._organization_name
95        info = self._fetch_organization_info()
96        return info.get("organizationName")

Display name of the organization.

email: str | None View Source

 98    @property
 99    def email(self) -> str | None:
100        """Email associated with the organization."""
101        if self._email is not None:
102            return self._email
103        info = self._fetch_organization_info()
104        return info.get("email")

Email associated with the organization.

def get_billing_status(self) -> airbyte.cloud.models.CloudOrganizationBillingInfo: View Source

106    def get_billing_status(self) -> CloudOrganizationBillingInfo:
107        """Fetch billing status for the organization or raise on failure."""
108        try:
109            info = api_util.get_organization_info(
110                organization_id=self.organization_id,
111                api_root=self._credentials.public_api_root,
112                config_api_root=self._credentials.config_api_root,
113                client_id=self._credentials.client_id,
114                client_secret=self._credentials.client_secret,
115                bearer_token=self._credentials.bearer_token,
116            )
117        except (requests.RequestException, ValueError) as ex:
118            raise AirbyteError(
119                message="Failed to retrieve organization billing information.",
120                context={"organization_id": self.organization_id},
121            ) from ex
122        billing = info.get("billing")
123        if not isinstance(billing, dict):
124            raise AirbyteError(
125                message="Organization info did not include billing details.",
126                context={"organization_id": self.organization_id},
127            )
128        payment_status = billing.get("paymentStatus")
129        subscription_status = billing.get("subscriptionStatus")
130        return CloudOrganizationBillingInfo(
131            payment_status=payment_status if isinstance(payment_status, str) else None,
132            subscription_status=(
133                subscription_status if isinstance(subscription_status, str) else None
134            ),
135            is_account_locked=api_util.is_account_locked(payment_status, subscription_status),
136        )

Fetch billing status for the organization or raise on failure.

payment_status: str | None View Source

138    @property
139    def payment_status(self) -> str | None:
140        """Payment status of the organization."""
141        info = self._fetch_organization_info()
142        return (info.get("billing") or {}).get("paymentStatus")

Payment status of the organization.

subscription_status: str | None View Source

144    @property
145    def subscription_status(self) -> str | None:
146        """Subscription status of the organization."""
147        info = self._fetch_organization_info()
148        return (info.get("billing") or {}).get("subscriptionStatus")

Subscription status of the organization.

is_account_locked: bool View Source

150    @property
151    def is_account_locked(self) -> bool:
152        """Whether the account is locked due to billing issues."""
153        return api_util.is_account_locked(self.payment_status, self.subscription_status)

Whether the account is locked due to billing issues.

@dataclass(init=False, kw_only=True)

class CloudWorkspace: View Source

 111@dataclass(init=False, kw_only=True)  # noqa: PLR0904  # Core cloud API facade.
 112class CloudWorkspace:
 113    """A remote workspace on the Airbyte Cloud.
 114
 115    By overriding `api_root`, you can use this class to interact with self-managed Airbyte
 116    instances, both OSS and Enterprise.
 117
 118    Two authentication methods are supported (mutually exclusive):
 119    1. OAuth2 client credentials (client_id + client_secret)
 120    2. Bearer token authentication
 121
 122    Example with client credentials:
 123        ```python
 124        workspace = CloudWorkspace(
 125            workspace_id="...",
 126            client_id="...",
 127            client_secret="...",
 128        )
 129        ```
 130
 131    Example with bearer token:
 132        ```python
 133        workspace = CloudWorkspace(
 134            workspace_id="...",
 135            bearer_token="...",
 136        )
 137        ```
 138    """
 139
 140    workspace_id: str
 141    client_id: SecretString | None
 142    client_secret: SecretString | None
 143    api_root: str
 144    config_api_root: str | None
 145    """The Config API root URL."""
 146    bearer_token: SecretString | None
 147
 148    # Internal credentials objects (set in __init__, excluded from repr)
 149    _credentials: _AirbyteCredentials = field(init=False, repr=False)
 150    _client_config: CloudClientConfig = field(init=False, repr=False)
 151
 152    def __init__(
 153        self,
 154        *,
 155        workspace_id: str | None = None,
 156        client_id: str | SecretString | None = None,
 157        client_secret: str | SecretString | None = None,
 158        api_root: str | None = None,
 159        config_api_root: str | None = None,
 160        bearer_token: str | SecretString | None = None,
 161    ) -> None:
 162        """Validate and initialize credentials."""
 163        env_vars = not (client_id or client_secret or bearer_token)
 164        credentials = _AirbyteCredentials.from_auth(
 165            workspace_id=workspace_id,
 166            client_id=client_id,
 167            client_secret=client_secret,
 168            bearer_token=bearer_token,
 169            public_api_root=api_root,
 170            config_api_root=config_api_root,
 171            env_vars=env_vars,
 172        )
 173        if not credentials.workspace_id:
 174            raise exc.PyAirbyteInputError(
 175                message="Workspace ID is required.",
 176                guidance=(
 177                    "Provide a workspace ID, or call `get_default_cloud_context` to discover "
 178                    "available workspaces."
 179                ),
 180            )
 181
 182        self._credentials = credentials
 183        self.workspace_id = credentials.workspace_id or ""
 184        self.client_id = credentials.client_id
 185        self.client_secret = credentials.client_secret
 186        self.bearer_token = credentials.bearer_token
 187        self.api_root = credentials.public_api_root
 188        self.config_api_root = credentials.config_api_root
 189
 190        # Create internal CloudClientConfig object (validates mutual exclusivity)
 191        self._client_config = CloudClientConfig(
 192            client_id=self.client_id,
 193            client_secret=self.client_secret,
 194            bearer_token=self.bearer_token,
 195            api_root=self.api_root,
 196            config_api_root=self.config_api_root,
 197        )
 198
 199    @classmethod
 200    def from_env(
 201        cls,
 202        workspace_id: str | None = None,
 203        *,
 204        api_root: str | None = None,
 205        config_api_root: str | None = None,
 206    ) -> CloudWorkspace:
 207        """Create a CloudWorkspace using credentials from environment variables.
 208
 209        This factory method resolves credentials from environment variables,
 210        providing a convenient way to create a workspace without explicitly
 211        passing credentials.
 212
 213        Two authentication methods are supported (mutually exclusive):
 214        1. Bearer token (checked first)
 215        2. OAuth2 client credentials (fallback)
 216
 217        Environment variables used:
 218            - `AIRBYTE_CLOUD_BEARER_TOKEN`: Bearer token (alternative to client credentials).
 219            - `AIRBYTE_CLOUD_CLIENT_ID`: OAuth client ID (for client credentials flow).
 220            - `AIRBYTE_CLOUD_CLIENT_SECRET`: OAuth client secret (for client credentials flow).
 221            - `AIRBYTE_CLOUD_WORKSPACE_ID`: The workspace ID (if not passed as argument).
 222            - `AIRBYTE_CLOUD_API_URL`: Optional. The API root URL (defaults to Airbyte Cloud).
 223            - `AIRBYTE_CLOUD_CONFIG_API_URL`: Optional. The Config API root URL.
 224
 225        Args:
 226            workspace_id: The workspace ID. If not provided, will be resolved from
 227                the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
 228            api_root: The API root URL. If not provided, will be resolved from
 229                the `AIRBYTE_CLOUD_API_URL` environment variable, or default to
 230                the Airbyte Cloud API.
 231            config_api_root: The Config API root URL. If not provided, will be resolved
 232                from the `AIRBYTE_CLOUD_CONFIG_API_URL` environment variable.
 233
 234        Returns:
 235            A CloudWorkspace instance configured with credentials from the environment.
 236
 237        Raises:
 238            PyAirbyteInputError: If required credentials are not found in
 239                the environment or are incomplete.
 240
 241        Example:
 242            ```python
 243            # With workspace_id from environment
 244            workspace = CloudWorkspace.from_env()
 245
 246            # With explicit workspace_id
 247            workspace = CloudWorkspace.from_env(workspace_id="your-workspace-id")
 248            ```
 249        """
 250        return cls(
 251            workspace_id=workspace_id,
 252            api_root=api_root,
 253            config_api_root=config_api_root,
 254        )
 255
 256    @property
 257    def workspace_url(self) -> str | None:
 258        """The web URL of the workspace."""
 259        return f"{get_web_url_root(self.api_root)}/workspaces/{self.workspace_id}"
 260
 261    @cached_property
 262    def _organization_info(self) -> dict[str, Any]:
 263        """Fetch and cache organization info for this workspace.
 264
 265        Uses the Config API endpoint for an efficient O(1) lookup.
 266        This is an internal method; use get_organization() for public access.
 267        """
 268        return api_util.get_workspace_organization_info(
 269            workspace_id=self.workspace_id,
 270            api_root=self.api_root,
 271            config_api_root=self.config_api_root,
 272            client_id=self.client_id,
 273            client_secret=self.client_secret,
 274            bearer_token=self.bearer_token,
 275        )
 276
 277    @overload
 278    def get_organization(self) -> CloudOrganization: ...
 279
 280    @overload
 281    def get_organization(
 282        self,
 283        *,
 284        raise_on_error: Literal[True],
 285    ) -> CloudOrganization: ...
 286
 287    @overload
 288    def get_organization(
 289        self,
 290        *,
 291        raise_on_error: Literal[False],
 292    ) -> CloudOrganization | None: ...
 293
 294    def get_organization(
 295        self,
 296        *,
 297        raise_on_error: bool = True,
 298    ) -> CloudOrganization | None:
 299        """Get the organization this workspace belongs to.
 300
 301        Fetching organization info requires ORGANIZATION_READER permissions on the organization,
 302        which may not be available with workspace-scoped credentials.
 303
 304        Args:
 305            raise_on_error: If True (default), raises AirbyteError on permission or API errors.
 306                If False, returns None instead of raising.
 307
 308        Returns:
 309            CloudOrganization object with organization_id and organization_name,
 310            or None if raise_on_error=False and an error occurred.
 311
 312        Raises:
 313            AirbyteError: If raise_on_error=True and the organization info cannot be fetched
 314                (e.g., due to insufficient permissions or missing data).
 315        """
 316        try:
 317            info = self._organization_info
 318        except (AirbyteError, NotImplementedError):
 319            if raise_on_error:
 320                raise
 321            return None
 322
 323        organization_id = info.get("organizationId")
 324        organization_name = info.get("organizationName")
 325
 326        # Validate that both organization_id and organization_name are non-null and non-empty
 327        if not organization_id or not organization_name:
 328            if raise_on_error:
 329                raise AirbyteError(
 330                    message="Organization info is incomplete.",
 331                    context={
 332                        "organization_id": organization_id,
 333                        "organization_name": organization_name,
 334                    },
 335                )
 336            return None
 337
 338        organization_credentials = self._credentials.with_organization_id(organization_id)
 339        return CloudOrganization(
 340            organization_id=organization_id,
 341            organization_name=organization_name,
 342            client_id=organization_credentials.client_id,
 343            client_secret=organization_credentials.client_secret,
 344            bearer_token=organization_credentials.bearer_token,
 345            public_api_root=organization_credentials.public_api_root,
 346            config_api_root=organization_credentials.config_api_root,
 347        )
 348
 349    # Test connection and creds
 350
 351    def connect(self) -> None:
 352        """Check that the workspace is reachable and raise an exception otherwise.
 353
 354        Note: It is not necessary to call this method before calling other operations. It
 355              serves primarily as a simple check to ensure that the workspace is reachable
 356              and credentials are correct.
 357        """
 358        _ = api_util.get_workspace(
 359            api_root=self.api_root,
 360            workspace_id=self.workspace_id,
 361            client_id=self.client_id,
 362            client_secret=self.client_secret,
 363            bearer_token=self.bearer_token,
 364        )
 365        print(f"Successfully connected to workspace: {self.workspace_url}")
 366
 367    # Get sources, destinations, and connections
 368
 369    def get_connection(
 370        self,
 371        connection_id: str,
 372    ) -> CloudConnection:
 373        """Get a connection by ID.
 374
 375        This method does not fetch data from the API. It returns a `CloudConnection` object,
 376        which will be loaded lazily as needed.
 377        """
 378        return CloudConnection(
 379            workspace=self,
 380            connection_id=connection_id,
 381        )
 382
 383    def get_source(
 384        self,
 385        source_id: str,
 386    ) -> CloudSource:
 387        """Get a source by ID.
 388
 389        This method does not fetch data from the API. It returns a `CloudSource` object,
 390        which will be loaded lazily as needed.
 391        """
 392        return CloudSource(
 393            workspace=self,
 394            connector_id=source_id,
 395        )
 396
 397    def get_destination(
 398        self,
 399        destination_id: str,
 400    ) -> CloudDestination:
 401        """Get a destination by ID.
 402
 403        This method does not fetch data from the API. It returns a `CloudDestination` object,
 404        which will be loaded lazily as needed.
 405        """
 406        return CloudDestination(
 407            workspace=self,
 408            connector_id=destination_id,
 409        )
 410
 411    def check_connector_setup(
 412        self,
 413        connector_type: Literal["source", "destination"],
 414        connector_id: str,
 415    ) -> CheckResult:
 416        """Run one connection check on a connector that belongs to this workspace.
 417
 418        Confirms a person has finished a deferred-credential setup in Airbyte Cloud. The
 419        connector's workspace is verified first so a check can never be run against a connector
 420        outside this workspace.
 421        """
 422        connector: CloudSource | CloudDestination
 423        if connector_type == "source":
 424            owner_id = api_util.get_source(
 425                source_id=connector_id,
 426                api_root=self.api_root,
 427                client_id=self.client_id,
 428                client_secret=self.client_secret,
 429                bearer_token=self.bearer_token,
 430            ).workspace_id
 431            connector = self.get_source(connector_id)
 432        else:
 433            owner_id = api_util.get_destination(
 434                destination_id=connector_id,
 435                api_root=self.api_root,
 436                client_id=self.client_id,
 437                client_secret=self.client_secret,
 438                bearer_token=self.bearer_token,
 439            ).workspace_id
 440            connector = self.get_destination(connector_id)
 441        if owner_id != self.workspace_id:
 442            raise exc.AirbyteMissingResourceError(
 443                resource_type=connector_type,
 444                resource_name_or_id=connector_id,
 445                context={"workspace_id": self.workspace_id},
 446            )
 447        try:
 448            return connector.check(raise_on_error=False)
 449        except AirbyteError as ex:
 450            status_code = (ex.context or {}).get("status_code")
 451            if status_code == HTTPStatus.UNPROCESSABLE_ENTITY:
 452                return CheckResult(success=False)
 453            raise AirbyteError(
 454                message="Cloud could not check the connector setup.",
 455                context={"status_code": status_code},
 456            ) from None
 457
 458    # Deploy sources and destinations
 459
 460    def deploy_source(
 461        self,
 462        name: str,
 463        source: Source | dict[str, Any],
 464        *,
 465        unique: bool = True,
 466        random_name_suffix: bool = False,
 467        definition_id: str | None = None,
 468        defer_credentials: bool = False,
 469    ) -> CloudSource:
 470        """Deploy a source to the workspace.
 471
 472        Returns the newly deployed source.
 473
 474        Args:
 475            name: The name to use when deploying.
 476            source: The source object to deploy, or (with `defer_credentials=True`) a
 477                dictionary of non-secret configuration values.
 478            unique: Whether to require a unique name. If `True`, duplicate names
 479                are not allowed. Defaults to `True`.
 480            random_name_suffix: Whether to append a random suffix to the name.
 481            definition_id: The source definition ID. Required with `defer_credentials=True`.
 482            defer_credentials: Save a draft with partial configuration. A person completes
 483                credentials and other missing settings at the returned source's `connector_url`.
 484                A successful connection check promotes the draft. Raises
 485                `AirbyteDeferredSetupError` if Cloud does not acknowledge draft mode.
 486        """
 487        if defer_credentials:
 488            return CloudSource(
 489                workspace=self,
 490                connector_id=self._deploy_deferred(
 491                    connector_type="source",
 492                    name=name,
 493                    config=source,
 494                    definition_id=definition_id,
 495                    unique=unique,
 496                    random_name_suffix=random_name_suffix,
 497                ),
 498            )
 499        if isinstance(source, dict):
 500            raise exc.PyAirbyteInputError(
 501                message="`source` must be a `Source` object unless `defer_credentials=True`.",
 502            )
 503
 504        source_config_dict = source._hydrated_config.copy()  # noqa: SLF001 (non-public API)
 505        source_config_dict["sourceType"] = source.name.replace("source-", "")
 506
 507        if random_name_suffix:
 508            name += f" (ID: {text_util.generate_random_suffix()})"
 509
 510        if unique:
 511            existing = self.list_sources(name=name)
 512            if existing:
 513                raise exc.AirbyteDuplicateResourcesError(
 514                    resource_type="source",
 515                    resource_name=name,
 516                )
 517
 518        deployed_source = api_util.create_source(
 519            name=name,
 520            api_root=self.api_root,
 521            workspace_id=self.workspace_id,
 522            config=source_config_dict,
 523            definition_id=definition_id,
 524            client_id=self.client_id,
 525            client_secret=self.client_secret,
 526            bearer_token=self.bearer_token,
 527        )
 528        return CloudSource(
 529            workspace=self,
 530            connector_id=deployed_source.source_id,
 531        )
 532
 533    def deploy_destination(
 534        self,
 535        name: str,
 536        destination: Destination | dict[str, Any],
 537        *,
 538        unique: bool = True,
 539        random_name_suffix: bool = False,
 540        definition_id: str | None = None,
 541        defer_credentials: bool = False,
 542    ) -> CloudDestination:
 543        """Deploy a destination to the workspace.
 544
 545        Returns the newly deployed destination ID.
 546
 547        Args:
 548            name: The name to use when deploying.
 549            destination: The destination to deploy. Can be a local Airbyte `Destination` object or a
 550                dictionary of configuration values.
 551            unique: Whether to require a unique name. If `True`, duplicate names
 552                are not allowed. Defaults to `True`.
 553            random_name_suffix: Whether to append a random suffix to the name.
 554            definition_id: The destination definition ID. Required with `defer_credentials=True`;
 555                otherwise the type is inferred from `destinationType`.
 556            defer_credentials: Create the destination without its credentials. See
 557                `deploy_source`.
 558        """
 559        if defer_credentials:
 560            return CloudDestination(
 561                workspace=self,
 562                connector_id=self._deploy_deferred(
 563                    connector_type="destination",
 564                    name=name,
 565                    config=destination,
 566                    definition_id=definition_id,
 567                    unique=unique,
 568                    random_name_suffix=random_name_suffix,
 569                ),
 570            )
 571
 572        if isinstance(destination, Destination):
 573            destination_conf_dict = destination._hydrated_config.copy()  # noqa: SLF001 (non-public API)
 574            destination_conf_dict["destinationType"] = destination.name.replace("destination-", "")
 575            # raise ValueError(destination_conf_dict)
 576        else:
 577            destination_conf_dict = destination.copy()
 578            if "destinationType" not in destination_conf_dict:
 579                raise exc.PyAirbyteInputError(
 580                    message="Missing `destinationType` in configuration dictionary.",
 581                )
 582
 583        if random_name_suffix:
 584            name += f" (ID: {text_util.generate_random_suffix()})"
 585
 586        if unique:
 587            existing = self.list_destinations(name=name)
 588            if existing:
 589                raise exc.AirbyteDuplicateResourcesError(
 590                    resource_type="destination",
 591                    resource_name=name,
 592                )
 593
 594        deployed_destination = api_util.create_destination(
 595            name=name,
 596            api_root=self.api_root,
 597            workspace_id=self.workspace_id,
 598            config=destination_conf_dict,  # Wants a dataclass but accepts dict
 599            client_id=self.client_id,
 600            client_secret=self.client_secret,
 601            bearer_token=self.bearer_token,
 602        )
 603        return CloudDestination(
 604            workspace=self,
 605            connector_id=deployed_destination.destination_id,
 606        )
 607
 608    def _deploy_deferred(
 609        self,
 610        *,
 611        connector_type: Literal["source", "destination"],
 612        name: str,
 613        config: object,
 614        definition_id: str | None,
 615        unique: bool,
 616        random_name_suffix: bool,
 617    ) -> str:
 618        """Create a connector with deferred credentials on the Config API and return its ID."""
 619        config_dict, definition_id = _deferred_credentials_config(
 620            config, definition_id=definition_id
 621        )
 622
 623        if random_name_suffix:
 624            name += f" (ID: {text_util.generate_random_suffix()})"
 625
 626        if unique:
 627            existing = (
 628                self.list_sources(name=name)
 629                if connector_type == "source"
 630                else self.list_destinations(name=name)
 631            )
 632            if existing:
 633                raise exc.AirbyteDuplicateResourcesError(
 634                    resource_type=connector_type,
 635                    resource_name=name,
 636                )
 637
 638        return api_util.create_connector_deferred(
 639            connector_type=connector_type,
 640            name=name,
 641            workspace_id=self.workspace_id,
 642            definition_id=definition_id,
 643            config=config_dict,
 644            api_root=self.api_root,
 645            config_api_root=self.config_api_root,
 646            client_id=self.client_id,
 647            client_secret=self.client_secret,
 648            bearer_token=self.bearer_token,
 649        )
 650
 651    def permanently_delete_source(
 652        self,
 653        source: str | CloudSource,
 654        *,
 655        safe_mode: bool = True,
 656    ) -> None:
 657        """Delete a source from the workspace.
 658
 659        You can pass either the source ID `str` or a deployed `Source` object.
 660
 661        Args:
 662            source: The source ID or CloudSource object to delete
 663            safe_mode: If True, requires the source name to contain "delete-me" or "deleteme"
 664                (case insensitive) to prevent accidental deletion. Defaults to True.
 665        """
 666        if not isinstance(source, (str, CloudSource)):
 667            raise exc.PyAirbyteInputError(
 668                message="Invalid source type.",
 669                input_value=type(source).__name__,
 670            )
 671
 672        api_util.delete_source(
 673            source_id=source.connector_id if isinstance(source, CloudSource) else source,
 674            source_name=source.name if isinstance(source, CloudSource) else None,
 675            api_root=self.api_root,
 676            client_id=self.client_id,
 677            client_secret=self.client_secret,
 678            bearer_token=self.bearer_token,
 679            safe_mode=safe_mode,
 680        )
 681
 682    # Deploy and delete destinations
 683
 684    def permanently_delete_destination(
 685        self,
 686        destination: str | CloudDestination,
 687        *,
 688        safe_mode: bool = True,
 689    ) -> None:
 690        """Delete a deployed destination from the workspace.
 691
 692        You can pass either the `Cache` class or the deployed destination ID as a `str`.
 693
 694        Args:
 695            destination: The destination ID or CloudDestination object to delete
 696            safe_mode: If True, requires the destination name to contain "delete-me" or "deleteme"
 697                (case insensitive) to prevent accidental deletion. Defaults to True.
 698        """
 699        if not isinstance(destination, (str, CloudDestination)):
 700            raise exc.PyAirbyteInputError(
 701                message="Invalid destination type.",
 702                input_value=type(destination).__name__,
 703            )
 704
 705        api_util.delete_destination(
 706            destination_id=(
 707                destination if isinstance(destination, str) else destination.destination_id
 708            ),
 709            destination_name=(
 710                destination.name if isinstance(destination, CloudDestination) else None
 711            ),
 712            api_root=self.api_root,
 713            client_id=self.client_id,
 714            client_secret=self.client_secret,
 715            bearer_token=self.bearer_token,
 716            safe_mode=safe_mode,
 717        )
 718
 719    # Deploy and delete connections
 720
 721    def deploy_connection(
 722        self,
 723        connection_name: str,
 724        *,
 725        source: CloudSource | str,
 726        selected_streams: list[str],
 727        destination: CloudDestination | str,
 728        table_prefix: str | None = None,
 729    ) -> CloudConnection:
 730        """Create a new connection between an already deployed source and destination.
 731
 732        Returns the newly deployed connection object.
 733
 734        Args:
 735            connection_name: The name of the connection.
 736            source: The deployed source. You can pass a source ID or a CloudSource object.
 737            destination: The deployed destination. You can pass a destination ID or a
 738                CloudDestination object.
 739            table_prefix: Optional. The table prefix to use when syncing to the destination.
 740            selected_streams: The selected stream names to sync within the connection.
 741        """
 742        if not selected_streams:
 743            raise exc.PyAirbyteInputError(
 744                guidance="You must provide `selected_streams` when creating a connection."
 745            )
 746
 747        source_id: str = source if isinstance(source, str) else source.connector_id
 748        destination_id: str = (
 749            destination if isinstance(destination, str) else destination.connector_id
 750        )
 751
 752        deployed_connection = api_util.create_connection(
 753            name=connection_name,
 754            source_id=source_id,
 755            destination_id=destination_id,
 756            api_root=self.api_root,
 757            workspace_id=self.workspace_id,
 758            selected_stream_names=selected_streams,
 759            prefix=table_prefix or "",
 760            client_id=self.client_id,
 761            client_secret=self.client_secret,
 762            bearer_token=self.bearer_token,
 763        )
 764
 765        return CloudConnection(
 766            workspace=self,
 767            connection_id=deployed_connection.connection_id,
 768            source=deployed_connection.source_id,
 769            destination=deployed_connection.destination_id,
 770        )
 771
 772    def permanently_delete_connection(
 773        self,
 774        connection: str | CloudConnection,
 775        *,
 776        cascade_delete_source: bool = False,
 777        cascade_delete_destination: bool = False,
 778        safe_mode: bool = True,
 779    ) -> None:
 780        """Delete a deployed connection from the workspace.
 781
 782        Args:
 783            connection: The connection ID or CloudConnection object to delete
 784            cascade_delete_source: If True, also delete the source after deleting the connection
 785            cascade_delete_destination: If True, also delete the destination after deleting
 786                the connection
 787            safe_mode: If True, requires the connection name to contain "delete-me" or "deleteme"
 788                (case insensitive) to prevent accidental deletion. Defaults to True. Also applies
 789                to cascade deletes.
 790        """
 791        if connection is None:
 792            raise ValueError("No connection ID provided.")
 793
 794        if isinstance(connection, str):
 795            connection = CloudConnection(
 796                workspace=self,
 797                connection_id=connection,
 798            )
 799
 800        api_util.delete_connection(
 801            connection_id=connection.connection_id,
 802            connection_name=connection.name,
 803            api_root=self.api_root,
 804            workspace_id=self.workspace_id,
 805            client_id=self.client_id,
 806            client_secret=self.client_secret,
 807            bearer_token=self.bearer_token,
 808            safe_mode=safe_mode,
 809        )
 810
 811        if cascade_delete_source:
 812            self.permanently_delete_source(
 813                source=connection.source_id,
 814                safe_mode=safe_mode,
 815            )
 816        if cascade_delete_destination:
 817            self.permanently_delete_destination(
 818                destination=connection.destination_id,
 819                safe_mode=safe_mode,
 820            )
 821
 822    # List workspaces, sources, destinations, and connections
 823
 824    def list_workspaces(
 825        self,
 826        name: str | None = None,
 827        *,
 828        name_filter: Callable | None = None,
 829        limit: int | None = None,
 830    ) -> list[CloudWorkspaceInfo]:
 831        """List workspaces available to the current credentials, with an optional limit."""
 832        return [
 833            CloudWorkspaceInfo.from_api_response(workspace)
 834            for workspace in api_util.list_workspaces(
 835                workspace_id="",
 836                api_root=self.api_root,
 837                name=name,
 838                name_filter=name_filter,
 839                client_id=self.client_id,
 840                client_secret=self.client_secret,
 841                bearer_token=self.bearer_token,
 842                limit=limit,
 843            )
 844        ]
 845
 846    def rename(
 847        self,
 848        name: str,
 849    ) -> CloudWorkspace:
 850        """Rename this workspace."""
 851        api_util.rename_workspace(
 852            workspace_id=self.workspace_id,
 853            name=name,
 854            api_root=self.api_root,
 855            client_id=self.client_id,
 856            client_secret=self.client_secret,
 857            bearer_token=self.bearer_token,
 858        )
 859        return self
 860
 861    def permanently_delete(
 862        self,
 863        *,
 864        workspace_name: str | None = None,
 865        safe_mode: bool = True,
 866    ) -> None:
 867        """Permanently delete this workspace if it has no connections.
 868
 869        When `safe_mode` is enabled, the workspace name must contain `delete-me`
 870        or `deleteme`. This also checks for existing connections before deleting
 871        and raises `AirbyteWorkspaceNotEmptyError` if the workspace is not empty.
 872        """
 873        api_util.permanently_delete_workspace(
 874            workspace_id=self.workspace_id,
 875            workspace_name=workspace_name,
 876            api_root=self.api_root,
 877            client_id=self.client_id,
 878            client_secret=self.client_secret,
 879            bearer_token=self.bearer_token,
 880            safe_mode=safe_mode,
 881        )
 882
 883    def list_connections(
 884        self,
 885        name: str | None = None,
 886        *,
 887        name_filter: Callable | None = None,
 888        limit: int | None = None,
 889    ) -> list[CloudConnection]:
 890        """List connections by name in the workspace, with an optional limit."""
 891        connections = api_util.list_connections(
 892            api_root=self.api_root,
 893            workspace_id=self.workspace_id,
 894            name=name,
 895            name_filter=name_filter,
 896            limit=limit,
 897            client_id=self.client_id,
 898            client_secret=self.client_secret,
 899            bearer_token=self.bearer_token,
 900        )
 901        return [
 902            CloudConnection._from_connection_response(  # noqa: SLF001 (non-public API)
 903                workspace=self,
 904                connection_response=connection,
 905            )
 906            for connection in connections
 907        ]
 908
 909    def list_sources(
 910        self,
 911        name: str | None = None,
 912        *,
 913        name_filter: Callable | None = None,
 914        limit: int | None = None,
 915    ) -> list[CloudSource]:
 916        """List all sources in the workspace, with an optional limit."""
 917        sources = api_util.list_sources(
 918            api_root=self.api_root,
 919            workspace_id=self.workspace_id,
 920            name=name,
 921            name_filter=name_filter,
 922            limit=limit,
 923            client_id=self.client_id,
 924            client_secret=self.client_secret,
 925            bearer_token=self.bearer_token,
 926        )
 927        return [
 928            CloudSource._from_source_response(  # noqa: SLF001 (non-public API)
 929                workspace=self,
 930                source_response=source,
 931            )
 932            for source in sources
 933        ]
 934
 935    def list_destinations(
 936        self,
 937        name: str | None = None,
 938        *,
 939        name_filter: Callable | None = None,
 940        limit: int | None = None,
 941    ) -> list[CloudDestination]:
 942        """List all destinations in the workspace, with an optional limit."""
 943        destinations = api_util.list_destinations(
 944            api_root=self.api_root,
 945            workspace_id=self.workspace_id,
 946            name=name,
 947            name_filter=name_filter,
 948            limit=limit,
 949            client_id=self.client_id,
 950            client_secret=self.client_secret,
 951            bearer_token=self.bearer_token,
 952        )
 953        return [
 954            CloudDestination._from_destination_response(  # noqa: SLF001 (non-public API)
 955                workspace=self,
 956                destination_response=destination,
 957            )
 958            for destination in destinations
 959        ]
 960
 961    def publish_custom_source_definition(
 962        self,
 963        name: str,
 964        *,
 965        manifest_yaml: dict[str, Any] | Path | str | None = None,
 966        docker_image: str | None = None,
 967        docker_tag: str | None = None,
 968        unique: bool = True,
 969        pre_validate: bool = True,
 970        testing_values: dict[str, Any] | None = None,
 971    ) -> CustomCloudSourceDefinition:
 972        """Publish a custom source connector definition.
 973
 974        You must specify EITHER manifest_yaml (for YAML connectors) OR both docker_image
 975        and docker_tag (for Docker connectors), but not both.
 976
 977        Args:
 978            name: Display name for the connector definition
 979            manifest_yaml: Low-code CDK manifest (dict, Path to YAML file, or YAML string)
 980            docker_image: Docker repository (e.g., 'airbyte/source-custom')
 981            docker_tag: Docker image tag (e.g., '1.0.0')
 982            unique: Whether to enforce name uniqueness
 983            pre_validate: Whether to validate manifest client-side (YAML only)
 984            testing_values: Optional configuration values to use for testing in the
 985                Connector Builder UI. If provided, these values are stored as the complete
 986                testing values object for the connector builder project (replaces any existing
 987                values), allowing immediate test read operations.
 988
 989        Returns:
 990            CustomCloudSourceDefinition object representing the created definition
 991
 992        Raises:
 993            PyAirbyteInputError: If both or neither of manifest_yaml and docker_image provided
 994            AirbyteDuplicateResourcesError: If unique=True and name already exists
 995        """
 996        is_yaml = manifest_yaml is not None
 997        is_docker = docker_image is not None
 998
 999        if is_yaml == is_docker:
1000            raise exc.PyAirbyteInputError(
1001                message=(
1002                    "Must specify EITHER manifest_yaml (for YAML connectors) OR "
1003                    "docker_image + docker_tag (for Docker connectors), but not both"
1004                ),
1005                context={
1006                    "manifest_yaml_provided": is_yaml,
1007                    "docker_image_provided": is_docker,
1008                },
1009            )
1010
1011        if is_docker and docker_tag is None:
1012            raise exc.PyAirbyteInputError(
1013                message="docker_tag is required when docker_image is specified",
1014                context={"docker_image": docker_image},
1015            )
1016
1017        if unique:
1018            existing = self.list_custom_source_definitions(
1019                definition_type="yaml" if is_yaml else "docker",
1020            )
1021            if any(d.name == name for d in existing):
1022                raise exc.AirbyteDuplicateResourcesError(
1023                    resource_type="custom_source_definition",
1024                    resource_name=name,
1025                )
1026
1027        if is_yaml:
1028            manifest_dict: dict[str, Any]
1029            if isinstance(manifest_yaml, Path):
1030                manifest_dict = yaml.safe_load(manifest_yaml.read_text())
1031            elif isinstance(manifest_yaml, str):
1032                manifest_dict = yaml.safe_load(manifest_yaml)
1033            elif manifest_yaml is not None:
1034                manifest_dict = manifest_yaml
1035            else:
1036                raise exc.PyAirbyteInputError(
1037                    message="manifest_yaml is required for YAML connectors",
1038                    context={"name": name},
1039                )
1040
1041            if pre_validate:
1042                api_util.validate_yaml_manifest(manifest_dict, raise_on_error=True)
1043
1044            result = api_util.create_custom_yaml_source_definition(
1045                name=name,
1046                workspace_id=self.workspace_id,
1047                manifest=manifest_dict,
1048                api_root=self.api_root,
1049                client_id=self.client_id,
1050                client_secret=self.client_secret,
1051                bearer_token=self.bearer_token,
1052            )
1053            custom_definition = CustomCloudSourceDefinition._from_yaml_response(  # noqa: SLF001
1054                self, result
1055            )
1056
1057            # Set testing values if provided
1058            if testing_values is not None:
1059                custom_definition.set_testing_values(testing_values)
1060
1061            return custom_definition
1062
1063        raise NotImplementedError(
1064            "Docker custom source definitions are not yet supported. "
1065            "Only YAML manifest-based custom sources are currently available."
1066        )
1067
1068    def list_custom_source_definitions(
1069        self,
1070        *,
1071        definition_type: Literal["yaml", "docker"],
1072    ) -> list[CustomCloudSourceDefinition]:
1073        """List custom source connector definitions.
1074
1075        Args:
1076            definition_type: Connector type to list ("yaml" or "docker"). Required.
1077
1078        Returns:
1079            List of CustomCloudSourceDefinition objects matching the specified type
1080        """
1081        if definition_type == "yaml":
1082            yaml_definitions = api_util.list_custom_yaml_source_definitions(
1083                workspace_id=self.workspace_id,
1084                api_root=self.api_root,
1085                client_id=self.client_id,
1086                client_secret=self.client_secret,
1087                bearer_token=self.bearer_token,
1088            )
1089            return [
1090                CustomCloudSourceDefinition._from_yaml_response(self, d)  # noqa: SLF001
1091                for d in yaml_definitions
1092            ]
1093
1094        raise NotImplementedError(
1095            "Docker custom source definitions are not yet supported. "
1096            "Only YAML manifest-based custom sources are currently available."
1097        )
1098
1099    def get_custom_source_definition(
1100        self,
1101        definition_id: str,
1102        *,
1103        definition_type: Literal["yaml", "docker"],
1104    ) -> CustomCloudSourceDefinition:
1105        """Get a specific custom source definition by ID.
1106
1107        Args:
1108            definition_id: The definition ID
1109            definition_type: Connector type ("yaml" or "docker"). Required.
1110
1111        Returns:
1112            CustomCloudSourceDefinition object
1113        """
1114        if definition_type == "yaml":
1115            result = api_util.get_custom_yaml_source_definition(
1116                workspace_id=self.workspace_id,
1117                definition_id=definition_id,
1118                api_root=self.api_root,
1119                client_id=self.client_id,
1120                client_secret=self.client_secret,
1121                bearer_token=self.bearer_token,
1122            )
1123            return CustomCloudSourceDefinition._from_yaml_response(self, result)  # noqa: SLF001
1124
1125        raise NotImplementedError(
1126            "Docker custom source definitions are not yet supported. "
1127            "Only YAML manifest-based custom sources are currently available."
1128        )

A remote workspace on the Airbyte Cloud.

By overriding api_root, you can use this class to interact with self-managed Airbyte instances, both OSS and Enterprise.

Two authentication methods are supported (mutually exclusive):

OAuth2 client credentials (client_id + client_secret)
Bearer token authentication

Example with client credentials:

workspace = CloudWorkspace(
    workspace_id="...",
    client_id="...",
    client_secret="...",
)

Example with bearer token:

workspace = CloudWorkspace(
    workspace_id="...",
    bearer_token="...",
)

152    def __init__(
153        self,
154        *,
155        workspace_id: str | None = None,
156        client_id: str | SecretString | None = None,
157        client_secret: str | SecretString | None = None,
158        api_root: str | None = None,
159        config_api_root: str | None = None,
160        bearer_token: str | SecretString | None = None,
161    ) -> None:
162        """Validate and initialize credentials."""
163        env_vars = not (client_id or client_secret or bearer_token)
164        credentials = _AirbyteCredentials.from_auth(
165            workspace_id=workspace_id,
166            client_id=client_id,
167            client_secret=client_secret,
168            bearer_token=bearer_token,
169            public_api_root=api_root,
170            config_api_root=config_api_root,
171            env_vars=env_vars,
172        )
173        if not credentials.workspace_id:
174            raise exc.PyAirbyteInputError(
175                message="Workspace ID is required.",
176                guidance=(
177                    "Provide a workspace ID, or call `get_default_cloud_context` to discover "
178                    "available workspaces."
179                ),
180            )
181
182        self._credentials = credentials
183        self.workspace_id = credentials.workspace_id or ""
184        self.client_id = credentials.client_id
185        self.client_secret = credentials.client_secret
186        self.bearer_token = credentials.bearer_token
187        self.api_root = credentials.public_api_root
188        self.config_api_root = credentials.config_api_root
189
190        # Create internal CloudClientConfig object (validates mutual exclusivity)
191        self._client_config = CloudClientConfig(
192            client_id=self.client_id,
193            client_secret=self.client_secret,
194            bearer_token=self.bearer_token,
195            api_root=self.api_root,
196            config_api_root=self.config_api_root,
197        )

Validate and initialize credentials.

workspace_id: str

client_id: airbyte.secrets.SecretString | None

client_secret: airbyte.secrets.SecretString | None

api_root: str

config_api_root: str | None

The Config API root URL.

bearer_token: airbyte.secrets.SecretString | None

@classmethod

def from_env( cls, workspace_id: str | None = None, *, api_root: str | None = None, config_api_root: str | None = None) -> CloudWorkspace: View Source

199    @classmethod
200    def from_env(
201        cls,
202        workspace_id: str | None = None,
203        *,
204        api_root: str | None = None,
205        config_api_root: str | None = None,
206    ) -> CloudWorkspace:
207        """Create a CloudWorkspace using credentials from environment variables.
208
209        This factory method resolves credentials from environment variables,
210        providing a convenient way to create a workspace without explicitly
211        passing credentials.
212
213        Two authentication methods are supported (mutually exclusive):
214        1. Bearer token (checked first)
215        2. OAuth2 client credentials (fallback)
216
217        Environment variables used:
218            - `AIRBYTE_CLOUD_BEARER_TOKEN`: Bearer token (alternative to client credentials).
219            - `AIRBYTE_CLOUD_CLIENT_ID`: OAuth client ID (for client credentials flow).
220            - `AIRBYTE_CLOUD_CLIENT_SECRET`: OAuth client secret (for client credentials flow).
221            - `AIRBYTE_CLOUD_WORKSPACE_ID`: The workspace ID (if not passed as argument).
222            - `AIRBYTE_CLOUD_API_URL`: Optional. The API root URL (defaults to Airbyte Cloud).
223            - `AIRBYTE_CLOUD_CONFIG_API_URL`: Optional. The Config API root URL.
224
225        Args:
226            workspace_id: The workspace ID. If not provided, will be resolved from
227                the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
228            api_root: The API root URL. If not provided, will be resolved from
229                the `AIRBYTE_CLOUD_API_URL` environment variable, or default to
230                the Airbyte Cloud API.
231            config_api_root: The Config API root URL. If not provided, will be resolved
232                from the `AIRBYTE_CLOUD_CONFIG_API_URL` environment variable.
233
234        Returns:
235            A CloudWorkspace instance configured with credentials from the environment.
236
237        Raises:
238            PyAirbyteInputError: If required credentials are not found in
239                the environment or are incomplete.
240
241        Example:
242            ```python
243            # With workspace_id from environment
244            workspace = CloudWorkspace.from_env()
245
246            # With explicit workspace_id
247            workspace = CloudWorkspace.from_env(workspace_id="your-workspace-id")
248            ```
249        """
250        return cls(
251            workspace_id=workspace_id,
252            api_root=api_root,
253            config_api_root=config_api_root,
254        )

Create a CloudWorkspace using credentials from environment variables.

This factory method resolves credentials from environment variables, providing a convenient way to create a workspace without explicitly passing credentials.

Two authentication methods are supported (mutually exclusive):

Bearer token (checked first)
OAuth2 client credentials (fallback)

Environment variables used:

AIRBYTE_CLOUD_BEARER_TOKEN: Bearer token (alternative to client credentials).

AIRBYTE_CLOUD_CLIENT_ID: OAuth client ID (for client credentials flow).

AIRBYTE_CLOUD_CLIENT_SECRET: OAuth client secret (for client credentials flow).

AIRBYTE_CLOUD_WORKSPACE_ID: The workspace ID (if not passed as argument).

AIRBYTE_CLOUD_API_URL: Optional. The API root URL (defaults to Airbyte Cloud).

AIRBYTE_CLOUD_CONFIG_API_URL: Optional. The Config API root URL.

Arguments:

workspace_id: The workspace ID. If not provided, will be resolved from the AIRBYTE_CLOUD_WORKSPACE_ID environment variable.
api_root: The API root URL. If not provided, will be resolved from the AIRBYTE_CLOUD_API_URL environment variable, or default to the Airbyte Cloud API.
config_api_root: The Config API root URL. If not provided, will be resolved from the AIRBYTE_CLOUD_CONFIG_API_URL environment variable.

Returns:

A CloudWorkspace instance configured with credentials from the environment.

Raises:

PyAirbyteInputError: If required credentials are not found in the environment or are incomplete.

Example:

# With workspace_id from environment
workspace = CloudWorkspace.from_env()

# With explicit workspace_id
workspace = CloudWorkspace.from_env(workspace_id="your-workspace-id")

workspace_url: str | None View Source

256    @property
257    def workspace_url(self) -> str | None:
258        """The web URL of the workspace."""
259        return f"{get_web_url_root(self.api_root)}/workspaces/{self.workspace_id}"

The web URL of the workspace.

def get_organization( self, *, raise_on_error: bool = True) -> CloudOrganization | None: View Source

294    def get_organization(
295        self,
296        *,
297        raise_on_error: bool = True,
298    ) -> CloudOrganization | None:
299        """Get the organization this workspace belongs to.
300
301        Fetching organization info requires ORGANIZATION_READER permissions on the organization,
302        which may not be available with workspace-scoped credentials.
303
304        Args:
305            raise_on_error: If True (default), raises AirbyteError on permission or API errors.
306                If False, returns None instead of raising.
307
308        Returns:
309            CloudOrganization object with organization_id and organization_name,
310            or None if raise_on_error=False and an error occurred.
311
312        Raises:
313            AirbyteError: If raise_on_error=True and the organization info cannot be fetched
314                (e.g., due to insufficient permissions or missing data).
315        """
316        try:
317            info = self._organization_info
318        except (AirbyteError, NotImplementedError):
319            if raise_on_error:
320                raise
321            return None
322
323        organization_id = info.get("organizationId")
324        organization_name = info.get("organizationName")
325
326        # Validate that both organization_id and organization_name are non-null and non-empty
327        if not organization_id or not organization_name:
328            if raise_on_error:
329                raise AirbyteError(
330                    message="Organization info is incomplete.",
331                    context={
332                        "organization_id": organization_id,
333                        "organization_name": organization_name,
334                    },
335                )
336            return None
337
338        organization_credentials = self._credentials.with_organization_id(organization_id)
339        return CloudOrganization(
340            organization_id=organization_id,
341            organization_name=organization_name,
342            client_id=organization_credentials.client_id,
343            client_secret=organization_credentials.client_secret,
344            bearer_token=organization_credentials.bearer_token,
345            public_api_root=organization_credentials.public_api_root,
346            config_api_root=organization_credentials.config_api_root,
347        )

Get the organization this workspace belongs to.

Fetching organization info requires ORGANIZATION_READER permissions on the organization, which may not be available with workspace-scoped credentials.

Arguments:

raise_on_error: If True (default), raises AirbyteError on permission or API errors. If False, returns None instead of raising.

Returns:

CloudOrganization object with organization_id and organization_name, or None if raise_on_error=False and an error occurred.

Raises:

AirbyteError: If raise_on_error=True and the organization info cannot be fetched (e.g., due to insufficient permissions or missing data).

def connect(self) -> None: View Source

351    def connect(self) -> None:
352        """Check that the workspace is reachable and raise an exception otherwise.
353
354        Note: It is not necessary to call this method before calling other operations. It
355              serves primarily as a simple check to ensure that the workspace is reachable
356              and credentials are correct.
357        """
358        _ = api_util.get_workspace(
359            api_root=self.api_root,
360            workspace_id=self.workspace_id,
361            client_id=self.client_id,
362            client_secret=self.client_secret,
363            bearer_token=self.bearer_token,
364        )
365        print(f"Successfully connected to workspace: {self.workspace_url}")

Check that the workspace is reachable and raise an exception otherwise.

Note: It is not necessary to call this method before calling other operations. It serves primarily as a simple check to ensure that the workspace is reachable and credentials are correct.

def get_connection(self, connection_id: str) -> CloudConnection: View Source

369    def get_connection(
370        self,
371        connection_id: str,
372    ) -> CloudConnection:
373        """Get a connection by ID.
374
375        This method does not fetch data from the API. It returns a `CloudConnection` object,
376        which will be loaded lazily as needed.
377        """
378        return CloudConnection(
379            workspace=self,
380            connection_id=connection_id,
381        )

Get a connection by ID.

This method does not fetch data from the API. It returns a CloudConnection object, which will be loaded lazily as needed.

def get_source(self, source_id: str) -> airbyte.cloud.connectors.CloudSource: View Source

383    def get_source(
384        self,
385        source_id: str,
386    ) -> CloudSource:
387        """Get a source by ID.
388
389        This method does not fetch data from the API. It returns a `CloudSource` object,
390        which will be loaded lazily as needed.
391        """
392        return CloudSource(
393            workspace=self,
394            connector_id=source_id,
395        )

Get a source by ID.

This method does not fetch data from the API. It returns a CloudSource object, which will be loaded lazily as needed.

def get_destination(self, destination_id: str) -> airbyte.cloud.connectors.CloudDestination: View Source

397    def get_destination(
398        self,
399        destination_id: str,
400    ) -> CloudDestination:
401        """Get a destination by ID.
402
403        This method does not fetch data from the API. It returns a `CloudDestination` object,
404        which will be loaded lazily as needed.
405        """
406        return CloudDestination(
407            workspace=self,
408            connector_id=destination_id,
409        )

Get a destination by ID.

This method does not fetch data from the API. It returns a CloudDestination object, which will be loaded lazily as needed.

def check_connector_setup( self, connector_type: Literal['source', 'destination'], connector_id: str) -> airbyte.cloud.models.CheckResult: View Source

411    def check_connector_setup(
412        self,
413        connector_type: Literal["source", "destination"],
414        connector_id: str,
415    ) -> CheckResult:
416        """Run one connection check on a connector that belongs to this workspace.
417
418        Confirms a person has finished a deferred-credential setup in Airbyte Cloud. The
419        connector's workspace is verified first so a check can never be run against a connector
420        outside this workspace.
421        """
422        connector: CloudSource | CloudDestination
423        if connector_type == "source":
424            owner_id = api_util.get_source(
425                source_id=connector_id,
426                api_root=self.api_root,
427                client_id=self.client_id,
428                client_secret=self.client_secret,
429                bearer_token=self.bearer_token,
430            ).workspace_id
431            connector = self.get_source(connector_id)
432        else:
433            owner_id = api_util.get_destination(
434                destination_id=connector_id,
435                api_root=self.api_root,
436                client_id=self.client_id,
437                client_secret=self.client_secret,
438                bearer_token=self.bearer_token,
439            ).workspace_id
440            connector = self.get_destination(connector_id)
441        if owner_id != self.workspace_id:
442            raise exc.AirbyteMissingResourceError(
443                resource_type=connector_type,
444                resource_name_or_id=connector_id,
445                context={"workspace_id": self.workspace_id},
446            )
447        try:
448            return connector.check(raise_on_error=False)
449        except AirbyteError as ex:
450            status_code = (ex.context or {}).get("status_code")
451            if status_code == HTTPStatus.UNPROCESSABLE_ENTITY:
452                return CheckResult(success=False)
453            raise AirbyteError(
454                message="Cloud could not check the connector setup.",
455                context={"status_code": status_code},
456            ) from None

Run one connection check on a connector that belongs to this workspace.

Confirms a person has finished a deferred-credential setup in Airbyte Cloud. The connector's workspace is verified first so a check can never be run against a connector outside this workspace.

def deploy_source( self, name: str, source: airbyte.Source | dict[str, typing.Any], *, unique: bool = True, random_name_suffix: bool = False, definition_id: str | None = None, defer_credentials: bool = False) -> airbyte.cloud.connectors.CloudSource: View Source

460    def deploy_source(
461        self,
462        name: str,
463        source: Source | dict[str, Any],
464        *,
465        unique: bool = True,
466        random_name_suffix: bool = False,
467        definition_id: str | None = None,
468        defer_credentials: bool = False,
469    ) -> CloudSource:
470        """Deploy a source to the workspace.
471
472        Returns the newly deployed source.
473
474        Args:
475            name: The name to use when deploying.
476            source: The source object to deploy, or (with `defer_credentials=True`) a
477                dictionary of non-secret configuration values.
478            unique: Whether to require a unique name. If `True`, duplicate names
479                are not allowed. Defaults to `True`.
480            random_name_suffix: Whether to append a random suffix to the name.
481            definition_id: The source definition ID. Required with `defer_credentials=True`.
482            defer_credentials: Save a draft with partial configuration. A person completes
483                credentials and other missing settings at the returned source's `connector_url`.
484                A successful connection check promotes the draft. Raises
485                `AirbyteDeferredSetupError` if Cloud does not acknowledge draft mode.
486        """
487        if defer_credentials:
488            return CloudSource(
489                workspace=self,
490                connector_id=self._deploy_deferred(
491                    connector_type="source",
492                    name=name,
493                    config=source,
494                    definition_id=definition_id,
495                    unique=unique,
496                    random_name_suffix=random_name_suffix,
497                ),
498            )
499        if isinstance(source, dict):
500            raise exc.PyAirbyteInputError(
501                message="`source` must be a `Source` object unless `defer_credentials=True`.",
502            )
503
504        source_config_dict = source._hydrated_config.copy()  # noqa: SLF001 (non-public API)
505        source_config_dict["sourceType"] = source.name.replace("source-", "")
506
507        if random_name_suffix:
508            name += f" (ID: {text_util.generate_random_suffix()})"
509
510        if unique:
511            existing = self.list_sources(name=name)
512            if existing:
513                raise exc.AirbyteDuplicateResourcesError(
514                    resource_type="source",
515                    resource_name=name,
516                )
517
518        deployed_source = api_util.create_source(
519            name=name,
520            api_root=self.api_root,
521            workspace_id=self.workspace_id,
522            config=source_config_dict,
523            definition_id=definition_id,
524            client_id=self.client_id,
525            client_secret=self.client_secret,
526            bearer_token=self.bearer_token,
527        )
528        return CloudSource(
529            workspace=self,
530            connector_id=deployed_source.source_id,
531        )

Deploy a source to the workspace.

Returns the newly deployed source.

Arguments:

name: The name to use when deploying.
source: The source object to deploy, or (with defer_credentials=True) a dictionary of non-secret configuration values.
unique: Whether to require a unique name. If True, duplicate names are not allowed. Defaults to True.
random_name_suffix: Whether to append a random suffix to the name.
definition_id: The source definition ID. Required with defer_credentials=True.
defer_credentials: Save a draft with partial configuration. A person completes credentials and other missing settings at the returned source's connector_url. A successful connection check promotes the draft. Raises AirbyteDeferredSetupError if Cloud does not acknowledge draft mode.

def deploy_destination( self, name: str, destination: airbyte.Destination | dict[str, typing.Any], *, unique: bool = True, random_name_suffix: bool = False, definition_id: str | None = None, defer_credentials: bool = False) -> airbyte.cloud.connectors.CloudDestination: View Source

533    def deploy_destination(
534        self,
535        name: str,
536        destination: Destination | dict[str, Any],
537        *,
538        unique: bool = True,
539        random_name_suffix: bool = False,
540        definition_id: str | None = None,
541        defer_credentials: bool = False,
542    ) -> CloudDestination:
543        """Deploy a destination to the workspace.
544
545        Returns the newly deployed destination ID.
546
547        Args:
548            name: The name to use when deploying.
549            destination: The destination to deploy. Can be a local Airbyte `Destination` object or a
550                dictionary of configuration values.
551            unique: Whether to require a unique name. If `True`, duplicate names
552                are not allowed. Defaults to `True`.
553            random_name_suffix: Whether to append a random suffix to the name.
554            definition_id: The destination definition ID. Required with `defer_credentials=True`;
555                otherwise the type is inferred from `destinationType`.
556            defer_credentials: Create the destination without its credentials. See
557                `deploy_source`.
558        """
559        if defer_credentials:
560            return CloudDestination(
561                workspace=self,
562                connector_id=self._deploy_deferred(
563                    connector_type="destination",
564                    name=name,
565                    config=destination,
566                    definition_id=definition_id,
567                    unique=unique,
568                    random_name_suffix=random_name_suffix,
569                ),
570            )
571
572        if isinstance(destination, Destination):
573            destination_conf_dict = destination._hydrated_config.copy()  # noqa: SLF001 (non-public API)
574            destination_conf_dict["destinationType"] = destination.name.replace("destination-", "")
575            # raise ValueError(destination_conf_dict)
576        else:
577            destination_conf_dict = destination.copy()
578            if "destinationType" not in destination_conf_dict:
579                raise exc.PyAirbyteInputError(
580                    message="Missing `destinationType` in configuration dictionary.",
581                )
582
583        if random_name_suffix:
584            name += f" (ID: {text_util.generate_random_suffix()})"
585
586        if unique:
587            existing = self.list_destinations(name=name)
588            if existing:
589                raise exc.AirbyteDuplicateResourcesError(
590                    resource_type="destination",
591                    resource_name=name,
592                )
593
594        deployed_destination = api_util.create_destination(
595            name=name,
596            api_root=self.api_root,
597            workspace_id=self.workspace_id,
598            config=destination_conf_dict,  # Wants a dataclass but accepts dict
599            client_id=self.client_id,
600            client_secret=self.client_secret,
601            bearer_token=self.bearer_token,
602        )
603        return CloudDestination(
604            workspace=self,
605            connector_id=deployed_destination.destination_id,
606        )

Deploy a destination to the workspace.

Returns the newly deployed destination ID.

Arguments:

name: The name to use when deploying.
destination: The destination to deploy. Can be a local Airbyte Destination object or a dictionary of configuration values.
unique: Whether to require a unique name. If True, duplicate names are not allowed. Defaults to True.
random_name_suffix: Whether to append a random suffix to the name.
definition_id: The destination definition ID. Required with defer_credentials=True; otherwise the type is inferred from destinationType.
defer_credentials: Create the destination without its credentials. See deploy_source.

def permanently_delete_source( self, source: str | airbyte.cloud.connectors.CloudSource, *, safe_mode: bool = True) -> None: View Source

651    def permanently_delete_source(
652        self,
653        source: str | CloudSource,
654        *,
655        safe_mode: bool = True,
656    ) -> None:
657        """Delete a source from the workspace.
658
659        You can pass either the source ID `str` or a deployed `Source` object.
660
661        Args:
662            source: The source ID or CloudSource object to delete
663            safe_mode: If True, requires the source name to contain "delete-me" or "deleteme"
664                (case insensitive) to prevent accidental deletion. Defaults to True.
665        """
666        if not isinstance(source, (str, CloudSource)):
667            raise exc.PyAirbyteInputError(
668                message="Invalid source type.",
669                input_value=type(source).__name__,
670            )
671
672        api_util.delete_source(
673            source_id=source.connector_id if isinstance(source, CloudSource) else source,
674            source_name=source.name if isinstance(source, CloudSource) else None,
675            api_root=self.api_root,
676            client_id=self.client_id,
677            client_secret=self.client_secret,
678            bearer_token=self.bearer_token,
679            safe_mode=safe_mode,
680        )

Delete a source from the workspace.

You can pass either the source ID str or a deployed Source object.

Arguments:

source: The source ID or CloudSource object to delete
safe_mode: If True, requires the source name to contain "delete-me" or "deleteme" (case insensitive) to prevent accidental deletion. Defaults to True.

def permanently_delete_destination( self, destination: str | airbyte.cloud.connectors.CloudDestination, *, safe_mode: bool = True) -> None: View Source

684    def permanently_delete_destination(
685        self,
686        destination: str | CloudDestination,
687        *,
688        safe_mode: bool = True,
689    ) -> None:
690        """Delete a deployed destination from the workspace.
691
692        You can pass either the `Cache` class or the deployed destination ID as a `str`.
693
694        Args:
695            destination: The destination ID or CloudDestination object to delete
696            safe_mode: If True, requires the destination name to contain "delete-me" or "deleteme"
697                (case insensitive) to prevent accidental deletion. Defaults to True.
698        """
699        if not isinstance(destination, (str, CloudDestination)):
700            raise exc.PyAirbyteInputError(
701                message="Invalid destination type.",
702                input_value=type(destination).__name__,
703            )
704
705        api_util.delete_destination(
706            destination_id=(
707                destination if isinstance(destination, str) else destination.destination_id
708            ),
709            destination_name=(
710                destination.name if isinstance(destination, CloudDestination) else None
711            ),
712            api_root=self.api_root,
713            client_id=self.client_id,
714            client_secret=self.client_secret,
715            bearer_token=self.bearer_token,
716            safe_mode=safe_mode,
717        )

Delete a deployed destination from the workspace.

You can pass either the Cache class or the deployed destination ID as a str.

Arguments:

destination: The destination ID or CloudDestination object to delete
safe_mode: If True, requires the destination name to contain "delete-me" or "deleteme" (case insensitive) to prevent accidental deletion. Defaults to True.

def deploy_connection( self, connection_name: str, *, source: airbyte.cloud.connectors.CloudSource | str, selected_streams: list[str], destination: airbyte.cloud.connectors.CloudDestination | str, table_prefix: str | None = None) -> CloudConnection: View Source

721    def deploy_connection(
722        self,
723        connection_name: str,
724        *,
725        source: CloudSource | str,
726        selected_streams: list[str],
727        destination: CloudDestination | str,
728        table_prefix: str | None = None,
729    ) -> CloudConnection:
730        """Create a new connection between an already deployed source and destination.
731
732        Returns the newly deployed connection object.
733
734        Args:
735            connection_name: The name of the connection.
736            source: The deployed source. You can pass a source ID or a CloudSource object.
737            destination: The deployed destination. You can pass a destination ID or a
738                CloudDestination object.
739            table_prefix: Optional. The table prefix to use when syncing to the destination.
740            selected_streams: The selected stream names to sync within the connection.
741        """
742        if not selected_streams:
743            raise exc.PyAirbyteInputError(
744                guidance="You must provide `selected_streams` when creating a connection."
745            )
746
747        source_id: str = source if isinstance(source, str) else source.connector_id
748        destination_id: str = (
749            destination if isinstance(destination, str) else destination.connector_id
750        )
751
752        deployed_connection = api_util.create_connection(
753            name=connection_name,
754            source_id=source_id,
755            destination_id=destination_id,
756            api_root=self.api_root,
757            workspace_id=self.workspace_id,
758            selected_stream_names=selected_streams,
759            prefix=table_prefix or "",
760            client_id=self.client_id,
761            client_secret=self.client_secret,
762            bearer_token=self.bearer_token,
763        )
764
765        return CloudConnection(
766            workspace=self,
767            connection_id=deployed_connection.connection_id,
768            source=deployed_connection.source_id,
769            destination=deployed_connection.destination_id,
770        )

Create a new connection between an already deployed source and destination.

Returns the newly deployed connection object.

Arguments:

connection_name: The name of the connection.
source: The deployed source. You can pass a source ID or a CloudSource object.
destination: The deployed destination. You can pass a destination ID or a CloudDestination object.
table_prefix: Optional. The table prefix to use when syncing to the destination.
selected_streams: The selected stream names to sync within the connection.

def permanently_delete_connection( self, connection: str | CloudConnection, *, cascade_delete_source: bool = False, cascade_delete_destination: bool = False, safe_mode: bool = True) -> None: View Source

772    def permanently_delete_connection(
773        self,
774        connection: str | CloudConnection,
775        *,
776        cascade_delete_source: bool = False,
777        cascade_delete_destination: bool = False,
778        safe_mode: bool = True,
779    ) -> None:
780        """Delete a deployed connection from the workspace.
781
782        Args:
783            connection: The connection ID or CloudConnection object to delete
784            cascade_delete_source: If True, also delete the source after deleting the connection
785            cascade_delete_destination: If True, also delete the destination after deleting
786                the connection
787            safe_mode: If True, requires the connection name to contain "delete-me" or "deleteme"
788                (case insensitive) to prevent accidental deletion. Defaults to True. Also applies
789                to cascade deletes.
790        """
791        if connection is None:
792            raise ValueError("No connection ID provided.")
793
794        if isinstance(connection, str):
795            connection = CloudConnection(
796                workspace=self,
797                connection_id=connection,
798            )
799
800        api_util.delete_connection(
801            connection_id=connection.connection_id,
802            connection_name=connection.name,
803            api_root=self.api_root,
804            workspace_id=self.workspace_id,
805            client_id=self.client_id,
806            client_secret=self.client_secret,
807            bearer_token=self.bearer_token,
808            safe_mode=safe_mode,
809        )
810
811        if cascade_delete_source:
812            self.permanently_delete_source(
813                source=connection.source_id,
814                safe_mode=safe_mode,
815            )
816        if cascade_delete_destination:
817            self.permanently_delete_destination(
818                destination=connection.destination_id,
819                safe_mode=safe_mode,
820            )

Delete a deployed connection from the workspace.

Arguments:

connection: The connection ID or CloudConnection object to delete
cascade_delete_source: If True, also delete the source after deleting the connection
cascade_delete_destination: If True, also delete the destination after deleting the connection
safe_mode: If True, requires the connection name to contain "delete-me" or "deleteme" (case insensitive) to prevent accidental deletion. Defaults to True. Also applies to cascade deletes.

def list_workspaces( self, name: str | None = None, *, name_filter: Callable | None = None, limit: int | None = None) -> list[CloudWorkspaceInfo]: View Source

824    def list_workspaces(
825        self,
826        name: str | None = None,
827        *,
828        name_filter: Callable | None = None,
829        limit: int | None = None,
830    ) -> list[CloudWorkspaceInfo]:
831        """List workspaces available to the current credentials, with an optional limit."""
832        return [
833            CloudWorkspaceInfo.from_api_response(workspace)
834            for workspace in api_util.list_workspaces(
835                workspace_id="",
836                api_root=self.api_root,
837                name=name,
838                name_filter=name_filter,
839                client_id=self.client_id,
840                client_secret=self.client_secret,
841                bearer_token=self.bearer_token,
842                limit=limit,
843            )
844        ]

List workspaces available to the current credentials, with an optional limit.

def rename(self, name: str) -> CloudWorkspace: View Source

846    def rename(
847        self,
848        name: str,
849    ) -> CloudWorkspace:
850        """Rename this workspace."""
851        api_util.rename_workspace(
852            workspace_id=self.workspace_id,
853            name=name,
854            api_root=self.api_root,
855            client_id=self.client_id,
856            client_secret=self.client_secret,
857            bearer_token=self.bearer_token,
858        )
859        return self

Rename this workspace.

def permanently_delete( self, *, workspace_name: str | None = None, safe_mode: bool = True) -> None: View Source

861    def permanently_delete(
862        self,
863        *,
864        workspace_name: str | None = None,
865        safe_mode: bool = True,
866    ) -> None:
867        """Permanently delete this workspace if it has no connections.
868
869        When `safe_mode` is enabled, the workspace name must contain `delete-me`
870        or `deleteme`. This also checks for existing connections before deleting
871        and raises `AirbyteWorkspaceNotEmptyError` if the workspace is not empty.
872        """
873        api_util.permanently_delete_workspace(
874            workspace_id=self.workspace_id,
875            workspace_name=workspace_name,
876            api_root=self.api_root,
877            client_id=self.client_id,
878            client_secret=self.client_secret,
879            bearer_token=self.bearer_token,
880            safe_mode=safe_mode,
881        )

Permanently delete this workspace if it has no connections.

When safe_mode is enabled, the workspace name must contain delete-me or deleteme. This also checks for existing connections before deleting and raises AirbyteWorkspaceNotEmptyError if the workspace is not empty.

def list_connections( self, name: str | None = None, *, name_filter: Callable | None = None, limit: int | None = None) -> list[CloudConnection]: View Source

883    def list_connections(
884        self,
885        name: str | None = None,
886        *,
887        name_filter: Callable | None = None,
888        limit: int | None = None,
889    ) -> list[CloudConnection]:
890        """List connections by name in the workspace, with an optional limit."""
891        connections = api_util.list_connections(
892            api_root=self.api_root,
893            workspace_id=self.workspace_id,
894            name=name,
895            name_filter=name_filter,
896            limit=limit,
897            client_id=self.client_id,
898            client_secret=self.client_secret,
899            bearer_token=self.bearer_token,
900        )
901        return [
902            CloudConnection._from_connection_response(  # noqa: SLF001 (non-public API)
903                workspace=self,
904                connection_response=connection,
905            )
906            for connection in connections
907        ]

List connections by name in the workspace, with an optional limit.

def list_sources( self, name: str | None = None, *, name_filter: Callable | None = None, limit: int | None = None) -> list[airbyte.cloud.connectors.CloudSource]: View Source

909    def list_sources(
910        self,
911        name: str | None = None,
912        *,
913        name_filter: Callable | None = None,
914        limit: int | None = None,
915    ) -> list[CloudSource]:
916        """List all sources in the workspace, with an optional limit."""
917        sources = api_util.list_sources(
918            api_root=self.api_root,
919            workspace_id=self.workspace_id,
920            name=name,
921            name_filter=name_filter,
922            limit=limit,
923            client_id=self.client_id,
924            client_secret=self.client_secret,
925            bearer_token=self.bearer_token,
926        )
927        return [
928            CloudSource._from_source_response(  # noqa: SLF001 (non-public API)
929                workspace=self,
930                source_response=source,
931            )
932            for source in sources
933        ]

List all sources in the workspace, with an optional limit.

def list_destinations( self, name: str | None = None, *, name_filter: Callable | None = None, limit: int | None = None) -> list[airbyte.cloud.connectors.CloudDestination]: View Source

935    def list_destinations(
936        self,
937        name: str | None = None,
938        *,
939        name_filter: Callable | None = None,
940        limit: int | None = None,
941    ) -> list[CloudDestination]:
942        """List all destinations in the workspace, with an optional limit."""
943        destinations = api_util.list_destinations(
944            api_root=self.api_root,
945            workspace_id=self.workspace_id,
946            name=name,
947            name_filter=name_filter,
948            limit=limit,
949            client_id=self.client_id,
950            client_secret=self.client_secret,
951            bearer_token=self.bearer_token,
952        )
953        return [
954            CloudDestination._from_destination_response(  # noqa: SLF001 (non-public API)
955                workspace=self,
956                destination_response=destination,
957            )
958            for destination in destinations
959        ]

List all destinations in the workspace, with an optional limit.

def publish_custom_source_definition( self, name: str, *, manifest_yaml: dict[str, typing.Any] | pathlib.Path | str | None = None, docker_image: str | None = None, docker_tag: str | None = None, unique: bool = True, pre_validate: bool = True, testing_values: dict[str, typing.Any] | None = None) -> airbyte.cloud.connectors.CustomCloudSourceDefinition: View Source

 961    def publish_custom_source_definition(
 962        self,
 963        name: str,
 964        *,
 965        manifest_yaml: dict[str, Any] | Path | str | None = None,
 966        docker_image: str | None = None,
 967        docker_tag: str | None = None,
 968        unique: bool = True,
 969        pre_validate: bool = True,
 970        testing_values: dict[str, Any] | None = None,
 971    ) -> CustomCloudSourceDefinition:
 972        """Publish a custom source connector definition.
 973
 974        You must specify EITHER manifest_yaml (for YAML connectors) OR both docker_image
 975        and docker_tag (for Docker connectors), but not both.
 976
 977        Args:
 978            name: Display name for the connector definition
 979            manifest_yaml: Low-code CDK manifest (dict, Path to YAML file, or YAML string)
 980            docker_image: Docker repository (e.g., 'airbyte/source-custom')
 981            docker_tag: Docker image tag (e.g., '1.0.0')
 982            unique: Whether to enforce name uniqueness
 983            pre_validate: Whether to validate manifest client-side (YAML only)
 984            testing_values: Optional configuration values to use for testing in the
 985                Connector Builder UI. If provided, these values are stored as the complete
 986                testing values object for the connector builder project (replaces any existing
 987                values), allowing immediate test read operations.
 988
 989        Returns:
 990            CustomCloudSourceDefinition object representing the created definition
 991
 992        Raises:
 993            PyAirbyteInputError: If both or neither of manifest_yaml and docker_image provided
 994            AirbyteDuplicateResourcesError: If unique=True and name already exists
 995        """
 996        is_yaml = manifest_yaml is not None
 997        is_docker = docker_image is not None
 998
 999        if is_yaml == is_docker:
1000            raise exc.PyAirbyteInputError(
1001                message=(
1002                    "Must specify EITHER manifest_yaml (for YAML connectors) OR "
1003                    "docker_image + docker_tag (for Docker connectors), but not both"
1004                ),
1005                context={
1006                    "manifest_yaml_provided": is_yaml,
1007                    "docker_image_provided": is_docker,
1008                },
1009            )
1010
1011        if is_docker and docker_tag is None:
1012            raise exc.PyAirbyteInputError(
1013                message="docker_tag is required when docker_image is specified",
1014                context={"docker_image": docker_image},
1015            )
1016
1017        if unique:
1018            existing = self.list_custom_source_definitions(
1019                definition_type="yaml" if is_yaml else "docker",
1020            )
1021            if any(d.name == name for d in existing):
1022                raise exc.AirbyteDuplicateResourcesError(
1023                    resource_type="custom_source_definition",
1024                    resource_name=name,
1025                )
1026
1027        if is_yaml:
1028            manifest_dict: dict[str, Any]
1029            if isinstance(manifest_yaml, Path):
1030                manifest_dict = yaml.safe_load(manifest_yaml.read_text())
1031            elif isinstance(manifest_yaml, str):
1032                manifest_dict = yaml.safe_load(manifest_yaml)
1033            elif manifest_yaml is not None:
1034                manifest_dict = manifest_yaml
1035            else:
1036                raise exc.PyAirbyteInputError(
1037                    message="manifest_yaml is required for YAML connectors",
1038                    context={"name": name},
1039                )
1040
1041            if pre_validate:
1042                api_util.validate_yaml_manifest(manifest_dict, raise_on_error=True)
1043
1044            result = api_util.create_custom_yaml_source_definition(
1045                name=name,
1046                workspace_id=self.workspace_id,
1047                manifest=manifest_dict,
1048                api_root=self.api_root,
1049                client_id=self.client_id,
1050                client_secret=self.client_secret,
1051                bearer_token=self.bearer_token,
1052            )
1053            custom_definition = CustomCloudSourceDefinition._from_yaml_response(  # noqa: SLF001
1054                self, result
1055            )
1056
1057            # Set testing values if provided
1058            if testing_values is not None:
1059                custom_definition.set_testing_values(testing_values)
1060
1061            return custom_definition
1062
1063        raise NotImplementedError(
1064            "Docker custom source definitions are not yet supported. "
1065            "Only YAML manifest-based custom sources are currently available."
1066        )

Publish a custom source connector definition.

You must specify EITHER manifest_yaml (for YAML connectors) OR both docker_image and docker_tag (for Docker connectors), but not both.

Arguments:

name: Display name for the connector definition
manifest_yaml: Low-code CDK manifest (dict, Path to YAML file, or YAML string)
docker_image: Docker repository (e.g., 'airbyte/source-custom')
docker_tag: Docker image tag (e.g., '1.0.0')
unique: Whether to enforce name uniqueness
pre_validate: Whether to validate manifest client-side (YAML only)
testing_values: Optional configuration values to use for testing in the Connector Builder UI. If provided, these values are stored as the complete testing values object for the connector builder project (replaces any existing values), allowing immediate test read operations.

Returns:

CustomCloudSourceDefinition object representing the created definition

Raises:

PyAirbyteInputError: If both or neither of manifest_yaml and docker_image provided
AirbyteDuplicateResourcesError: If unique=True and name already exists

def list_custom_source_definitions( self, *, definition_type: Literal['yaml', 'docker']) -> list[airbyte.cloud.connectors.CustomCloudSourceDefinition]: View Source

1068    def list_custom_source_definitions(
1069        self,
1070        *,
1071        definition_type: Literal["yaml", "docker"],
1072    ) -> list[CustomCloudSourceDefinition]:
1073        """List custom source connector definitions.
1074
1075        Args:
1076            definition_type: Connector type to list ("yaml" or "docker"). Required.
1077
1078        Returns:
1079            List of CustomCloudSourceDefinition objects matching the specified type
1080        """
1081        if definition_type == "yaml":
1082            yaml_definitions = api_util.list_custom_yaml_source_definitions(
1083                workspace_id=self.workspace_id,
1084                api_root=self.api_root,
1085                client_id=self.client_id,
1086                client_secret=self.client_secret,
1087                bearer_token=self.bearer_token,
1088            )
1089            return [
1090                CustomCloudSourceDefinition._from_yaml_response(self, d)  # noqa: SLF001
1091                for d in yaml_definitions
1092            ]
1093
1094        raise NotImplementedError(
1095            "Docker custom source definitions are not yet supported. "
1096            "Only YAML manifest-based custom sources are currently available."
1097        )

List custom source connector definitions.

Arguments:

definition_type: Connector type to list ("yaml" or "docker"). Required.

Returns:

List of CustomCloudSourceDefinition objects matching the specified type

def get_custom_source_definition( self, definition_id: str, *, definition_type: Literal['yaml', 'docker']) -> airbyte.cloud.connectors.CustomCloudSourceDefinition: View Source

1099    def get_custom_source_definition(
1100        self,
1101        definition_id: str,
1102        *,
1103        definition_type: Literal["yaml", "docker"],
1104    ) -> CustomCloudSourceDefinition:
1105        """Get a specific custom source definition by ID.
1106
1107        Args:
1108            definition_id: The definition ID
1109            definition_type: Connector type ("yaml" or "docker"). Required.
1110
1111        Returns:
1112            CustomCloudSourceDefinition object
1113        """
1114        if definition_type == "yaml":
1115            result = api_util.get_custom_yaml_source_definition(
1116                workspace_id=self.workspace_id,
1117                definition_id=definition_id,
1118                api_root=self.api_root,
1119                client_id=self.client_id,
1120                client_secret=self.client_secret,
1121                bearer_token=self.bearer_token,
1122            )
1123            return CustomCloudSourceDefinition._from_yaml_response(self, result)  # noqa: SLF001
1124
1125        raise NotImplementedError(
1126            "Docker custom source definitions are not yet supported. "
1127            "Only YAML manifest-based custom sources are currently available."
1128        )

Get a specific custom source definition by ID.

Arguments:

definition_id: The definition ID
definition_type: Connector type ("yaml" or "docker"). Required.

Returns:

CustomCloudSourceDefinition object

@dataclass

class CloudClientConfig: View Source

 59@dataclass
 60class CloudClientConfig:
 61    """Client configuration for Airbyte Cloud API.
 62
 63    This class encapsulates the authentication and API configuration needed to connect
 64    to Airbyte Cloud, OSS, or Enterprise instances. It supports two mutually
 65    exclusive authentication methods:
 66
 67    1. OAuth2 client credentials flow (client_id + client_secret)
 68    2. Bearer token authentication
 69
 70    Exactly one authentication method must be provided. Providing both or neither
 71    will raise a validation error.
 72
 73    Attributes:
 74        client_id: OAuth2 client ID for client credentials flow.
 75        client_secret: OAuth2 client secret for client credentials flow.
 76        bearer_token: Pre-generated bearer token for direct authentication.
 77        api_root: The API root URL. Defaults to Airbyte Cloud API.
 78        config_api_root: The Config API root URL.
 79    """
 80
 81    client_id: SecretString | None = None
 82    """OAuth2 client ID for client credentials authentication."""
 83
 84    client_secret: SecretString | None = None
 85    """OAuth2 client secret for client credentials authentication."""
 86
 87    bearer_token: SecretString | None = None
 88    """Bearer token for direct authentication (alternative to client credentials)."""
 89
 90    api_root: str = api_util.CLOUD_API_ROOT
 91    """The API root URL. Defaults to Airbyte Cloud API."""
 92
 93    config_api_root: str | None = None
 94    """The Config API root URL."""
 95
 96    def __post_init__(self) -> None:
 97        """Validate credentials and ensure secrets are properly wrapped."""
 98        # Wrap secrets in SecretString if they aren't already
 99        if self.client_id is not None:
100            self.client_id = SecretString(self.client_id)
101        if self.client_secret is not None:
102            self.client_secret = SecretString(self.client_secret)
103        if self.bearer_token is not None:
104            self.bearer_token = SecretString(self.bearer_token)
105
106        # Validate mutual exclusivity
107        has_client_credentials = self.client_id is not None or self.client_secret is not None
108        has_bearer_token = self.bearer_token is not None
109
110        if has_client_credentials and has_bearer_token:
111            raise PyAirbyteInputError(
112                message="Cannot use both client credentials and bearer token authentication.",
113                guidance=(
114                    "Provide either client_id and client_secret together, "
115                    "or bearer_token alone, but not both."
116                ),
117            )
118
119        if has_client_credentials and (self.client_id is None or self.client_secret is None):
120            # If using client credentials, both must be provided
121            raise PyAirbyteInputError(
122                message="Incomplete client credentials.",
123                guidance=(
124                    "When using client credentials authentication, "
125                    "both client_id and client_secret must be provided."
126                ),
127            )
128
129        if not has_client_credentials and not has_bearer_token:
130            raise PyAirbyteInputError(
131                message="No authentication credentials provided.",
132                guidance=(
133                    "Provide either client_id and client_secret together for OAuth2 "
134                    "client credentials flow, or bearer_token for direct authentication."
135                ),
136            )
137
138    @property
139    def uses_bearer_token(self) -> bool:
140        """Return True if using bearer token authentication."""
141        return self.bearer_token is not None
142
143    @property
144    def uses_client_credentials(self) -> bool:
145        """Return True if using client credentials authentication."""
146        return self.client_id is not None and self.client_secret is not None
147
148    @classmethod
149    def from_env(
150        cls,
151        *,
152        api_root: str | None = None,
153        config_api_root: str | None = None,
154    ) -> CloudClientConfig:
155        """Create CloudClientConfig from environment variables.
156
157        This factory method resolves credentials from environment variables,
158        providing a convenient way to create credentials without explicitly
159        passing secrets.
160
161        Environment variables used:
162            - `AIRBYTE_CLOUD_CLIENT_ID`: OAuth client ID (for client credentials flow).
163            - `AIRBYTE_CLOUD_CLIENT_SECRET`: OAuth client secret (for client credentials flow).
164            - `AIRBYTE_CLOUD_BEARER_TOKEN`: Bearer token (alternative to client credentials).
165            - `AIRBYTE_CLOUD_API_URL`: Optional. The API root URL (defaults to Airbyte Cloud).
166            - `AIRBYTE_CLOUD_CONFIG_API_URL`: Optional. The Config API root URL.
167
168        The method will first check for a bearer token. If not found, it will
169        attempt to use client credentials.
170
171        Args:
172            api_root: The API root URL. If not provided, will be resolved from
173                the `AIRBYTE_CLOUD_API_URL` environment variable, or default to
174                the Airbyte Cloud API.
175            config_api_root: The Config API root URL. If not provided, will be resolved
176                from the `AIRBYTE_CLOUD_CONFIG_API_URL` environment variable.
177
178        Returns:
179            A CloudClientConfig instance configured with credentials from the environment.
180
181        Raises:
182            PyAirbyteSecretNotFoundError: If required credentials are not found in
183                the environment.
184        """
185        resolved_api_root = resolve_cloud_api_url(api_root)
186        resolved_config_api_root = resolve_cloud_config_api_url(config_api_root)
187
188        # Try bearer token first
189        bearer_token = resolve_cloud_bearer_token()
190        if bearer_token:
191            return cls(
192                bearer_token=bearer_token,
193                api_root=resolved_api_root,
194                config_api_root=resolved_config_api_root,
195            )
196
197        # Fall back to client credentials
198        return cls(
199            client_id=resolve_cloud_client_id(),
200            client_secret=resolve_cloud_client_secret(),
201            api_root=resolved_api_root,
202            config_api_root=resolved_config_api_root,
203        )

Client configuration for Airbyte Cloud API.

This class encapsulates the authentication and API configuration needed to connect to Airbyte Cloud, OSS, or Enterprise instances. It supports two mutually exclusive authentication methods:

OAuth2 client credentials flow (client_id + client_secret)
Bearer token authentication

Exactly one authentication method must be provided. Providing both or neither will raise a validation error.

Attributes:

client_id: OAuth2 client ID for client credentials flow.
client_secret: OAuth2 client secret for client credentials flow.
bearer_token: Pre-generated bearer token for direct authentication.
api_root: The API root URL. Defaults to Airbyte Cloud API.
config_api_root: The Config API root URL.

CloudClientConfig( client_id: airbyte.secrets.SecretString | None = None, client_secret: airbyte.secrets.SecretString | None = None, bearer_token: airbyte.secrets.SecretString | None = None, api_root: str = 'https://api.airbyte.com/v1', config_api_root: str | None = None)

client_id: airbyte.secrets.SecretString | None = None

OAuth2 client ID for client credentials authentication.

client_secret: airbyte.secrets.SecretString | None = None

OAuth2 client secret for client credentials authentication.

bearer_token: airbyte.secrets.SecretString | None = None

Bearer token for direct authentication (alternative to client credentials).

api_root: str = 'https://api.airbyte.com/v1'

The API root URL. Defaults to Airbyte Cloud API.

config_api_root: str | None = None

The Config API root URL.

uses_bearer_token: bool View Source

138    @property
139    def uses_bearer_token(self) -> bool:
140        """Return True if using bearer token authentication."""
141        return self.bearer_token is not None

Return True if using bearer token authentication.

uses_client_credentials: bool View Source

143    @property
144    def uses_client_credentials(self) -> bool:
145        """Return True if using client credentials authentication."""
146        return self.client_id is not None and self.client_secret is not None

Return True if using client credentials authentication.

@classmethod

def from_env( cls, *, api_root: str | None = None, config_api_root: str | None = None) -> CloudClientConfig: View Source

148    @classmethod
149    def from_env(
150        cls,
151        *,
152        api_root: str | None = None,
153        config_api_root: str | None = None,
154    ) -> CloudClientConfig:
155        """Create CloudClientConfig from environment variables.
156
157        This factory method resolves credentials from environment variables,
158        providing a convenient way to create credentials without explicitly
159        passing secrets.
160
161        Environment variables used:
162            - `AIRBYTE_CLOUD_CLIENT_ID`: OAuth client ID (for client credentials flow).
163            - `AIRBYTE_CLOUD_CLIENT_SECRET`: OAuth client secret (for client credentials flow).
164            - `AIRBYTE_CLOUD_BEARER_TOKEN`: Bearer token (alternative to client credentials).
165            - `AIRBYTE_CLOUD_API_URL`: Optional. The API root URL (defaults to Airbyte Cloud).
166            - `AIRBYTE_CLOUD_CONFIG_API_URL`: Optional. The Config API root URL.
167
168        The method will first check for a bearer token. If not found, it will
169        attempt to use client credentials.
170
171        Args:
172            api_root: The API root URL. If not provided, will be resolved from
173                the `AIRBYTE_CLOUD_API_URL` environment variable, or default to
174                the Airbyte Cloud API.
175            config_api_root: The Config API root URL. If not provided, will be resolved
176                from the `AIRBYTE_CLOUD_CONFIG_API_URL` environment variable.
177
178        Returns:
179            A CloudClientConfig instance configured with credentials from the environment.
180
181        Raises:
182            PyAirbyteSecretNotFoundError: If required credentials are not found in
183                the environment.
184        """
185        resolved_api_root = resolve_cloud_api_url(api_root)
186        resolved_config_api_root = resolve_cloud_config_api_url(config_api_root)
187
188        # Try bearer token first
189        bearer_token = resolve_cloud_bearer_token()
190        if bearer_token:
191            return cls(
192                bearer_token=bearer_token,
193                api_root=resolved_api_root,
194                config_api_root=resolved_config_api_root,
195            )
196
197        # Fall back to client credentials
198        return cls(
199            client_id=resolve_cloud_client_id(),
200            client_secret=resolve_cloud_client_secret(),
201            api_root=resolved_api_root,
202            config_api_root=resolved_config_api_root,
203        )

Create CloudClientConfig from environment variables.

This factory method resolves credentials from environment variables, providing a convenient way to create credentials without explicitly passing secrets.

Environment variables used:

AIRBYTE_CLOUD_CLIENT_ID: OAuth client ID (for client credentials flow).

AIRBYTE_CLOUD_CLIENT_SECRET: OAuth client secret (for client credentials flow).

AIRBYTE_CLOUD_BEARER_TOKEN: Bearer token (alternative to client credentials).

AIRBYTE_CLOUD_API_URL: Optional. The API root URL (defaults to Airbyte Cloud).

AIRBYTE_CLOUD_CONFIG_API_URL: Optional. The Config API root URL.

The method will first check for a bearer token. If not found, it will attempt to use client credentials.

Arguments:

api_root: The API root URL. If not provided, will be resolved from the AIRBYTE_CLOUD_API_URL environment variable, or default to the Airbyte Cloud API.
config_api_root: The Config API root URL. If not provided, will be resolved from the AIRBYTE_CLOUD_CONFIG_API_URL environment variable.

Returns:

A CloudClientConfig instance configured with credentials from the environment.

Raises:

PyAirbyteSecretNotFoundError: If required credentials are not found in the environment.

class CloudDefaultContextInfo(pydantic.main.BaseModel): View Source

192class CloudDefaultContextInfo(BaseModel):
193    """Explicit organization and workspace affinities for the authenticated user."""
194
195    user_id: str | None
196    """The Airbyte user ID, if available."""
197
198    user_name: str | None
199    """The authenticated user's name, if available."""
200
201    user_email: str | None
202    """The authenticated user's email, if available."""
203
204    default_workspace_id: str | None
205    """The resolved default workspace ID, if available."""
206
207    default_workspace_name: str | None
208    """The resolved default workspace name, if available."""
209
210    default_workspace_verified: bool
211    """Whether the resolved default workspace was verified as accessible."""
212
213    unvalidated_workspace_count: int = 0
214    """Number of direct workspace grants not validated due to the validation cap."""
215
216    default_organization_id: str | None
217    """The organization containing the resolved default workspace, if available."""
218
219    default_organization_name: str | None
220    """The name of the organization containing the resolved default workspace, if available."""
221
222    configured_workspace_id: str | None
223    """The explicitly configured workspace ID, if available."""
224
225    configured_organization_id: str | None
226    """The configured organization ID, if available."""
227
228    member_organizations: list[CloudOrganizationInfo]
229    """Organizations identified by explicit organization membership grants."""
230
231    member_workspaces: list[CloudWorkspaceInfo]
232    """Workspaces identified by explicit workspace membership grants."""
233
234    member_organizations_truncated: bool
235    """True if organization memberships beyond the returned list were omitted."""
236
237    member_workspaces_truncated: bool
238    """True if workspace memberships beyond the returned list were omitted."""
239
240    discovery_hints: list[str]
241    """Hints for discovering additional organizations or workspaces."""

Explicit organization and workspace affinities for the authenticated user.

user_id: str | None = PydanticUndefined

The Airbyte user ID, if available.

user_name: str | None = PydanticUndefined

The authenticated user's name, if available.

user_email: str | None = PydanticUndefined

The authenticated user's email, if available.

default_workspace_id: str | None = PydanticUndefined

The resolved default workspace ID, if available.

default_workspace_name: str | None = PydanticUndefined

The resolved default workspace name, if available.

default_workspace_verified: bool = PydanticUndefined

Whether the resolved default workspace was verified as accessible.

unvalidated_workspace_count: int = 0

Number of direct workspace grants not validated due to the validation cap.

default_organization_id: str | None = PydanticUndefined

The organization containing the resolved default workspace, if available.

default_organization_name: str | None = PydanticUndefined

The name of the organization containing the resolved default workspace, if available.

configured_workspace_id: str | None = PydanticUndefined

The explicitly configured workspace ID, if available.

configured_organization_id: str | None = PydanticUndefined

The configured organization ID, if available.

member_organizations: list[airbyte.cloud.models.CloudOrganizationInfo] = PydanticUndefined

Organizations identified by explicit organization membership grants.

member_workspaces: list[CloudWorkspaceInfo] = PydanticUndefined

Workspaces identified by explicit workspace membership grants.

member_organizations_truncated: bool = PydanticUndefined

True if organization memberships beyond the returned list were omitted.

member_workspaces_truncated: bool = PydanticUndefined

True if workspace memberships beyond the returned list were omitted.

discovery_hints: list[str] = PydanticUndefined

Hints for discovering additional organizations or workspaces.

class CloudWorkspaceInfo(pydantic.main.BaseModel): View Source

134class CloudWorkspaceInfo(BaseModel):
135    """Information about an Airbyte workspace."""
136
137    model_config = ConfigDict(populate_by_name=True)
138
139    workspace_id: str = Field(alias="workspaceId")
140    """The workspace ID."""
141
142    name: str
143    """The workspace name."""
144
145    data_residency: str | None = Field(default=None, alias="dataResidency")
146    """The data residency setting for the workspace, if available."""
147
148    organization_id: str | None = Field(default=None, alias="organizationId")
149    """The organization ID for the workspace, if available."""
150
151    organization_name: str | None = Field(default=None, alias="organizationName")
152    """The organization name for the workspace, if available."""
153
154    notifications: dict[str, object | None] | list[dict[str, object | None]] = Field(
155        default_factory=dict
156    )
157    """Workspace notification settings."""
158
159    @classmethod
160    def from_api_response(cls, workspace: _WorkspaceResponseLike) -> CloudWorkspaceInfo:
161        """Create a public model from an internal API workspace response."""
162        return cls(
163            workspace_id=workspace.workspace_id,
164            name=workspace.name,
165            data_residency=workspace.data_residency,
166            organization_id=getattr(workspace, "organization_id", None),
167            notifications=_notifications_to_dict(workspace.notifications),
168        )
169
170    @classmethod
171    def from_mapping(cls, workspace: Mapping[str, object]) -> CloudWorkspaceInfo:
172        """Create a public model from a workspace mapping."""
173        return cls.model_validate(workspace)
174
175    def to_dict(self) -> dict[str, object]:
176        """Return a JSON-serializable dictionary."""
177        return self.model_dump(mode="json")

Information about an Airbyte workspace.

workspace_id: str = PydanticUndefined

The workspace ID.

name: str = PydanticUndefined

The workspace name.

data_residency: str | None = None

The data residency setting for the workspace, if available.

organization_id: str | None = None

The organization ID for the workspace, if available.

organization_name: str | None = None

The organization name for the workspace, if available.

notifications: dict[str, object | None] | list[dict[str, object | None]] = PydanticUndefined

Workspace notification settings.

@classmethod

def from_api_response( cls, workspace: airbyte.cloud.models._WorkspaceResponseLike) -> CloudWorkspaceInfo: View Source

159    @classmethod
160    def from_api_response(cls, workspace: _WorkspaceResponseLike) -> CloudWorkspaceInfo:
161        """Create a public model from an internal API workspace response."""
162        return cls(
163            workspace_id=workspace.workspace_id,
164            name=workspace.name,
165            data_residency=workspace.data_residency,
166            organization_id=getattr(workspace, "organization_id", None),
167            notifications=_notifications_to_dict(workspace.notifications),
168        )

Create a public model from an internal API workspace response.

@classmethod

def from_mapping( cls, workspace: Mapping[str, object]) -> CloudWorkspaceInfo: View Source

170    @classmethod
171    def from_mapping(cls, workspace: Mapping[str, object]) -> CloudWorkspaceInfo:
172        """Create a public model from a workspace mapping."""
173        return cls.model_validate(workspace)

Create a public model from a workspace mapping.

def to_dict(self) -> dict[str, object]: View Source

175    def to_dict(self) -> dict[str, object]:
176        """Return a JSON-serializable dictionary."""
177        return self.model_dump(mode="json")

Return a JSON-serializable dictionary.

class JobStatusEnum(builtins.str, enum.Enum): View Source

 92class JobStatusEnum(str, Enum):
 93    """Status values for an Airbyte Cloud job."""
 94
 95    PENDING = "pending"
 96    RUNNING = "running"
 97    INCOMPLETE = "incomplete"
 98    FAILED = "failed"
 99    SUCCEEDED = "succeeded"
100    CANCELLED = "cancelled"

Status values for an Airbyte Cloud job.

PENDING = <JobStatusEnum.PENDING: 'pending'>

RUNNING = <JobStatusEnum.RUNNING: 'running'>

INCOMPLETE = <JobStatusEnum.INCOMPLETE: 'incomplete'>

FAILED = <JobStatusEnum.FAILED: 'failed'>

SUCCEEDED = <JobStatusEnum.SUCCEEDED: 'succeeded'>

CANCELLED = <JobStatusEnum.CANCELLED: 'cancelled'>

class JobTypeEnum(builtins.str, enum.Enum): View Source

103class JobTypeEnum(str, Enum):
104    """Job type values for Airbyte Cloud jobs."""
105
106    SYNC = "sync"
107    RESET = "reset"
108    REFRESH = "refresh"
109    CLEAR = "clear"

Job type values for Airbyte Cloud jobs.

SYNC = <JobTypeEnum.SYNC: 'sync'>

RESET = <JobTypeEnum.RESET: 'reset'>

REFRESH = <JobTypeEnum.REFRESH: 'refresh'>

CLEAR = <JobTypeEnum.CLEAR: 'clear'>

class WorkspacePrivilegeScope(builtins.str, enum.Enum): View Source

112class WorkspacePrivilegeScope(str, Enum):
113    """How broadly `list_workspaces` searches for workspaces."""
114
115    MEMBER_OF = "member_of"
116    ORGANIZATION_ADMIN = "organization_admin"
117    INSTANCE_ADMIN = "instance_admin"
118    ANY = "any"

How broadly list_workspaces searches for workspaces.

MEMBER_OF = <WorkspacePrivilegeScope.MEMBER_OF: 'member_of'>

ORGANIZATION_ADMIN = <WorkspacePrivilegeScope.ORGANIZATION_ADMIN: 'organization_admin'>

INSTANCE_ADMIN = <WorkspacePrivilegeScope.INSTANCE_ADMIN: 'instance_admin'>

ANY = <WorkspacePrivilegeScope.ANY: 'any'>

airbyte.cloud

Self-managed Airbyte instances

Examples

Basic Sync Example:

Example Read From Cloud Destination:

Example with client credentials:

Example with bearer token:

Environment variables used:

Arguments:

Returns:

Raises:

Example:

Arguments:

Returns:

Raises:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Arguments:

Returns:

Raises:

Arguments:

Returns:

Arguments:

Returns:

Returns:

Raises:

Arguments:

Returns:

Arguments:

Returns:

Arguments:

Returns:

Raises:

Arguments:

Returns:

Arguments:

Raises:

Returns:

Arguments:

Returns:

Arguments:

Arguments:

Returns:

Arguments:

Returns:

Arguments:

Returns:

Returns:

Arguments:

Raises:

Arguments:

Examples:

Arguments:

Attributes:

Environment variables used:

Arguments:

Returns:

Raises: