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

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

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="...",
)

111    def __init__(
112        self,
113        *,
114        workspace_id: str | None = None,
115        client_id: str | SecretString | None = None,
116        client_secret: str | SecretString | None = None,
117        api_root: str | None = None,
118        config_api_root: str | None = None,
119        bearer_token: str | SecretString | None = None,
120    ) -> None:
121        """Validate and initialize credentials."""
122        env_vars = not (client_id or client_secret or bearer_token)
123        credentials = _AirbyteCredentials.from_auth(
124            workspace_id=workspace_id,
125            client_id=client_id,
126            client_secret=client_secret,
127            bearer_token=bearer_token,
128            public_api_root=api_root,
129            config_api_root=config_api_root,
130            env_vars=env_vars,
131        )
132        if not credentials.workspace_id:
133            raise exc.PyAirbyteInputError(
134                message="Workspace ID is required.",
135                guidance=(
136                    "Provide a workspace ID, or call `get_default_cloud_context` to discover "
137                    "available workspaces."
138                ),
139            )
140
141        self._credentials = credentials
142        self.workspace_id = credentials.workspace_id or ""
143        self.client_id = credentials.client_id
144        self.client_secret = credentials.client_secret
145        self.bearer_token = credentials.bearer_token
146        self.api_root = credentials.public_api_root
147        self.config_api_root = credentials.config_api_root
148
149        # Create internal CloudClientConfig object (validates mutual exclusivity)
150        self._client_config = CloudClientConfig(
151            client_id=self.client_id,
152            client_secret=self.client_secret,
153            bearer_token=self.bearer_token,
154            api_root=self.api_root,
155            config_api_root=self.config_api_root,
156        )

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

158    @classmethod
159    def from_env(
160        cls,
161        workspace_id: str | None = None,
162        *,
163        api_root: str | None = None,
164        config_api_root: str | None = None,
165    ) -> CloudWorkspace:
166        """Create a CloudWorkspace using credentials from environment variables.
167
168        This factory method resolves credentials from environment variables,
169        providing a convenient way to create a workspace without explicitly
170        passing credentials.
171
172        Two authentication methods are supported (mutually exclusive):
173        1. Bearer token (checked first)
174        2. OAuth2 client credentials (fallback)
175
176        Environment variables used:
177            - `AIRBYTE_CLOUD_BEARER_TOKEN`: Bearer token (alternative to client credentials).
178            - `AIRBYTE_CLOUD_CLIENT_ID`: OAuth client ID (for client credentials flow).
179            - `AIRBYTE_CLOUD_CLIENT_SECRET`: OAuth client secret (for client credentials flow).
180            - `AIRBYTE_CLOUD_WORKSPACE_ID`: The workspace ID (if not passed as argument).
181            - `AIRBYTE_CLOUD_API_URL`: Optional. The API root URL (defaults to Airbyte Cloud).
182            - `AIRBYTE_CLOUD_CONFIG_API_URL`: Optional. The Config API root URL.
183
184        Args:
185            workspace_id: The workspace ID. If not provided, will be resolved from
186                the `AIRBYTE_CLOUD_WORKSPACE_ID` environment variable.
187            api_root: The API root URL. If not provided, will be resolved from
188                the `AIRBYTE_CLOUD_API_URL` environment variable, or default to
189                the Airbyte Cloud API.
190            config_api_root: The Config API root URL. If not provided, will be resolved
191                from the `AIRBYTE_CLOUD_CONFIG_API_URL` environment variable.
192
193        Returns:
194            A CloudWorkspace instance configured with credentials from the environment.
195
196        Raises:
197            PyAirbyteInputError: If required credentials are not found in
198                the environment or are incomplete.
199
200        Example:
201            ```python
202            # With workspace_id from environment
203            workspace = CloudWorkspace.from_env()
204
205            # With explicit workspace_id
206            workspace = CloudWorkspace.from_env(workspace_id="your-workspace-id")
207            ```
208        """
209        return cls(
210            workspace_id=workspace_id,
211            api_root=api_root,
212            config_api_root=config_api_root,
213        )

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

