airbyte_ops_mcp.registry

37@dataclass
38class AuditResult:
39    """Result of auditing which connectors have unpublished versions."""
40
41    unpublished: list[UnpublishedConnector] = field(default_factory=list)
42    checked_count: int = 0
43    skipped_archived: list[str] = field(default_factory=list)
44    skipped_rc: list[str] = field(default_factory=list)
45    skipped_disabled: list[str] = field(default_factory=list)
46    errors: list[str] = field(default_factory=list)

107@dataclass
108class CompileResult:
109    """Result of a registry compile operation."""
110
111    target: str
112    connectors_scanned: int = 0
113    versions_found: int = 0
114    yanked_versions: int = 0
115    latest_updated: int = 0
116    latest_already_current: int = 0
117    cloud_registry_entries: int = 0
118    oss_registry_entries: int = 0
119    composite_registry_entries: int = 0
120    metrics_connector_count: int = 0
121    metrics_registry_entries: int = 0
122    metrics_source: str | None = None
123    metrics_error: str | None = None
124    version_indexes_written: int = 0
125    specs_secrets_mask_properties: int = 0
126    errors: list[str] = field(default_factory=list)
127    dry_run: bool = False
128
129    @property
130    def status(self) -> str:
131        if self.dry_run:
132            return "dry-run"
133        if self.errors:
134            return "completed-with-errors"
135        return "success"
136
137    def summary(self) -> str:
138        return (
139            f"[{self.status}] Scanned {self.connectors_scanned} connectors, "
140            f"{self.versions_found} versions ({self.yanked_versions} yanked). "
141            f"Latest updated: {self.latest_updated}, "
142            f"already current: {self.latest_already_current}. "
143            f"Registry entries: cloud={self.cloud_registry_entries}, "
144            f"oss={self.oss_registry_entries}, "
145            f"composite={self.composite_registry_entries}. "
146            f"Metrics loaded for {self.metrics_connector_count} connectors, "
147            f"injected into {self.metrics_registry_entries} registry entries. "
148            f"Version indexes: {self.version_indexes_written}. "
149            f"Specs secrets mask: {self.specs_secrets_mask_properties} properties. "
150            f"Errors: {len(self.errors)}."
151        )

 88class ConnectorLanguage(StrEnum):
 89    """Connector implementation languages."""
 90
 91    PYTHON = "python"
 92    JAVA = "java"
 93    LOW_CODE = "low-code"
 94    MANIFEST_ONLY = "manifest-only"
 95
 96    @classmethod
 97    def parse(cls, value: str) -> ConnectorLanguage:
 98        """Parse a string into a `ConnectorLanguage`, raising `ValueError` on mismatch."""
 99        try:
100            return cls(value)
101        except ValueError:
102            valid = ", ".join(f"`{m.value}`" for m in cls)
103            raise ValueError(
104                f"Unrecognized language: {value!r}. Expected one of: {valid}."
105            ) from None

73class ConnectorListResult(BaseModel):
74    """Result of listing connectors in the registry."""
75
76    bucket_name: str = Field(description="The GCS bucket name")
77    connector_count: int = Field(description="Number of connectors found")
78    connectors: list[str] = Field(description="List of connector names")

12class ConnectorMetadata(BaseModel):
13    """Connector metadata from metadata.yaml.
14
15    This model represents the essential metadata about a connector
16    read from its metadata.yaml file in the Airbyte monorepo.
17    """
18
19    name: str = Field(description="The connector technical name")
20    docker_repository: str = Field(description="The Docker repository")
21    docker_image_tag: str = Field(description="The Docker image tag/version")
22    support_level: str | None = Field(
23        default=None, description="The support level (certified, community, etc.)"
24    )
25    definition_id: str | None = Field(
26        default=None, description="The connector definition ID"
27    )

70class ConnectorType(StrEnum):
71    """Connector type: source or destination."""
72
73    SOURCE = "source"
74    DESTINATION = "destination"
75
76    @classmethod
77    def parse(cls, value: str) -> ConnectorType:
78        """Parse a string into a `ConnectorType`, raising `ValueError` on mismatch."""
79        try:
80            return cls(value)
81        except ValueError:
82            valid = ", ".join(f"`{m.value}`" for m in cls)
83            raise ValueError(
84                f"Unrecognized connector type: {value!r}. Expected one of: {valid}."
85            ) from None

