airbyte.cloud.workspaces

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