215    @property
216    def workspace_url(self) -> str | None:
217        """The web URL of the workspace."""
218        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

253    def get_organization(
254        self,
255        *,
256        raise_on_error: bool = True,
257    ) -> CloudOrganization | None:
258        """Get the organization this workspace belongs to.
259
260        Fetching organization info requires ORGANIZATION_READER permissions on the organization,
261        which may not be available with workspace-scoped credentials.
262
263        Args:
264            raise_on_error: If True (default), raises AirbyteError on permission or API errors.
265                If False, returns None instead of raising.
266
267        Returns:
268            CloudOrganization object with organization_id and organization_name,
269            or None if raise_on_error=False and an error occurred.
270
271        Raises:
272            AirbyteError: If raise_on_error=True and the organization info cannot be fetched
273                (e.g., due to insufficient permissions or missing data).
274        """
275        try:
276            info = self._organization_info
277        except (AirbyteError, NotImplementedError):
278            if raise_on_error:
279                raise
280            return None
281
282        organization_id = info.get("organizationId")
283        organization_name = info.get("organizationName")
284
285        # Validate that both organization_id and organization_name are non-null and non-empty
286        if not organization_id or not organization_name:
287            if raise_on_error:
288                raise AirbyteError(
289                    message="Organization info is incomplete.",
290                    context={
291                        "organization_id": organization_id,
292                        "organization_name": organization_name,
293                    },
294                )
295            return None
296
297        organization_credentials = self._credentials.with_organization_id(organization_id)
298        return CloudOrganization(
299            organization_id=organization_id,
300            organization_name=organization_name,
301            client_id=organization_credentials.client_id,
302            client_secret=organization_credentials.client_secret,
303            bearer_token=organization_credentials.bearer_token,
304            public_api_root=organization_credentials.public_api_root,
305            config_api_root=organization_credentials.config_api_root,
306        )

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

310    def connect(self) -> None:
311        """Check that the workspace is reachable and raise an exception otherwise.
312
313        Note: It is not necessary to call this method before calling other operations. It
314              serves primarily as a simple check to ensure that the workspace is reachable
315              and credentials are correct.
316        """
317        _ = api_util.get_workspace(
318            api_root=self.api_root,
319            workspace_id=self.workspace_id,
320            client_id=self.client_id,
321            client_secret=self.client_secret,
322            bearer_token=self.bearer_token,
323        )
324        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

328    def get_connection(
329        self,
330        connection_id: str,
331    ) -> CloudConnection:
332        """Get a connection by ID.
333
334        This method does not fetch data from the API. It returns a `CloudConnection` object,
335        which will be loaded lazily as needed.
336        """
337        return CloudConnection(
338            workspace=self,
339            connection_id=connection_id,
340        )

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

342    def get_source(
343        self,
344        source_id: str,
345    ) -> CloudSource:
346        """Get a source by ID.
347
348        This method does not fetch data from the API. It returns a `CloudSource` object,
349        which will be loaded lazily as needed.
350        """
351        return CloudSource(
352            workspace=self,
353            connector_id=source_id,
354        )

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

356    def get_destination(
357        self,
358        destination_id: str,
359    ) -> CloudDestination:
360        """Get a destination by ID.
361
362        This method does not fetch data from the API. It returns a `CloudDestination` object,
363        which will be loaded lazily as needed.
364        """
365        return CloudDestination(
366            workspace=self,
367            connector_id=destination_id,
368        )

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 deploy_source( self, name: str, source: airbyte.Source, *, unique: bool = True, random_name_suffix: bool = False) -> airbyte.cloud.connectors.CloudSource: View Source

