airbyte.cloud

 25@dataclass(init=False, kw_only=True)
 26class CloudClient:
 27    """Authenticated client for Airbyte Cloud and self-managed Airbyte APIs."""
 28
 29    _credentials: _AirbyteCredentials
 30
 31    def __init__(
 32        self,
 33        *,
 34        client_id: str | SecretString | None = None,
 35        client_secret: str | SecretString | None = None,
 36        bearer_token: str | SecretString | None = None,
 37        public_api_root: str | None = None,
 38        config_api_root: str | None = None,
 39        workspace_id: str | None = None,
 40        organization_id: str | None = None,
 41    ) -> None:
 42        """Initialize a `CloudClient` from explicit auth values."""
 43        self._credentials = _AirbyteCredentials.from_auth(
 44            client_id=client_id,
 45            client_secret=client_secret,
 46            bearer_token=bearer_token,
 47            public_api_root=public_api_root,
 48            config_api_root=config_api_root,
 49            workspace_id=workspace_id,
 50            organization_id=organization_id,
 51            env_vars=False,
 52        )
 53
 54    @property
 55    def client_id(self) -> SecretString | None:
 56        """OAuth client ID used for authentication."""
 57        return self._credentials.client_id
 58
 59    @property
 60    def client_secret(self) -> SecretString | None:
 61        """OAuth client secret used for authentication."""
 62        return self._credentials.client_secret
 63
 64    @property
 65    def bearer_token(self) -> SecretString | None:
 66        """Bearer token used for authentication."""
 67        return self._credentials.bearer_token
 68
 69    @property
 70    def public_api_root(self) -> str:
 71        """Airbyte Public API root."""
 72        return self._credentials.public_api_root
 73
 74    @property
 75    def config_api_root(self) -> str | None:
 76        """Airbyte Config API root."""
 77        return self._credentials.config_api_root
 78
 79    @property
 80    def organization_id(self) -> str | None:
 81        """Default organization ID for organization-scoped operations."""
 82        return self._credentials.organization_id
 83
 84    @classmethod
 85    def from_auth(
 86        cls,
 87        *,
 88        env_vars: bool = False,
 89        organization_id: str | None = None,
 90        client_id: str | SecretString | None = None,
 91        client_secret: str | SecretString | None = None,
 92        bearer_token: str | SecretString | None = None,
 93        public_api_root: str | None = None,
 94        config_api_root: str | None = None,
 95    ) -> CloudClient:
 96        """Create a client from explicit inputs and optionally environment variables.
 97
 98        When `env_vars` is True, environment variables are checked as a fallback
 99        after any explicitly provided values.
100        """
101        credentials = _AirbyteCredentials.from_auth(
102            organization_id=organization_id,
103            client_id=client_id,
104            client_secret=client_secret,
105            bearer_token=bearer_token,
106            public_api_root=public_api_root,
107            config_api_root=config_api_root,
108            env_vars=env_vars,
109        )
110        return cls._from_credentials(credentials)
111
112    @classmethod
113    def _from_credentials(cls, credentials: _AirbyteCredentials) -> CloudClient:
114        """Create a client from resolved Cloud credentials."""
115        return cls(
116            client_id=credentials.client_id,
117            client_secret=credentials.client_secret,
118            bearer_token=credentials.bearer_token,
119            public_api_root=credentials.public_api_root,
120            config_api_root=credentials.config_api_root,
121            workspace_id=credentials.workspace_id,
122            organization_id=credentials.organization_id,
123        )
124
125    def get_workspace(self, workspace_id: str | None = None) -> CloudWorkspace:
126        """Create a `CloudWorkspace` using this client's credentials."""
127        resolved_workspace_id = workspace_id or self._credentials.workspace_id
128        if not resolved_workspace_id:
129            raise exc.PyAirbyteInputError(
130                message="Workspace ID is required.",
131                guidance="Provide a workspace ID.",
132            )
133
134        credentials = self._credentials.with_workspace_id(resolved_workspace_id)
135        return CloudWorkspace(
136            workspace_id=credentials.workspace_id,
137            client_id=credentials.client_id,
138            client_secret=credentials.client_secret,
139            bearer_token=credentials.bearer_token,
140            api_root=credentials.public_api_root,
141            config_api_root=credentials.config_api_root,
142        )
143
144    def create_workspace(
145        self,
146        *,
147        name: str,
148        organization_id: str | None = None,
149        region_id: str | None = None,
150    ) -> CloudWorkspaceInfo:
151        """Create an Airbyte workspace."""
152        resolved_organization_id = organization_id or self.organization_id
153        workspace = api_util.create_workspace(
154            name=name,
155            organization_id=resolved_organization_id,
156            region_id=region_id,
157            api_root=self.public_api_root,
158            client_id=self.client_id,
159            client_secret=self.client_secret,
160            bearer_token=self.bearer_token,
161        )
162        return CloudWorkspaceInfo.from_api_response(workspace)
163
164    def rename_workspace(
165        self,
166        workspace_id: str,
167        *,
168        name: str,
169    ) -> CloudWorkspaceInfo:
170        """Rename an Airbyte workspace."""
171        workspace = api_util.rename_workspace(
172            workspace_id=workspace_id,
173            name=name,
174            api_root=self.public_api_root,
175            client_id=self.client_id,
176            client_secret=self.client_secret,
177            bearer_token=self.bearer_token,
178        )
179        return CloudWorkspaceInfo.from_api_response(workspace)
180
181    def permanently_delete_workspace(
182        self,
183        workspace_id: str,
184        *,
185        workspace_name: str | None = None,
186        safe_mode: bool = True,
187    ) -> None:
188        """Permanently delete an Airbyte workspace if it has no connections.
189
190        When `safe_mode` is enabled, the workspace name must contain `delete-me`
191        or `deleteme`. This also checks for existing connections before deleting
192        and raises `AirbyteWorkspaceNotEmptyError` if the workspace is not empty.
193        """
194        api_util.permanently_delete_workspace(
195            workspace_id=workspace_id,
196            workspace_name=workspace_name,
197            api_root=self.public_api_root,
198            client_id=self.client_id,
199            client_secret=self.client_secret,
200            bearer_token=self.bearer_token,
201            safe_mode=safe_mode,
202        )
203
204    @overload
205    def list_workspaces(
206        self,
207        name: str | None = None,
208        *,
209        organization_id: None = None,
210        name_contains: str | None = None,
211        name_filter: Callable[[str], bool] | None = None,
212        limit: int | None = None,
213    ) -> list[CloudWorkspaceInfo]:
214        raise NotImplementedError
215
216    @overload
217    def list_workspaces(
218        self,
219        name: str | None = None,
220        *,
221        organization_id: str,
222        name_contains: str | None = None,
223        name_filter: Callable[[str], bool] | None = None,
224        limit: int | None = None,
225    ) -> list[CloudWorkspaceInfo]:
226        raise NotImplementedError
227
228    def list_workspaces(
229        self,
230        name: str | None = None,
231        *,
232        organization_id: str | None = None,
233        name_contains: str | None = None,
234        name_filter: Callable[[str], bool] | None = None,
235        limit: int | None = None,
236    ) -> list[CloudWorkspaceInfo]:
237        """List workspaces available to this client."""
238        if organization_id is not None or self.organization_id is not None:
239            resolved_organization_id = organization_id or self.organization_id
240            if not resolved_organization_id:
241                raise exc.PyAirbyteInputError(
242                    message="Organization ID is required.",
243                    guidance="Provide an organization ID.",
244                )
245            workspaces = api_util.list_workspaces_in_organization(
246                organization_id=resolved_organization_id,
247                api_root=self.public_api_root,
248                config_api_root=self.config_api_root,
249                client_id=self.client_id,
250                client_secret=self.client_secret,
251                bearer_token=self.bearer_token,
252                name_contains=name_contains or name,
253                limit=None if name_filter is not None else limit,
254            )
255            workspace_infos = [
256                CloudWorkspaceInfo.from_mapping(workspace) for workspace in workspaces
257            ]
258            if name_filter is not None:
259                workspace_infos = [
260                    workspace for workspace in workspace_infos if name_filter(workspace.name)
261                ]
262                if limit is not None:
263                    workspace_infos = workspace_infos[:limit]
264            return workspace_infos
265        if name_contains is not None:
266            if name_filter is not None:
267                raise exc.PyAirbyteInputError(
268                    message="You can provide name_contains or name_filter, but not both."
269                )
270            name_substring = name_contains
271
272            def name_filter(workspace_name: str) -> bool:
273                return name_substring in workspace_name
274
275        return [
276            CloudWorkspaceInfo.from_api_response(workspace)
277            for workspace in api_util.list_workspaces(
278                workspace_id="",
279                api_root=self.public_api_root,
280                name=name,
281                name_filter=name_filter,
282                client_id=self.client_id,
283                client_secret=self.client_secret,
284                bearer_token=self.bearer_token,
285                limit=limit,
286            )
287        ]
288
289    def list_organizations(self) -> list[CloudOrganization]:
290        """List all organizations available to this client."""
291        return [
292            CloudOrganization(
293                organization_id=organization.organization_id,
294                organization_name=organization.organization_name,
295                email=organization.email,
296                client_id=self.client_id,
297                client_secret=self.client_secret,
298                bearer_token=self.bearer_token,
299                public_api_root=self.public_api_root,
300                config_api_root=self.config_api_root,
301            )
302            for organization in api_util.list_organizations_for_user(
303                api_root=self.public_api_root,
304                client_id=self.client_id,
305                client_secret=self.client_secret,
306                bearer_token=self.bearer_token,
307            )
308        ]
309
310    def get_organization(
311        self,
312        organization_id: str | None = None,
313        *,
314        organization_name: str | None = None,
315    ) -> CloudOrganization:
316        """Resolve an organization by ID or exact name."""
317        resolved_organization_id = organization_id or self.organization_id
318        if resolved_organization_id and organization_name:
319            raise exc.PyAirbyteInputError(
320                message="Provide either organization ID or organization name."
321            )
322        if not resolved_organization_id and not organization_name:
323            raise exc.PyAirbyteInputError(
324                message="Organization ID or organization name is required."
325            )
326
327        organizations = self.list_organizations()
328        if resolved_organization_id:
329            matching_organizations = [
330                organization
331                for organization in organizations
332                if organization.organization_id == resolved_organization_id
333            ]
334        else:
335            matching_organizations = [
336                organization
337                for organization in organizations
338                if organization.organization_name == organization_name
339            ]
340
341        if not matching_organizations:
342            raise AirbyteMissingResourceError(
343                resource_type="organization",
344                resource_name_or_id=resolved_organization_id or organization_name,
345            )
346        if len(matching_organizations) > 1:
347            raise exc.PyAirbyteInputError(
348                message="Organization name matches multiple organizations.",
349                context={"organization_name": organization_name},
350            )
351
352        return matching_organizations[0]

 18class CloudOrganization:
 19    """Information about an organization in Airbyte Cloud.
 20
 21    This class provides lazy loading of organization attributes including billing status.
 22    It is typically created via `CloudWorkspace.get_organization()`.
 23    """
 24
 25    def __init__(
 26        self,
 27        organization_id: str,
 28        organization_name: str | None = None,
 29        email: str | None = None,
 30        *,
 31        client_id: str | SecretString | None = None,
 32        client_secret: str | SecretString | None = None,
 33        bearer_token: str | SecretString | None = None,
 34        public_api_root: str | None = None,
 35        config_api_root: str | None = None,
 36    ) -> None:
 37        """Initialize a `CloudOrganization`."""
 38        self.organization_id = organization_id
 39        """The organization ID."""
 40
 41        self._organization_name = organization_name
 42        """Display name of the organization."""
 43
 44        self._email = email
 45        """Email associated with the organization."""
 46
 47        self._credentials = _AirbyteCredentials(
 48            client_id=SecretString(client_id) if client_id else None,
 49            client_secret=SecretString(client_secret) if client_secret else None,
 50            bearer_token=SecretString(bearer_token) if bearer_token else None,
 51            public_api_root=public_api_root or api_util.CLOUD_API_ROOT,
 52            config_api_root=config_api_root,
 53            organization_id=organization_id,
 54        )
 55        self._organization_info: dict[str, Any] | None = None
 56        self._organization_info_fetch_failed: bool = False
 57
 58    def _fetch_organization_info(self, *, force_refresh: bool = False) -> dict[str, Any]:
 59        """Fetch and cache organization info including billing status."""
 60        if force_refresh:
 61            self._organization_info_fetch_failed = False
 62
 63        if self._organization_info_fetch_failed and self._organization_info is None:
 64            return {}
 65
 66        if not force_refresh and self._organization_info is not None:
 67            return self._organization_info
 68
 69        try:
 70            self._organization_info = api_util.get_organization_info(
 71                organization_id=self.organization_id,
 72                api_root=self._credentials.public_api_root,
 73                config_api_root=self._credentials.config_api_root,
 74                client_id=self._credentials.client_id,
 75                client_secret=self._credentials.client_secret,
 76                bearer_token=self._credentials.bearer_token,
 77            )
 78        except Exception as ex:
 79            logger.debug("Failed to fetch organization info.", exc_info=ex)
 80            if self._organization_info is None:
 81                self._organization_info_fetch_failed = True
 82            return self._organization_info or {}
 83        else:
 84            return self._organization_info
 85
 86    @property
 87    def organization_name(self) -> str | None:
 88        """Display name of the organization."""
 89        if self._organization_name is not None:
 90            return self._organization_name
 91        info = self._fetch_organization_info()
 92        return info.get("organizationName")
 93
 94    @property
 95    def email(self) -> str | None:
 96        """Email associated with the organization."""
 97        if self._email is not None:
 98            return self._email
 99        info = self._fetch_organization_info()
100        return info.get("email")
101
102    @property
103    def payment_status(self) -> str | None:
104        """Payment status of the organization."""
105        info = self._fetch_organization_info()
106        return (info.get("billing") or {}).get("paymentStatus")
107
108    @property
109    def subscription_status(self) -> str | None:
110        """Subscription status of the organization."""
111        info = self._fetch_organization_info()
112        return (info.get("billing") or {}).get("subscriptionStatus")
113
114    @property
115    def is_account_locked(self) -> bool:
116        """Whether the account is locked due to billing issues."""
117        return api_util.is_account_locked(self.payment_status, self.subscription_status)

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