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 ConnectorMetadata(BaseModel):
38    """Connector metadata from metadata.yaml."""
39
40    model_config = {"extra": "allow"}
41
42    dockerRepository: str = Field(..., description="Docker repository for the connector image")
43    dockerImageTag: str = Field(..., description="Docker image tag for the connector")
44
45    tags: list[str] = Field(
46        default=[],
47        description="List of tags for the connector",
48    )
49
50    @property
51    def language(self) -> ConnectorLanguage:
52        """Get the connector language."""
53        for tag in self.tags:
54            if tag.startswith("language:"):
55                language = tag.split(":", 1)[1]
56                if language == "python":
57                    return ConnectorLanguage.PYTHON
58                elif language == "java":
59                    return ConnectorLanguage.JAVA
60                elif language == "low-code":
61                    return ConnectorLanguage.LOW_CODE
62                elif language == "manifest-only":
63                    return ConnectorLanguage.MANIFEST_ONLY
64
65        return ConnectorLanguage.UNKNOWN
66
67    connectorBuildOptions: ConnectorBuildOptions | None = Field(
68        None, description="Options for building the connector"
69    )
70
71
72class MetadataFile(BaseModel):
73    """Represents the structure of a metadata.yaml file."""
74
75    model_config = {"extra": "allow"}
76
77    data: ConnectorMetadata = Field(..., description="Connector metadata")
78
79    @classmethod
80    def from_file(
81        cls,
82        file_path: Path,
83    ) -> MetadataFile:
84        """Load metadata from a YAML file."""
85        if not file_path.exists():
86            raise FileNotFoundError(f"Metadata file not found: {file_path!s}")
87
88        metadata_content = file_path.read_text()
89        metadata_dict = yaml.safe_load(metadata_content)
90
91        if not metadata_dict or "data" not in metadata_dict:
92            raise ValueError(
93                "Invalid metadata format: missing 'data' field in YAML file '{file_path!s}'"
94            )
95
96        metadata_file = MetadataFile.model_validate(metadata_dict)
97        return metadata_file