372    def deploy_source(
373        self,
374        name: str,
375        source: Source,
376        *,
377        unique: bool = True,
378        random_name_suffix: bool = False,
379    ) -> CloudSource:
380        """Deploy a source to the workspace.
381
382        Returns the newly deployed source.
383
384        Args:
385            name: The name to use when deploying.
386            source: The source object to deploy.
387            unique: Whether to require a unique name. If `True`, duplicate names
388                are not allowed. Defaults to `True`.
389            random_name_suffix: Whether to append a random suffix to the name.
390        """
391        source_config_dict = source._hydrated_config.copy()  # noqa: SLF001 (non-public API)
392        source_config_dict["sourceType"] = source.name.replace("source-", "")
393
394        if random_name_suffix:
395            name += f" (ID: {text_util.generate_random_suffix()})"
396
397        if unique:
398            existing = self.list_sources(name=name)
399            if existing:
400                raise exc.AirbyteDuplicateResourcesError(
401                    resource_type="source",
402                    resource_name=name,
403                )
404
405        deployed_source = api_util.create_source(
406            name=name,
407            api_root=self.api_root,
408            workspace_id=self.workspace_id,
409            config=source_config_dict,
410            client_id=self.client_id,
411            client_secret=self.client_secret,
412            bearer_token=self.bearer_token,
413        )
414        return CloudSource(
415            workspace=self,
416            connector_id=deployed_source.source_id,
417        )

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.
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.

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

419    def deploy_destination(
420        self,
421        name: str,
422        destination: Destination | dict[str, Any],
423        *,
424        unique: bool = True,
425        random_name_suffix: bool = False,
426    ) -> CloudDestination:
427        """Deploy a destination to the workspace.
428
429        Returns the newly deployed destination ID.
430
431        Args:
432            name: The name to use when deploying.
433            destination: The destination to deploy. Can be a local Airbyte `Destination` object or a
434                dictionary of configuration values.
435            unique: Whether to require a unique name. If `True`, duplicate names
436                are not allowed. Defaults to `True`.
437            random_name_suffix: Whether to append a random suffix to the name.
438        """
439        if isinstance(destination, Destination):
440            destination_conf_dict = destination._hydrated_config.copy()  # noqa: SLF001 (non-public API)
441            destination_conf_dict["destinationType"] = destination.name.replace("destination-", "")
442            # raise ValueError(destination_conf_dict)
443        else:
444            destination_conf_dict = destination.copy()
445            if "destinationType" not in destination_conf_dict:
446                raise exc.PyAirbyteInputError(
447                    message="Missing `destinationType` in configuration dictionary.",
448                )
449
450        if random_name_suffix:
451            name += f" (ID: {text_util.generate_random_suffix()})"
452
453        if unique:
454            existing = self.list_destinations(name=name)
455            if existing:
456                raise exc.AirbyteDuplicateResourcesError(
457                    resource_type="destination",
458                    resource_name=name,
459                )
460
461        deployed_destination = api_util.create_destination(
462            name=name,
463            api_root=self.api_root,
464            workspace_id=self.workspace_id,
465            config=destination_conf_dict,  # Wants a dataclass but accepts dict
466            client_id=self.client_id,
467            client_secret=self.client_secret,
468            bearer_token=self.bearer_token,
469        )
470        return CloudDestination(
471            workspace=self,
472            connector_id=deployed_destination.destination_id,
473        )

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.

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

475    def permanently_delete_source(
476        self,
477        source: str | CloudSource,
478        *,
479        safe_mode: bool = True,
480    ) -> None:
481        """Delete a source from the workspace.
482
483        You can pass either the source ID `str` or a deployed `Source` object.
484
485        Args:
486            source: The source ID or CloudSource object to delete
487            safe_mode: If True, requires the source name to contain "delete-me" or "deleteme"
488                (case insensitive) to prevent accidental deletion. Defaults to True.
489        """
490        if not isinstance(source, (str, CloudSource)):
491            raise exc.PyAirbyteInputError(
492                message="Invalid source type.",
493                input_value=type(source).__name__,
494            )
495
496        api_util.delete_source(
497            source_id=source.connector_id if isinstance(source, CloudSource) else source,
498            source_name=source.name if isinstance(source, CloudSource) else None,
499            api_root=self.api_root,
500            client_id=self.client_id,
501            client_secret=self.client_secret,
502            bearer_token=self.bearer_token,
503            safe_mode=safe_mode,
504        )

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