127@dataclass
128class GenerateResult:
129    """Result of a local artifact generation run."""
130
131    connector_name: str
132    version: str
133    docker_image: str
134    output_dir: str
135    artifacts_written: list[str] = field(default_factory=list)
136    errors: list[str] = field(default_factory=list)
137    validation_errors: list[str] = field(default_factory=list)
138    dry_run: bool = False
139
140    @property
141    def success(self) -> bool:
142        return len(self.errors) == 0 and len(self.validation_errors) == 0
143
144    def to_dict(self) -> dict[str, Any]:
145        return {
146            "connector_name": self.connector_name,
147            "version": self.version,
148            "docker_image": self.docker_image,
149            "output_dir": self.output_dir,
150            "artifacts_written": self.artifacts_written,
151            "errors": self.errors,
152            "validation_errors": self.validation_errors,
153            "dry_run": self.dry_run,
154            "success": self.success,
155        }

30class MetadataPublishResult(BaseModel):
31    """Result of a metadata publish operation to GCS.
32
33    This model provides detailed information about the outcome of
34    publishing connector metadata to the registry.
35    """
36
37    connector_name: str = Field(description="The connector technical name")
38    version: str = Field(description="The version that was published")
39    bucket_name: str = Field(description="The GCS bucket name")
40    versioned_path: str = Field(description="The versioned GCS path")
41    latest_path: str | None = Field(
42        default=None, description="The latest GCS path if updated"
43    )
44    versioned_uploaded: bool = Field(
45        default=False, description="Whether the versioned metadata was uploaded"
46    )
47    latest_uploaded: bool = Field(
48        default=False, description="Whether the latest metadata was uploaded"
49    )
50    status: Literal["success", "dry-run", "already-up-to-date"] = Field(
51        description="The status of the operation"
52    )
53    message: str = Field(description="Status message describing the outcome")
54
55    def __str__(self) -> str:
56        """Return a string representation of the publish result."""
57        return f"[{self.status}] {self.connector_name}:{self.version} -> {self.versioned_path}"

51@dataclass
52class PublishArtifactsResult:
53    """Result of a version-artifacts publish operation."""
54
55    connector_name: str
56    version: str
57    target: str
58    gcs_destination: str
59    files_uploaded: list[str] = field(default_factory=list)
60    errors: list[str] = field(default_factory=list)
61    validation_errors: list[str] = field(default_factory=list)
62    dry_run: bool = False
63
64    @property
65    def success(self) -> bool:
66        return len(self.errors) == 0 and len(self.validation_errors) == 0
67
68    @property
69    def status(self) -> str:
70        if self.dry_run:
71            return "dry-run"
72        if self.errors or self.validation_errors:
73            return "completed-with-errors"
74        return "success"

154@dataclass
155class PurgeLatestResult:
156    """Result of a purge-latest operation."""
157
158    target: str
159    connectors_found: int = 0
160    latest_dirs_deleted: int = 0
161    errors: list[str] = field(default_factory=list)
162    dry_run: bool = False
163
164    @property
165    def status(self) -> str:
166        if self.dry_run:
167            return "dry-run"
168        if self.errors:
169            return "completed-with-errors"
170        return "success"
171
172    def summary(self) -> str:
173        return (
174            f"[{self.status}] Found {self.connectors_found} connectors, "
175            f"deleted {self.latest_dirs_deleted} latest/ directories. "
176            f"Errors: {len(self.errors)}."
177        )

