airbyte_cdk.models.connector_metadata

  1"""Models to represent the structure of a `metadata.yaml` file."""
  2
  3from __future__ import annotations
  4
  5from enum import Enum
  6from pathlib import Path
  7
  8import yaml
  9from pydantic import BaseModel, Field
 10
 11
 12class ConnectorLanguage(str, Enum):
 13    """Connector implementation language."""
 14
 15    PYTHON = "python"
 16    JAVA = "java"
 17    LOW_CODE = "low-code"
 18    MANIFEST_ONLY = "manifest-only"
 19    UNKNOWN = "unknown"
 20
 21
 22class ConnectorBuildOptions(BaseModel):
 23    """Connector build options from metadata.yaml."""
 24
 25    model_config = {"extra": "allow"}
 26
 27    baseImage: str | None = Field(
 28        None,
 29        description="Base image to use for building the connector",
 30    )
 31    path: str | None = Field(
 32        None,
 33        description="Path to the connector code within the repository",
 34    )
 35
 36
 37class SuggestedStreams(BaseModel):
 38    """Suggested streams from metadata.yaml."""
 39
 40    streams: list[str] = Field(
 41        default=[],
 42        description="List of suggested streams for the connector",
 43    )
 44
 45
 46class ConnectorMetadata(BaseModel):
 47    """Connector metadata from metadata.yaml."""
 48
 49    model_config = {"extra": "allow"}
 50
 51    dockerRepository: str = Field(..., description="Docker repository for the connector image")
 52    dockerImageTag: str = Field(..., description="Docker image tag for the connector")
 53
 54    tags: list[str] = Field(
 55        default=[],
 56        description="List of tags for the connector",
 57    )
 58
 59    suggestedStreams: SuggestedStreams | None = Field(
 60        default=None,
 61        description="Suggested streams for the connector",
 62    )
 63
 64    @property
 65    def language(self) -> ConnectorLanguage:
 66        """Get the connector language."""
 67        for tag in self.tags:
 68            if tag.startswith("language:"):
 69                language = tag.split(":", 1)[1]
 70                if language == "python":
 71                    return ConnectorLanguage.PYTHON
 72                elif language == "java":
 73                    return ConnectorLanguage.JAVA
 74                elif language == "low-code":
 75                    return ConnectorLanguage.LOW_CODE
 76                elif language == "manifest-only":
 77                    return ConnectorLanguage.MANIFEST_ONLY
 78
 79        return ConnectorLanguage.UNKNOWN
 80
 81    connectorBuildOptions: ConnectorBuildOptions | None = Field(
 82        None, description="Options for building the connector"
 83    )
 84
 85
 86class MetadataFile(BaseModel):
 87    """Represents the structure of a metadata.yaml file."""
 88
 89    model_config = {"extra": "allow"}
 90
 91    data: ConnectorMetadata = Field(..., description="Connector metadata")
 92
 93    @classmethod
 94    def from_file(
 95        cls,
 96        file_path: Path,
 97    ) -> MetadataFile:
 98        """Load metadata from a YAML file."""
 99        if not file_path.exists():
100            raise FileNotFoundError(f"Metadata file not found: {file_path!s}")
101
102        metadata_content = file_path.read_text()
103        metadata_dict = yaml.safe_load(metadata_content)
104
105        if not metadata_dict or "data" not in metadata_dict:
106            raise ValueError(
107                "Invalid metadata format: missing 'data' field in YAML file '{file_path!s}'"
108            )
109
110        metadata_file = MetadataFile.model_validate(metadata_dict)
111        return metadata_file