508    def permanently_delete_destination(
509        self,
510        destination: str | CloudDestination,
511        *,
512        safe_mode: bool = True,
513    ) -> None:
514        """Delete a deployed destination from the workspace.
515
516        You can pass either the `Cache` class or the deployed destination ID as a `str`.
517
518        Args:
519            destination: The destination ID or CloudDestination object to delete
520            safe_mode: If True, requires the destination name to contain "delete-me" or "deleteme"
521                (case insensitive) to prevent accidental deletion. Defaults to True.
522        """
523        if not isinstance(destination, (str, CloudDestination)):
524            raise exc.PyAirbyteInputError(
525                message="Invalid destination type.",
526                input_value=type(destination).__name__,
527            )
528
529        api_util.delete_destination(
530            destination_id=(
531                destination if isinstance(destination, str) else destination.destination_id
532            ),
533            destination_name=(
534                destination.name if isinstance(destination, CloudDestination) else None
535            ),
536            api_root=self.api_root,
537            client_id=self.client_id,
538            client_secret=self.client_secret,
539            bearer_token=self.bearer_token,
540            safe_mode=safe_mode,
541        )

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

545    def deploy_connection(
546        self,
547        connection_name: str,
548        *,
549        source: CloudSource | str,
550        selected_streams: list[str],
551        destination: CloudDestination | str,
552        table_prefix: str | None = None,
553    ) -> CloudConnection:
554        """Create a new connection between an already deployed source and destination.
555
556        Returns the newly deployed connection object.
557
558        Args:
559            connection_name: The name of the connection.
560            source: The deployed source. You can pass a source ID or a CloudSource object.
561            destination: The deployed destination. You can pass a destination ID or a
562                CloudDestination object.
563            table_prefix: Optional. The table prefix to use when syncing to the destination.
564            selected_streams: The selected stream names to sync within the connection.
565        """
566        if not selected_streams:
567            raise exc.PyAirbyteInputError(
568                guidance="You must provide `selected_streams` when creating a connection."
569            )
570
571        source_id: str = source if isinstance(source, str) else source.connector_id
572        destination_id: str = (
573            destination if isinstance(destination, str) else destination.connector_id
574        )
575
576        deployed_connection = api_util.create_connection(
577            name=connection_name,
578            source_id=source_id,
579            destination_id=destination_id,
580            api_root=self.api_root,
581            workspace_id=self.workspace_id,
582            selected_stream_names=selected_streams,
583            prefix=table_prefix or "",
584            client_id=self.client_id,
585            client_secret=self.client_secret,
586            bearer_token=self.bearer_token,
587        )
588
589        return CloudConnection(
590            workspace=self,
591            connection_id=deployed_connection.connection_id,
592            source=deployed_connection.source_id,
593            destination=deployed_connection.destination_id,
594        )

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

596    def permanently_delete_connection(
597        self,
598        connection: str | CloudConnection,
599        *,
600        cascade_delete_source: bool = False,
601        cascade_delete_destination: bool = False,
602        safe_mode: bool = True,
603    ) -> None:
604        """Delete a deployed connection from the workspace.
605
606        Args:
607            connection: The connection ID or CloudConnection object to delete
608            cascade_delete_source: If True, also delete the source after deleting the connection
609            cascade_delete_destination: If True, also delete the destination after deleting
610                the connection
611            safe_mode: If True, requires the connection name to contain "delete-me" or "deleteme"
612                (case insensitive) to prevent accidental deletion. Defaults to True. Also applies
613                to cascade deletes.
614        """
615        if connection is None:
616            raise ValueError("No connection ID provided.")
617
618        if isinstance(connection, str):
619            connection = CloudConnection(
620                workspace=self,
621                connection_id=connection,
622            )
623
624        api_util.delete_connection(
625            connection_id=connection.connection_id,
626            connection_name=connection.name,
627            api_root=self.api_root,
628            workspace_id=self.workspace_id,
629            client_id=self.client_id,
630            client_secret=self.client_secret,
631            bearer_token=self.bearer_token,
632            safe_mode=safe_mode,
633        )
634
635        if cascade_delete_source:
636            self.permanently_delete_source(
637                source=connection.source_id,
638                safe_mode=safe_mode,
639            )
640        if cascade_delete_destination:
641            self.permanently_delete_destination(
642                destination=connection.destination_id,
643                safe_mode=safe_mode,
644            )

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

