airbyte_ops_mcp.cloud_admin

16class ConnectorVersionInfo(BaseModel):
17    """Information about a cloud connector's version.
18
19    This model represents the current version state of a deployed connector,
20    including whether a version override (pin) is active.
21    """
22
23    connector_id: str = Field(description="The ID of the deployed connector")
24    connector_type: Literal["source", "destination"] = Field(
25        description="The type of connector (source or destination)"
26    )
27    version: str = Field(description="The current version string (e.g., '0.1.0')")
28    is_version_pinned: bool = Field(
29        description="Whether a version override is active for this connector"
30    )
31
32    def __str__(self) -> str:
33        """Return a string representation of the version."""
34        pinned_suffix = " (pinned)" if self.is_version_pinned else ""
35        return (
36            f"{self.connector_type} {self.connector_id}: {self.version}{pinned_suffix}"
37        )

207class VersionOverrideOperationResult(BaseModel):
208    """Result of a version override operation (set or clear).
209
210    This model provides detailed information about the outcome of a version
211    pinning or unpinning operation.
212    """
213
214    success: bool = Field(description="Whether the operation succeeded")
215    message: str = Field(description="Human-readable message describing the result")
216    connector_id: str = Field(description="The ID of the connector that was modified")
217    connector_type: Literal["source", "destination"] = Field(
218        description="The type of connector (source or destination)"
219    )
220    previous_version: str | None = Field(
221        default=None,
222        description="The version before the operation (None if not available)",
223    )
224    new_version: str | None = Field(
225        default=None,
226        description="The version after the operation (None if cleared or failed)",
227    )
228    was_pinned_before: bool | None = Field(
229        default=None,
230        description="Whether a pin was active before the operation",
231    )
232    is_pinned_after: bool | None = Field(
233        default=None,
234        description="Whether a pin is active after the operation",
235    )
236    customer_tier: str | None = Field(
237        default=None,
238        description="Customer tier of the affected entity (TIER_0, TIER_1, TIER_2, UNKNOWN). "
239        "Included as a guardrail annotation.",
240    )
241    is_eu: bool | None = Field(
242        default=None,
243        description="Whether the affected entity is in the EU region.",
244    )
245    tier_warning: str | None = Field(
246        default=None,
247        description="Warning message if the operation targets a sensitive customer tier.",
248    )
249    warnings: list[str] = Field(
250        default_factory=list,
251        description="Warnings raised by this operation.",
252    )
253
254    def __str__(self) -> str:
255        """Return a string representation of the operation result."""
256        if self.success:
257            return f"✓ {self.message}"
258        return f"✗ {self.message}"