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        )

173class VersionOverrideOperationResult(BaseModel):
174    """Result of a version override operation (set or clear).
175
176    This model provides detailed information about the outcome of a version
177    pinning or unpinning operation.
178    """
179
180    success: bool = Field(description="Whether the operation succeeded")
181    message: str = Field(description="Human-readable message describing the result")
182    connector_id: str = Field(description="The ID of the connector that was modified")
183    connector_type: Literal["source", "destination"] = Field(
184        description="The type of connector (source or destination)"
185    )
186    previous_version: str | None = Field(
187        default=None,
188        description="The version before the operation (None if not available)",
189    )
190    new_version: str | None = Field(
191        default=None,
192        description="The version after the operation (None if cleared or failed)",
193    )
194    was_pinned_before: bool | None = Field(
195        default=None,
196        description="Whether a pin was active before the operation",
197    )
198    is_pinned_after: bool | None = Field(
199        default=None,
200        description="Whether a pin is active after the operation",
201    )
202    customer_tier: str | None = Field(
203        default=None,
204        description="Customer tier of the affected entity (TIER_0, TIER_1, TIER_2, UNKNOWN). "
205        "Included as a guardrail annotation.",
206    )
207    is_eu: bool | None = Field(
208        default=None,
209        description="Whether the affected entity is in the EU region.",
210    )
211    tier_warning: str | None = Field(
212        default=None,
213        description="Warning message if the operation targets a sensitive customer tier.",
214    )
215    warnings: list[str] = Field(
216        default_factory=list,
217        description="Warnings raised by this operation.",
218    )
219
220    def __str__(self) -> str:
221        """Return a string representation of the operation result."""
222        if self.success:
223            return f"✓ {self.message}"
224        return f"✗ {self.message}"