648    def list_workspaces(
649        self,
650        name: str | None = None,
651        *,
652        name_filter: Callable | None = None,
653        limit: int | None = None,
654    ) -> list[CloudWorkspaceInfo]:
655        """List workspaces available to the current credentials, with an optional limit."""
656        return [
657            CloudWorkspaceInfo.from_api_response(workspace)
658            for workspace in api_util.list_workspaces(
659                workspace_id="",
660                api_root=self.api_root,
661                name=name,
662                name_filter=name_filter,
663                client_id=self.client_id,
664                client_secret=self.client_secret,
665                bearer_token=self.bearer_token,
666                limit=limit,
667            )
668        ]

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

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

670    def rename(
671        self,
672        name: str,
673    ) -> CloudWorkspace:
674        """Rename this workspace."""
675        api_util.rename_workspace(
676            workspace_id=self.workspace_id,
677            name=name,
678            api_root=self.api_root,
679            client_id=self.client_id,
680            client_secret=self.client_secret,
681            bearer_token=self.bearer_token,
682        )
683        return self

Rename this workspace.

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

685    def permanently_delete(
686        self,
687        *,
688        workspace_name: str | None = None,
689        safe_mode: bool = True,
690    ) -> None:
691        """Permanently delete this workspace if it has no connections.
692
693        When `safe_mode` is enabled, the workspace name must contain `delete-me`
694        or `deleteme`. This also checks for existing connections before deleting
695        and raises `AirbyteWorkspaceNotEmptyError` if the workspace is not empty.
696        """
697        api_util.permanently_delete_workspace(
698            workspace_id=self.workspace_id,
699            workspace_name=workspace_name,
700            api_root=self.api_root,
701            client_id=self.client_id,
702            client_secret=self.client_secret,
703            bearer_token=self.bearer_token,
704            safe_mode=safe_mode,
705        )

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

707    def list_connections(
708        self,
709        name: str | None = None,
710        *,
711        name_filter: Callable | None = None,
712        limit: int | None = None,
713    ) -> list[CloudConnection]:
714        """List connections by name in the workspace, with an optional limit."""
715        connections = api_util.list_connections(
716            api_root=self.api_root,
717            workspace_id=self.workspace_id,
718            name=name,
719            name_filter=name_filter,
720            limit=limit,
721            client_id=self.client_id,
722            client_secret=self.client_secret,
723            bearer_token=self.bearer_token,
724        )
725        return [
726            CloudConnection._from_connection_response(  # noqa: SLF001 (non-public API)
727                workspace=self,
728                connection_response=connection,
729            )
730            for connection in connections
731        ]

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

733    def list_sources(
734        self,
735        name: str | None = None,
736        *,
737        name_filter: Callable | None = None,
738        limit: int | None = None,
739    ) -> list[CloudSource]:
740        """List all sources in the workspace, with an optional limit."""
741        sources = api_util.list_sources(
742            api_root=self.api_root,
743            workspace_id=self.workspace_id,
744            name=name,
745            name_filter=name_filter,
746            limit=limit,
747            client_id=self.client_id,
748            client_secret=self.client_secret,
749            bearer_token=self.bearer_token,
750        )
751        return [
752            CloudSource._from_source_response(  # noqa: SLF001 (non-public API)
753                workspace=self,
754                source_response=source,
755            )
756            for source in sources
757        ]

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

