airbyte.cloud

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