43@dataclass
44class RebuildResult:
45    """Result of a registry rebuild operation."""
46
47    source_bucket: str
48    output_mode: OutputMode
49    output_root: str
50    connectors_processed: int = 0
51    blobs_copied: int = 0
52    blobs_skipped: int = 0
53    errors: list[str] = field(default_factory=list)
54    dry_run: bool = False
55
56    @property
57    def status(self) -> str:
58        """Return the status of the rebuild operation."""
59        if self.dry_run:
60            return "dry-run"
61        if self.errors:
62            return "completed-with-errors"
63        return "success"
64
65    def summary(self) -> str:
66        """Return a human-readable summary."""
67        return (
68            f"[{self.status}] Rebuilt {self.connectors_processed} connectors, "
69            f"{self.blobs_copied} blobs copied, {self.blobs_skipped} skipped, "
70            f"{len(self.errors)} errors. Output: {self.output_root}"
71        )

 40class Registry(ABC):
 41    """A configured connector registry store.
 42
 43    A Registry is bound to a specific `airbyte_ops_mcp.registry.store.RegistryStore`
 44    (store type + env + optional prefix), and provides methods used by the CLI to
 45    read/write registry contents.
 46    """
 47
 48    def __init__(self, store: RegistryStore) -> None:
 49        self.store = store
 50
 51    @property
 52    def store_type(self) -> StoreType:
 53        return self.store.store_type
 54
 55    @property
 56    def bucket_name(self) -> str:
 57        return self.store.bucket
 58
 59    @property
 60    def prefix(self) -> str:
 61        return self.store.prefix
 62
 63    def _require_no_prefix(self, op_name: str) -> None:
 64        """Raise if this op doesn't support prefixed targets."""
 65
 66        if self.prefix:
 67            raise NotImplementedError(
 68                f"Operation '{op_name}' does not yet support store prefixes (got prefix='{self.prefix}')."
 69            )
 70
 71    # ---------------------------------------------------------------------
 72    # Read operations
 73    # ---------------------------------------------------------------------
 74
 75    @abstractmethod
 76    def list_connectors(
 77        self,
 78        *,
 79        support_level: SupportLevel | None = None,
 80        min_support_level: SupportLevel | None = None,
 81        connector_type: ConnectorType | None = None,
 82        language: ConnectorLanguage | None = None,
 83    ) -> list[str]:
 84        raise NotImplementedError(
 85            _op_not_implemented_message(self.store_type, "list_connectors")
 86        )
 87
 88    def list_connector_versions(self, connector_name: str) -> list[str]:
 89        raise NotImplementedError(
 90            _op_not_implemented_message(self.store_type, "list_connector_versions")
 91        )
 92
 93    def get_connector_metadata(
 94        self, connector_name: str, version: str = "latest"
 95    ) -> dict[str, Any]:
 96        raise NotImplementedError(
 97            _op_not_implemented_message(self.store_type, "get_connector_metadata")
 98        )
 99
100    # ---------------------------------------------------------------------
101    # Write / mutate operations
102    # ---------------------------------------------------------------------
103
104    def yank(
105        self,
106        connector_name: str,
107        version: str,
108        reason: str = "",
109        approval_url: str = "",
110        dry_run: bool = False,
111    ) -> YankResult:
112        raise NotImplementedError(_op_not_implemented_message(self.store_type, "yank"))
113
114    def unyank(
115        self,
116        connector_name: str,
117        version: str,
118        dry_run: bool = False,
119    ) -> YankResult:
120        raise NotImplementedError(
121            _op_not_implemented_message(self.store_type, "unyank")
122        )
123
124    def finalize_progressive_rollout_marker(
125        self,
126        connector_name: str,
127        outcome: Literal["promoted", "aborted"],
128        version: str | None = None,
129        dry_run: bool = False,
130    ) -> ProgressiveRolloutMarkerResult:
131        raise NotImplementedError(
132            _op_not_implemented_message(
133                self.store_type,
134                "finalize_progressive_rollout_marker",
135            )
136        )
137
138    def publish_version_artifacts(
139        self,
140        connector_name: str,
141        version: str,
142        artifacts_dir: Path,
143        dry_run: bool = False,
144        with_validate: bool = True,
145    ) -> PublishArtifactsResult:
146        raise NotImplementedError(
147            _op_not_implemented_message(self.store_type, "publish_version_artifacts")
148        )
149
150    def delete_dev_latest(
151        self,
152        connector_name: list[str] | None = None,
153        dry_run: bool = False,
154    ) -> PurgeLatestResult:
155        raise NotImplementedError(
156            _op_not_implemented_message(self.store_type, "delete_dev_latest")
157        )
158
159    def compile(
160        self,
161        connector_name: list[str] | None = None,
162        dry_run: bool = False,
163        with_secrets_mask: bool = False,
164        with_legacy_migration: str | None = None,
165        with_metrics: bool = True,
166        force: bool = False,
167    ) -> CompileResult:
168        raise NotImplementedError(
169            _op_not_implemented_message(self.store_type, "compile")
170        )
171
172    def marketing_stubs_check(self, repo_root: Path) -> dict[str, Any]:
173        raise NotImplementedError(
174            _op_not_implemented_message(self.store_type, "marketing_stubs_check")
175        )
176
177    def marketing_stubs_sync(
178        self,
179        repo_root: Path,
180        dry_run: bool = False,
181    ) -> dict[str, Any]:
182        raise NotImplementedError(
183            _op_not_implemented_message(self.store_type, "marketing_stubs_sync")
184        )
185
186    def mirror(
187        self,
188        output_mode: OutputMode,
189        output_path_root: str | None = None,
190        gcs_bucket: str | None = None,
191        s3_bucket: str | None = None,
192        dry_run: bool = False,
193        connector_name: list[str] | None = None,
194    ) -> RebuildResult:
195        raise NotImplementedError(
196            _op_not_implemented_message(self.store_type, "mirror")
197        )