759    def list_destinations(
760        self,
761        name: str | None = None,
762        *,
763        name_filter: Callable | None = None,
764        limit: int | None = None,
765    ) -> list[CloudDestination]:
766        """List all destinations in the workspace, with an optional limit."""
767        destinations = api_util.list_destinations(
768            api_root=self.api_root,
769            workspace_id=self.workspace_id,
770            name=name,
771            name_filter=name_filter,
772            limit=limit,
773            client_id=self.client_id,
774            client_secret=self.client_secret,
775            bearer_token=self.bearer_token,
776        )
777        return [
778            CloudDestination._from_destination_response(  # noqa: SLF001 (non-public API)
779                workspace=self,
780                destination_response=destination,
781            )
782            for destination in destinations
783        ]

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

785    def publish_custom_source_definition(
786        self,
787        name: str,
788        *,
789        manifest_yaml: dict[str, Any] | Path | str | None = None,
790        docker_image: str | None = None,
791        docker_tag: str | None = None,
792        unique: bool = True,
793        pre_validate: bool = True,
794        testing_values: dict[str, Any] | None = None,
795    ) -> CustomCloudSourceDefinition:
796        """Publish a custom source connector definition.
797
798        You must specify EITHER manifest_yaml (for YAML connectors) OR both docker_image
799        and docker_tag (for Docker connectors), but not both.
800
801        Args:
802            name: Display name for the connector definition
803            manifest_yaml: Low-code CDK manifest (dict, Path to YAML file, or YAML string)
804            docker_image: Docker repository (e.g., 'airbyte/source-custom')
805            docker_tag: Docker image tag (e.g., '1.0.0')
806            unique: Whether to enforce name uniqueness
807            pre_validate: Whether to validate manifest client-side (YAML only)
808            testing_values: Optional configuration values to use for testing in the
809                Connector Builder UI. If provided, these values are stored as the complete
810                testing values object for the connector builder project (replaces any existing
811                values), allowing immediate test read operations.
812
813        Returns:
814            CustomCloudSourceDefinition object representing the created definition
815
816        Raises:
817            PyAirbyteInputError: If both or neither of manifest_yaml and docker_image provided
818            AirbyteDuplicateResourcesError: If unique=True and name already exists
819        """
820        is_yaml = manifest_yaml is not None
821        is_docker = docker_image is not None
822
823        if is_yaml == is_docker:
824            raise exc.PyAirbyteInputError(
825                message=(
826                    "Must specify EITHER manifest_yaml (for YAML connectors) OR "
827                    "docker_image + docker_tag (for Docker connectors), but not both"
828                ),
829                context={
830                    "manifest_yaml_provided": is_yaml,
831                    "docker_image_provided": is_docker,
832                },
833            )
834
835        if is_docker and docker_tag is None:
836            raise exc.PyAirbyteInputError(
837                message="docker_tag is required when docker_image is specified",
838                context={"docker_image": docker_image},
839            )
840
841        if unique:
842            existing = self.list_custom_source_definitions(
843                definition_type="yaml" if is_yaml else "docker",
844            )
845            if any(d.name == name for d in existing):
846                raise exc.AirbyteDuplicateResourcesError(
847                    resource_type="custom_source_definition",
848                    resource_name=name,
849                )
850
851        if is_yaml:
852            manifest_dict: dict[str, Any]
853            if isinstance(manifest_yaml, Path):
854                manifest_dict = yaml.safe_load(manifest_yaml.read_text())
855            elif isinstance(manifest_yaml, str):
856                manifest_dict = yaml.safe_load(manifest_yaml)
857            elif manifest_yaml is not None:
858                manifest_dict = manifest_yaml
859            else:
860                raise exc.PyAirbyteInputError(
861                    message="manifest_yaml is required for YAML connectors",
862                    context={"name": name},
863                )
864
865            if pre_validate:
866                api_util.validate_yaml_manifest(manifest_dict, raise_on_error=True)
867
868            result = api_util.create_custom_yaml_source_definition(
869                name=name,
870                workspace_id=self.workspace_id,
871                manifest=manifest_dict,
872                api_root=self.api_root,
873                client_id=self.client_id,
874                client_secret=self.client_secret,
875                bearer_token=self.bearer_token,
876            )
877            custom_definition = CustomCloudSourceDefinition._from_yaml_response(  # noqa: SLF001
878                self, result
879            )
880
881            # Set testing values if provided
882            if testing_values is not None:
883                custom_definition.set_testing_values(testing_values)
884
885            return custom_definition
886
887        raise NotImplementedError(
888            "Docker custom source definitions are not yet supported. "
889            "Only YAML manifest-based custom sources are currently available."
890        )

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

