diff --git a/airbyte/_util/api_util.py b/airbyte/_util/api_util.py index a6ae76f2..633a5b80 100644 --- a/airbyte/_util/api_util.py +++ b/airbyte/_util/api_util.py @@ -933,3 +933,192 @@ def check_connector( "response": json_result, }, ) + + +def validate_yaml_manifest( + manifest: Any, # noqa: ANN401 + *, + raise_on_error: bool = True, +) -> tuple[bool, str | None]: + """Validate a YAML connector manifest structure. + + Performs basic client-side validation before sending to API. + + Args: + manifest: The manifest to validate (should be a dictionary). + raise_on_error: Whether to raise an exception on validation failure. + + Returns: + Tuple of (is_valid, error_message) + """ + if not isinstance(manifest, dict): + error = "Manifest must be a dictionary" + if raise_on_error: + raise PyAirbyteInputError(message=error, context={"manifest": manifest}) + return False, error + + required_fields = ["version", "type"] + missing = [f for f in required_fields if f not in manifest] + if missing: + error = f"Manifest missing required fields: {', '.join(missing)}" + if raise_on_error: + raise PyAirbyteInputError(message=error, context={"manifest": manifest}) + return False, error + + if manifest.get("type") != "DeclarativeSource": + error = f"Manifest type must be 'DeclarativeSource', got '{manifest.get('type')}'" + if raise_on_error: + raise PyAirbyteInputError(message=error, context={"manifest": manifest}) + return False, error + + return True, None + + +def create_custom_yaml_source_definition( + name: str, + *, + workspace_id: str, + manifest: dict[str, Any], + api_root: str, + client_id: SecretString, + client_secret: SecretString, +) -> models.DeclarativeSourceDefinitionResponse: + """Create a custom YAML source definition.""" + airbyte_instance = get_airbyte_server_instance( + api_root=api_root, + client_id=client_id, + client_secret=client_secret, + ) + + request_body = models.CreateDeclarativeSourceDefinitionRequest( + name=name, + manifest=manifest, + ) + request = api.CreateDeclarativeSourceDefinitionRequest( + workspace_id=workspace_id, + create_declarative_source_definition_request=request_body, + ) + response = airbyte_instance.declarative_source_definitions.create_declarative_source_definition( + request + ) + if response.declarative_source_definition_response is None: + raise AirbyteError( + message="Failed to create custom YAML source definition", + context={"name": name, "workspace_id": workspace_id}, + ) + return response.declarative_source_definition_response + + +def list_custom_yaml_source_definitions( + workspace_id: str, + *, + api_root: str, + client_id: SecretString, + client_secret: SecretString, +) -> list[models.DeclarativeSourceDefinitionResponse]: + """List all custom YAML source definitions in a workspace.""" + airbyte_instance = get_airbyte_server_instance( + api_root=api_root, + client_id=client_id, + client_secret=client_secret, + ) + + request = api.ListDeclarativeSourceDefinitionsRequest( + workspace_id=workspace_id, + ) + response = airbyte_instance.declarative_source_definitions.list_declarative_source_definitions( + request + ) + if response.declarative_source_definitions_response is None: + raise AirbyteError( + message="Failed to list custom YAML source definitions", + context={"workspace_id": workspace_id}, + ) + return response.declarative_source_definitions_response.data + + +def get_custom_yaml_source_definition( + workspace_id: str, + definition_id: str, + *, + api_root: str, + client_id: SecretString, + client_secret: SecretString, +) -> models.DeclarativeSourceDefinitionResponse: + """Get a specific custom YAML source definition.""" + airbyte_instance = get_airbyte_server_instance( + api_root=api_root, + client_id=client_id, + client_secret=client_secret, + ) + + request = api.GetDeclarativeSourceDefinitionRequest( + workspace_id=workspace_id, + definition_id=definition_id, + ) + response = airbyte_instance.declarative_source_definitions.get_declarative_source_definition( + request + ) + if response.declarative_source_definition_response is None: + raise AirbyteError( + message="Failed to get custom YAML source definition", + context={"workspace_id": workspace_id, "definition_id": definition_id}, + ) + return response.declarative_source_definition_response + + +def update_custom_yaml_source_definition( + workspace_id: str, + definition_id: str, + *, + manifest: dict[str, Any], + api_root: str, + client_id: SecretString, + client_secret: SecretString, +) -> models.DeclarativeSourceDefinitionResponse: + """Update a custom YAML source definition.""" + airbyte_instance = get_airbyte_server_instance( + api_root=api_root, + client_id=client_id, + client_secret=client_secret, + ) + + request_body = models.UpdateDeclarativeSourceDefinitionRequest( + manifest=manifest, + ) + request = api.UpdateDeclarativeSourceDefinitionRequest( + workspace_id=workspace_id, + definition_id=definition_id, + update_declarative_source_definition_request=request_body, + ) + response = airbyte_instance.declarative_source_definitions.update_declarative_source_definition( + request + ) + if response.declarative_source_definition_response is None: + raise AirbyteError( + message="Failed to update custom YAML source definition", + context={"workspace_id": workspace_id, "definition_id": definition_id}, + ) + return response.declarative_source_definition_response + + +def delete_custom_yaml_source_definition( + workspace_id: str, + definition_id: str, + *, + api_root: str, + client_id: SecretString, + client_secret: SecretString, +) -> None: + """Delete a custom YAML source definition.""" + airbyte_instance = get_airbyte_server_instance( + api_root=api_root, + client_id=client_id, + client_secret=client_secret, + ) + + request = api.DeleteDeclarativeSourceDefinitionRequest( + workspace_id=workspace_id, + definition_id=definition_id, + ) + airbyte_instance.declarative_source_definitions.delete_declarative_source_definition(request) diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py index 658ce7a6..550336b1 100644 --- a/airbyte/cloud/connectors.py +++ b/airbyte/cloud/connectors.py @@ -41,10 +41,13 @@ import abc from dataclasses import dataclass -from typing import TYPE_CHECKING, ClassVar, Literal +from pathlib import Path +from typing import TYPE_CHECKING, Any, ClassVar, Literal +import yaml from airbyte_api import models as api_models # noqa: TC002 +from airbyte import exceptions as exc from airbyte._util import api_util @@ -253,3 +256,243 @@ def _from_destination_response( ) result._connector_info = destination_response # noqa: SLF001 # Accessing Non-Public API return result + + +class CloudCustomSourceDefinition: + """A custom source connector definition in Airbyte Cloud. + + This represents either a YAML (declarative) or Docker-based custom source definition. + """ + + def __init__( + self, + workspace: CloudWorkspace, + definition_id: str, + connector_type: Literal["yaml", "docker"], + ) -> None: + """Initialize a custom source definition object. + + Note: Only YAML connectors are currently supported. Docker connectors + will raise NotImplementedError. + """ + self.workspace = workspace + self.definition_id = definition_id + self.connector_type = connector_type + self._definition_info: api_models.DeclarativeSourceDefinitionResponse | None = None + + def _fetch_definition_info( + self, + ) -> api_models.DeclarativeSourceDefinitionResponse: + """Fetch definition info from the API.""" + if self.connector_type == "yaml": + return api_util.get_custom_yaml_source_definition( + workspace_id=self.workspace.workspace_id, + definition_id=self.definition_id, + api_root=self.workspace.api_root, + client_id=self.workspace.client_id, + client_secret=self.workspace.client_secret, + ) + raise NotImplementedError( + "Docker custom source definitions are not yet supported. " + "Only YAML manifest-based custom sources are currently available." + ) + + @property + def name(self) -> str: + """Get the display name of the custom connector definition.""" + if not self._definition_info: + self._definition_info = self._fetch_definition_info() + return self._definition_info.name + + @property + def manifest(self) -> dict[str, Any] | None: + """Get the Low-code CDK manifest. Only present for YAML connectors.""" + if self.connector_type != "yaml": + return None + if not self._definition_info: + self._definition_info = self._fetch_definition_info() + return self._definition_info.manifest + + @property + def version(self) -> str | None: + """Get the manifest version. Only present for YAML connectors.""" + if self.connector_type != "yaml": + return None + if not self._definition_info: + self._definition_info = self._fetch_definition_info() + return self._definition_info.version + + @property + def docker_repository(self) -> str | None: + """Get the Docker repository. Only present for Docker connectors. + + Note: Docker connectors are not yet supported and will raise NotImplementedError. + """ + if self.connector_type != "docker": + return None + raise NotImplementedError( + "Docker custom source definitions are not yet supported. " + "Only YAML manifest-based custom sources are currently available." + ) + + @property + def docker_image_tag(self) -> str | None: + """Get the Docker image tag. Only present for Docker connectors. + + Note: Docker connectors are not yet supported and will raise NotImplementedError. + """ + if self.connector_type != "docker": + return None + raise NotImplementedError( + "Docker custom source definitions are not yet supported. " + "Only YAML manifest-based custom sources are currently available." + ) + + @property + def documentation_url(self) -> str | None: + """Get the documentation URL. Only present for Docker connectors. + + Note: Docker connectors are not yet supported and will raise NotImplementedError. + """ + if self.connector_type != "docker": + return None + raise NotImplementedError( + "Docker custom source definitions are not yet supported. " + "Only YAML manifest-based custom sources are currently available." + ) + + @property + def definition_url(self) -> str: + """Get the web URL of the custom source definition.""" + return ( + f"{self.workspace.workspace_url}/settings/custom-connectors/" + f"sources/{self.definition_id}" + ) + + def permanently_delete(self) -> None: + """Permanently delete this custom source definition.""" + self.workspace.permanently_delete_custom_source_definition( + self.definition_id, + custom_connector_type=self.connector_type, + ) + + def update_definition( + self, + *, + manifest_yaml: dict[str, Any] | Path | str | None = None, + docker_tag: str | None = None, + pre_validate: bool = True, + ) -> CloudCustomSourceDefinition: + """Update this custom source definition. + + You must specify EXACTLY ONE of manifest_yaml (for YAML connectors) OR + docker_tag (for Docker connectors), but not both. + + For YAML connectors: updates the manifest + For Docker connectors: Not yet supported (raises NotImplementedError) + + Args: + manifest_yaml: New manifest (YAML connectors only) + docker_tag: New Docker tag (Docker connectors only, not yet supported) + pre_validate: Whether to validate manifest (YAML only) + + Returns: + Updated CloudCustomSourceDefinition object + + Raises: + PyAirbyteInputError: If both or neither parameters are provided + NotImplementedError: If docker_tag is provided (Docker not yet supported) + """ + is_yaml = manifest_yaml is not None + is_docker = docker_tag is not None + + if is_yaml == is_docker: + raise exc.PyAirbyteInputError( + message=( + "Must specify EXACTLY ONE of manifest_yaml (for YAML) OR " + "docker_tag (for Docker), but not both" + ), + context={ + "manifest_yaml_provided": is_yaml, + "docker_tag_provided": is_docker, + }, + ) + + if is_yaml: + manifest_dict: dict[str, Any] + if isinstance(manifest_yaml, Path): + manifest_dict = yaml.safe_load(manifest_yaml.read_text()) + elif isinstance(manifest_yaml, str): + manifest_dict = yaml.safe_load(manifest_yaml) + else: + manifest_dict = manifest_yaml # type: ignore[assignment] + + if pre_validate: + api_util.validate_yaml_manifest(manifest_dict, raise_on_error=True) + + result = api_util.update_custom_yaml_source_definition( + workspace_id=self.workspace.workspace_id, + definition_id=self.definition_id, + manifest=manifest_dict, + api_root=self.workspace.api_root, + client_id=self.workspace.client_id, + client_secret=self.workspace.client_secret, + ) + return CloudCustomSourceDefinition._from_yaml_response(self.workspace, result) + + raise NotImplementedError( + "Docker custom source definitions are not yet supported. " + "Only YAML manifest-based custom sources are currently available." + ) + + def rename( + self, + new_name: str, # noqa: ARG002 + ) -> CloudCustomSourceDefinition: + """Rename this custom source definition. + + Note: Only Docker custom sources can be renamed. YAML custom sources + cannot be renamed as their names are derived from the manifest. + + Args: + new_name: New display name for the connector + + Returns: + Updated CloudCustomSourceDefinition object + + Raises: + PyAirbyteInputError: If attempting to rename a YAML connector + NotImplementedError: If attempting to rename a Docker connector (not yet supported) + """ + if self.connector_type == "yaml": + raise exc.PyAirbyteInputError( + message="Cannot rename YAML custom source definitions", + context={"definition_id": self.definition_id}, + ) + + raise NotImplementedError( + "Docker custom source definitions are not yet supported. " + "Only YAML manifest-based custom sources are currently available." + ) + + def __repr__(self) -> str: + """String representation.""" + return ( + f"CloudCustomSourceDefinition(definition_id={self.definition_id}, " + f"name={self.name}, connector_type={self.connector_type})" + ) + + @classmethod + def _from_yaml_response( + cls, + workspace: CloudWorkspace, + response: api_models.DeclarativeSourceDefinitionResponse, + ) -> CloudCustomSourceDefinition: + """Internal factory method for YAML connectors.""" + result = cls( + workspace=workspace, + definition_id=response.id, + connector_type="yaml", + ) + result._definition_info = response # noqa: SLF001 + return result diff --git a/airbyte/cloud/workspaces.py b/airbyte/cloud/workspaces.py index 621b3d63..c0ccb6f4 100644 --- a/airbyte/cloud/workspaces.py +++ b/airbyte/cloud/workspaces.py @@ -36,13 +36,20 @@ from __future__ import annotations from dataclasses import dataclass -from typing import TYPE_CHECKING, Any +from pathlib import Path +from typing import TYPE_CHECKING, Any, Literal + +import yaml from airbyte import exceptions as exc from airbyte._util import api_util, text_util from airbyte._util.api_util import get_web_url_root from airbyte.cloud.connections import CloudConnection -from airbyte.cloud.connectors import CloudDestination, CloudSource +from airbyte.cloud.connectors import ( + CloudCustomSourceDefinition, + CloudDestination, + CloudSource, +) from airbyte.destinations.base import Destination from airbyte.secrets.base import SecretString @@ -450,3 +457,182 @@ def list_destinations( for destination in destinations if name is None or destination.name == name ] + + def publish_custom_source_definition( + self, + name: str, + *, + manifest_yaml: dict[str, Any] | Path | str | None = None, + docker_image: str | None = None, + docker_tag: str | None = None, + unique: bool = True, + pre_validate: bool = True, + ) -> CloudCustomSourceDefinition: + """Publish a custom source connector definition. + + You must specify EITHER manifest_yaml (for YAML connectors) OR both docker_image + and docker_tag (for Docker connectors), but not both. + + Args: + name: Display name for the connector definition + manifest_yaml: Low-code CDK manifest (dict, Path to YAML file, or YAML string) + docker_image: Docker repository (e.g., 'airbyte/source-custom') + docker_tag: Docker image tag (e.g., '1.0.0') + unique: Whether to enforce name uniqueness + pre_validate: Whether to validate manifest client-side (YAML only) + + Returns: + CloudCustomSourceDefinition object representing the created definition + + Raises: + PyAirbyteInputError: If both or neither of manifest_yaml and docker_image provided + AirbyteDuplicateResourcesError: If unique=True and name already exists + """ + is_yaml = manifest_yaml is not None + is_docker = docker_image is not None + + if is_yaml == is_docker: + raise exc.PyAirbyteInputError( + message=( + "Must specify EITHER manifest_yaml (for YAML connectors) OR " + "docker_image + docker_tag (for Docker connectors), but not both" + ), + context={ + "manifest_yaml_provided": is_yaml, + "docker_image_provided": is_docker, + }, + ) + + if is_docker and docker_tag is None: + raise exc.PyAirbyteInputError( + message="docker_tag is required when docker_image is specified", + context={"docker_image": docker_image}, + ) + + if unique: + existing = self.list_custom_source_definitions( + custom_connector_type="yaml" if is_yaml else "docker", + ) + if any(d.name == name for d in existing): + raise exc.AirbyteDuplicateResourcesError( + resource_type="custom_source_definition", + resource_name=name, + ) + + if is_yaml: + manifest_dict: dict[str, Any] + if isinstance(manifest_yaml, Path): + manifest_dict = yaml.safe_load(manifest_yaml.read_text()) + elif isinstance(manifest_yaml, str): + manifest_dict = yaml.safe_load(manifest_yaml) + elif manifest_yaml is not None: + manifest_dict = manifest_yaml + else: + raise exc.PyAirbyteInputError( + message="manifest_yaml is required for YAML connectors", + context={"name": name}, + ) + + if pre_validate: + api_util.validate_yaml_manifest(manifest_dict, raise_on_error=True) + + result = api_util.create_custom_yaml_source_definition( + name=name, + workspace_id=self.workspace_id, + manifest=manifest_dict, + api_root=self.api_root, + client_id=self.client_id, + client_secret=self.client_secret, + ) + return CloudCustomSourceDefinition._from_yaml_response(self, result) # noqa: SLF001 + + raise NotImplementedError( + "Docker custom source definitions are not yet supported. " + "Only YAML manifest-based custom sources are currently available." + ) + + def list_custom_source_definitions( + self, + *, + custom_connector_type: Literal["yaml", "docker"], + ) -> list[CloudCustomSourceDefinition]: + """List custom source connector definitions. + + Args: + custom_connector_type: Connector type to list ("yaml" or "docker"). Required. + + Returns: + List of CloudCustomSourceDefinition objects matching the specified type + """ + if custom_connector_type == "yaml": + yaml_definitions = api_util.list_custom_yaml_source_definitions( + workspace_id=self.workspace_id, + api_root=self.api_root, + client_id=self.client_id, + client_secret=self.client_secret, + ) + return [ + CloudCustomSourceDefinition._from_yaml_response(self, d) # noqa: SLF001 + for d in yaml_definitions + ] + + raise NotImplementedError( + "Docker custom source definitions are not yet supported. " + "Only YAML manifest-based custom sources are currently available." + ) + + def get_custom_source_definition( + self, + definition_id: str, + *, + custom_connector_type: Literal["yaml", "docker"], + ) -> CloudCustomSourceDefinition: + """Get a specific custom source definition by ID. + + Args: + definition_id: The definition ID + custom_connector_type: Connector type ("yaml" or "docker"). Required. + + Returns: + CloudCustomSourceDefinition object + """ + if custom_connector_type == "yaml": + result = api_util.get_custom_yaml_source_definition( + workspace_id=self.workspace_id, + definition_id=definition_id, + api_root=self.api_root, + client_id=self.client_id, + client_secret=self.client_secret, + ) + return CloudCustomSourceDefinition._from_yaml_response(self, result) # noqa: SLF001 + + raise NotImplementedError( + "Docker custom source definitions are not yet supported. " + "Only YAML manifest-based custom sources are currently available." + ) + + def permanently_delete_custom_source_definition( + self, + definition_id: str, + *, + custom_connector_type: Literal["yaml", "docker"], + ) -> None: + """Permanently delete a custom source definition. + + Args: + definition_id: The definition ID to delete + custom_connector_type: Connector type ("yaml" or "docker"). Required. + """ + if custom_connector_type == "yaml": + api_util.delete_custom_yaml_source_definition( + workspace_id=self.workspace_id, + definition_id=definition_id, + api_root=self.api_root, + client_id=self.client_id, + client_secret=self.client_secret, + ) + else: + raise NotImplementedError( + "Docker custom source definitions are not yet supported. " + "Only YAML manifest-based custom sources are currently available." + ) diff --git a/airbyte/mcp/cloud_ops.py b/airbyte/mcp/cloud_ops.py index ecb43d4b..737a073c 100644 --- a/airbyte/mcp/cloud_ops.py +++ b/airbyte/mcp/cloud_ops.py @@ -1,6 +1,7 @@ # Copyright (c) 2024 Airbyte, Inc., all rights reserved. """Airbyte Cloud MCP operations.""" +from pathlib import Path from typing import Annotated, Any from fastmcp import FastMCP @@ -501,6 +502,126 @@ def list_deployed_cloud_connections() -> list[CloudConnection]: return workspace.list_connections() +def publish_custom_source_definition( + name: Annotated[ + str, + Field(description="The name for the custom connector definition."), + ], + *, + manifest_yaml: Annotated[ + str | Path | None, + Field( + description=( + "The Low-code CDK manifest as a YAML string or file path. " + "Required for YAML connectors." + ), + default=None, + ), + ] = None, + unique: Annotated[ + bool, + Field( + description="Whether to require a unique name.", + default=True, + ), + ] = True, + pre_validate: Annotated[ + bool, + Field( + description="Whether to validate the manifest client-side before publishing.", + default=True, + ), + ] = True, +) -> str: + """Publish a custom YAML source connector definition to Airbyte Cloud. + + Note: Only YAML (declarative) connectors are currently supported. + Docker-based custom sources are not yet available. + """ + try: + workspace: CloudWorkspace = _get_cloud_workspace() + result = workspace.publish_custom_source_definition( + name=name, + manifest_yaml=manifest_yaml, + unique=unique, + pre_validate=pre_validate, + ) + except Exception as ex: + return f"Failed to publish custom source definition '{name}': {ex}" + else: + return ( + f"Successfully published custom YAML source definition '{name}' " + f"with ID '{result.definition_id}' (version {result.version or 'N/A'})" + ) + + +def list_custom_source_definitions() -> list[dict[str, Any]]: + """List custom YAML source definitions in the Airbyte Cloud workspace. + + Note: Only YAML (declarative) connectors are currently supported. + Docker-based custom sources are not yet available. + """ + workspace: CloudWorkspace = _get_cloud_workspace() + definitions = workspace.list_custom_source_definitions( + custom_connector_type="yaml", + ) + + return [ + { + "definition_id": d.definition_id, + "name": d.name, + "connector_type": d.connector_type, + "manifest": d.manifest, + "version": d.version, + } + for d in definitions + ] + + +def update_custom_source_definition( + definition_id: Annotated[ + str, + Field(description="The ID of the definition to update."), + ], + manifest_yaml: Annotated[ + str | Path, + Field( + description="New manifest as YAML string or file path.", + ), + ], + *, + pre_validate: Annotated[ + bool, + Field( + description="Whether to validate the manifest client-side before updating.", + default=True, + ), + ] = True, +) -> str: + """Update a custom YAML source definition in Airbyte Cloud. + + Note: Only YAML (declarative) connectors are currently supported. + Docker-based custom sources are not yet available. + """ + try: + workspace: CloudWorkspace = _get_cloud_workspace() + definition = workspace.get_custom_source_definition( + definition_id=definition_id, + custom_connector_type="yaml", + ) + result = definition.update_definition( + manifest_yaml=manifest_yaml, + pre_validate=pre_validate, + ) + except Exception as ex: + return f"Failed to update custom source definition '{definition_id}': {ex}" + else: + return ( + f"Successfully updated custom YAML source definition. " + f"Name: {result.name}, version: {result.version or 'N/A'}" + ) + + def register_cloud_ops_tools(app: FastMCP) -> None: """@private Register tools with the FastMCP app. @@ -517,3 +638,6 @@ def register_cloud_ops_tools(app: FastMCP) -> None: app.tool(list_deployed_cloud_source_connectors) app.tool(list_deployed_cloud_destination_connectors) app.tool(list_deployed_cloud_connections) + app.tool(publish_custom_source_definition) + app.tool(list_custom_source_definitions) + app.tool(update_custom_source_definition) diff --git a/tests/integration_tests/cloud/test_custom_definitions.py b/tests/integration_tests/cloud/test_custom_definitions.py new file mode 100644 index 00000000..f91a4e45 --- /dev/null +++ b/tests/integration_tests/cloud/test_custom_definitions.py @@ -0,0 +1,120 @@ +# Copyright (c) 2024 Airbyte, Inc., all rights reserved. +"""Tests for custom connector definition publishing.""" + +import pytest + +from airbyte.cloud.workspaces import CloudWorkspace + + +TEST_YAML_MANIFEST = { + "version": "0.1.0", + "type": "DeclarativeSource", + "check": { + "type": "CheckStream", + "stream_names": ["test_stream"], + }, + "definitions": { + "base_requester": { + "type": "HttpRequester", + "url_base": "https://httpbin.org", + }, + }, + "streams": [ + { + "type": "DeclarativeStream", + "name": "test_stream", + "primary_key": ["id"], + "retriever": { + "type": "SimpleRetriever", + "requester": {"$ref": "#/definitions/base_requester", "path": "/get"}, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, + }, + } + ], + "spec": { + "type": "Spec", + "connection_specification": { + "type": "object", + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": {}, + }, + }, +} + + +@pytest.mark.requires_creds +def test_publish_custom_yaml_source( + cloud_workspace: CloudWorkspace, +) -> None: + """Test publishing a custom YAML source definition.""" + from airbyte._util import text_util + + name = f"test-yaml-source-{text_util.generate_random_suffix()}" + + result = cloud_workspace.publish_custom_source_definition( + name=name, + manifest_yaml=TEST_YAML_MANIFEST, + unique=True, + pre_validate=True, + ) + + assert result.definition_id + assert result.name == name + assert result.manifest is not None + assert result.version is not None + assert result.connector_type == "yaml" + + definition_id = result.definition_id + + try: + all_definitions = cloud_workspace.list_custom_source_definitions( + custom_connector_type="yaml", + ) + definitions = [d for d in all_definitions if d.name == name] + assert len(definitions) == 1 + assert definitions[0].definition_id == definition_id + + fetched = cloud_workspace.get_custom_source_definition( + definition_id, + custom_connector_type="yaml", + ) + assert fetched.definition_id == definition_id + assert fetched.name == name + assert fetched.connector_type == "yaml" + + updated_manifest = TEST_YAML_MANIFEST.copy() + updated_manifest["version"] = "0.2.0" + updated = fetched.update_definition( + manifest_yaml=updated_manifest, + ) + assert updated.manifest["version"] == "0.2.0" + + finally: + cloud_workspace.permanently_delete_custom_source_definition( + definition_id, + custom_connector_type="yaml", + ) + + +@pytest.mark.requires_creds +def test_yaml_validation_error( + cloud_workspace: CloudWorkspace, +) -> None: + """Test that validation catches invalid manifests.""" + from airbyte._util import text_util + from airbyte.exceptions import PyAirbyteInputError + + name = f"test-invalid-{text_util.generate_random_suffix()}" + invalid_manifest = {"version": "0.1.0"} + + with pytest.raises(PyAirbyteInputError) as exc_info: + cloud_workspace.publish_custom_source_definition( + name=name, + manifest_yaml=invalid_manifest, + pre_validate=True, + ) + + assert "type" in str(exc_info.value).lower()