60class RegistryEntryResult(BaseModel):
61    """Result of reading a registry entry from GCS.
62
63    This model wraps the raw metadata dictionary with additional context.
64    """
65
66    connector_name: str = Field(description="The connector technical name")
67    version: str = Field(description="The version that was read")
68    bucket_name: str = Field(description="The GCS bucket name")
69    gcs_path: str = Field(description="The GCS path that was read")
70    metadata: dict = Field(description="The raw metadata dictionary")

132@dataclass(frozen=True, kw_only=True)
133class RegistryStore:
134    """Parsed store target (type + environment + optional prefix).
135
136    Examples::
137
138        RegistryStore.parse("coral:dev")
139        # -> RegistryStore(store_type=StoreType.CORAL, env="dev", prefix="")
140        RegistryStore.parse("coral:dev/aj-test")
141        # -> RegistryStore(store_type=StoreType.CORAL, env="dev", prefix="aj-test")
142        RegistryStore.parse("sonar:prod")
143        # -> RegistryStore(store_type=StoreType.SONAR, env="prod", prefix="")
144    """
145
146    store_type: StoreType
147    env: str
148    prefix: str = ""
149
150    # -- Derived helpers -----------------------------------------------------
151
152    @property
153    def bucket(self) -> str:
154        """Resolve the concrete bucket name for this target."""
155        env_map = BUCKET_MAP.get(self.store_type)
156        if env_map is None:
157            raise ValueError(f"Unknown store type: {self.store_type!r}")
158        bucket_name = env_map.get(self.env)
159        if bucket_name is None:
160            raise ValueError(
161                f"Unknown environment '{self.env}' for store type '{self.store_type.value}'. "
162                f"Expected one of: {', '.join(sorted(env_map))}."
163            )
164        return bucket_name
165
166    @property
167    def bucket_root(self) -> str:
168        """Bucket name with optional prefix appended (`bucket/prefix`)."""
169        if self.prefix:
170            return f"{self.bucket}/{self.prefix}"
171        return self.bucket
172
173    # -- Parsing -------------------------------------------------------------
174
175    @classmethod
176    def parse(cls, target: str) -> RegistryStore:
177        """Parse a store target string.
178
179        Accepted formats:
180
181            "coral:dev"
182
183            "coral:prod"
184            "coral:dev/aj-test100"
185            "sonar:prod"
186
187        Raises:
188            ValueError: If the string cannot be parsed or references an
189                unknown store type / environment.
190        """
191        if ":" not in target:
192            raise ValueError(
193                f"Invalid store target '{target}'. "
194                "Expected format: '<store_type>:<env>[/<prefix>]' "
195                "(e.g. 'coral:dev', 'sonar:prod', 'coral:dev/my-test')."
196            )
197
198        store_part, env_part = target.split(":", 1)
199
200        # Validate store type
201        store_part_lower = store_part.lower()
202        valid_types = {t.value: t for t in StoreType}
203        if store_part_lower not in valid_types:
204            raise ValueError(
205                f"Unknown store type '{store_part}'. "
206                f"Expected one of: {', '.join(sorted(valid_types))}."
207            )
208        store_type = valid_types[store_part_lower]
209
210        # Split env and prefix
211        env_key, _, prefix = env_part.partition("/")
212        prefix = prefix.strip("/")
213
214        # Validate env
215        env_map = BUCKET_MAP.get(store_type, {})
216        if env_key not in env_map:
217            raise ValueError(
218                f"Unknown environment '{env_key}' for store type '{store_type.value}'. "
219                f"Expected one of: {', '.join(sorted(env_map))}."
220            )
221
222        return cls(store_type=store_type, env=env_key, prefix=prefix)