892    def list_custom_source_definitions(
893        self,
894        *,
895        definition_type: Literal["yaml", "docker"],
896    ) -> list[CustomCloudSourceDefinition]:
897        """List custom source connector definitions.
898
899        Args:
900            definition_type: Connector type to list ("yaml" or "docker"). Required.
901
902        Returns:
903            List of CustomCloudSourceDefinition objects matching the specified type
904        """
905        if definition_type == "yaml":
906            yaml_definitions = api_util.list_custom_yaml_source_definitions(
907                workspace_id=self.workspace_id,
908                api_root=self.api_root,
909                client_id=self.client_id,
910                client_secret=self.client_secret,
911                bearer_token=self.bearer_token,
912            )
913            return [
914                CustomCloudSourceDefinition._from_yaml_response(self, d)  # noqa: SLF001
915                for d in yaml_definitions
916            ]
917
918        raise NotImplementedError(
919            "Docker custom source definitions are not yet supported. "
920            "Only YAML manifest-based custom sources are currently available."
921        )

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

923    def get_custom_source_definition(
924        self,
925        definition_id: str,
926        *,
927        definition_type: Literal["yaml", "docker"],
928    ) -> CustomCloudSourceDefinition:
929        """Get a specific custom source definition by ID.
930
931        Args:
932            definition_id: The definition ID
933            definition_type: Connector type ("yaml" or "docker"). Required.
934
935        Returns:
936            CustomCloudSourceDefinition object
937        """
938        if definition_type == "yaml":
939            result = api_util.get_custom_yaml_source_definition(
940                workspace_id=self.workspace_id,
941                definition_id=definition_id,
942                api_root=self.api_root,
943                client_id=self.client_id,
944                client_secret=self.client_secret,
945                bearer_token=self.bearer_token,
946            )
947            return CustomCloudSourceDefinition._from_yaml_response(self, result)  # noqa: SLF001
948
949        raise NotImplementedError(
950            "Docker custom source definitions are not yet supported. "
951            "Only YAML manifest-based custom sources are currently available."
952        )

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

