airbyte.cloud

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