160class CloudDefaultContextInfo(BaseModel):
161    """Explicit organization and workspace affinities for the authenticated user."""
162
163    user_id: str | None
164    """The Airbyte user ID, if available."""
165
166    user_name: str | None
167    """The authenticated user's name, if available."""
168
169    user_email: str | None
170    """The authenticated user's email, if available."""
171
172    default_workspace_id: str | None
173    """The resolved default workspace ID, if available."""
174
175    default_workspace_name: str | None
176    """The resolved default workspace name, if available."""
177
178    default_workspace_verified: bool
179    """Whether the resolved default workspace was verified as accessible."""
180
181    unvalidated_workspace_count: int = 0
182    """Number of direct workspace grants not validated due to the validation cap."""
183
184    default_organization_id: str | None
185    """The organization containing the resolved default workspace, if available."""
186
187    default_organization_name: str | None
188    """The name of the organization containing the resolved default workspace, if available."""
189
190    configured_workspace_id: str | None
191    """The explicitly configured workspace ID, if available."""
192
193    configured_organization_id: str | None
194    """The configured organization ID, if available."""
195
196    member_organizations: list[CloudOrganizationInfo]
197    """Organizations identified by explicit organization membership grants."""
198
199    member_workspaces: list[CloudWorkspaceInfo]
200    """Workspaces identified by explicit workspace membership grants."""
201
202    member_organizations_truncated: bool
203    """True if organization memberships beyond the returned list were omitted."""
204
205    member_workspaces_truncated: bool
206    """True if workspace memberships beyond the returned list were omitted."""
207
208    discovery_hints: list[str]
209    """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

102class CloudWorkspaceInfo(BaseModel):
103    """Information about an Airbyte workspace."""
104
105    model_config = ConfigDict(populate_by_name=True)
106
107    workspace_id: str = Field(alias="workspaceId")
108    """The workspace ID."""
109
110    name: str
111    """The workspace name."""
112
113    data_residency: str | None = Field(default=None, alias="dataResidency")
114    """The data residency setting for the workspace, if available."""
115
116    organization_id: str | None = Field(default=None, alias="organizationId")
117    """The organization ID for the workspace, if available."""
118
119    organization_name: str | None = Field(default=None, alias="organizationName")
120    """The organization name for the workspace, if available."""
121
122    notifications: dict[str, object | None] | list[dict[str, object | None]] = Field(
123        default_factory=dict
124    )
125    """Workspace notification settings."""
126
127    @classmethod
128    def from_api_response(cls, workspace: _WorkspaceResponseLike) -> CloudWorkspaceInfo:
129        """Create a public model from an internal API workspace response."""
130        return cls(
131            workspace_id=workspace.workspace_id,
132            name=workspace.name,
133            data_residency=workspace.data_residency,
134            organization_id=getattr(workspace, "organization_id", None),
135            notifications=_notifications_to_dict(workspace.notifications),
136        )
137
138    @classmethod
139    def from_mapping(cls, workspace: Mapping[str, object]) -> CloudWorkspaceInfo:
140        """Create a public model from a workspace mapping."""
141        return cls.model_validate(workspace)
142
143    def to_dict(self) -> dict[str, object]:
144        """Return a JSON-serializable dictionary."""
145        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

127    @classmethod
128    def from_api_response(cls, workspace: _WorkspaceResponseLike) -> CloudWorkspaceInfo:
129        """Create a public model from an internal API workspace response."""
130        return cls(
131            workspace_id=workspace.workspace_id,
132            name=workspace.name,
133            data_residency=workspace.data_residency,
134            organization_id=getattr(workspace, "organization_id", None),
135            notifications=_notifications_to_dict(workspace.notifications),
136        )

Create a public model from an internal API workspace response.

@classmethod

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

138    @classmethod
139    def from_mapping(cls, workspace: Mapping[str, object]) -> CloudWorkspaceInfo:
140        """Create a public model from a workspace mapping."""
141        return cls.model_validate(workspace)

Create a public model from a workspace mapping.

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

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

Return a JSON-serializable dictionary.

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

60class JobStatusEnum(str, Enum):
61    """Status values for an Airbyte Cloud job."""
62
63    PENDING = "pending"
64    RUNNING = "running"
65    INCOMPLETE = "incomplete"
66    FAILED = "failed"
67    SUCCEEDED = "succeeded"
68    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

71class JobTypeEnum(str, Enum):
72    """Job type values for Airbyte Cloud jobs."""
73
74    SYNC = "sync"
75    RESET = "reset"
76    REFRESH = "refresh"
77    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

80class WorkspacePrivilegeScope(str, Enum):
81    """How broadly `list_workspaces` searches for workspaces."""
82
83    MEMBER_OF = "member_of"
84    ORGANIZATION_ADMIN = "organization_admin"
85    INSTANCE_ADMIN = "instance_admin"
86    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: