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        )

40class VersionOverrideOperationResult(BaseModel):
41    """Result of a version override operation (set or clear).
42
43    This model provides detailed information about the outcome of a version
44    pinning or unpinning operation.
45    """
46
47    success: bool = Field(description="Whether the operation succeeded")
48    message: str = Field(description="Human-readable message describing the result")
49    connector_id: str = Field(description="The ID of the connector that was modified")
50    connector_type: Literal["source", "destination"] = Field(
51        description="The type of connector (source or destination)"
52    )
53    previous_version: str | None = Field(
54        default=None,
55        description="The version before the operation (None if not available)",
56    )
57    new_version: str | None = Field(
58        default=None,
59        description="The version after the operation (None if cleared or failed)",
60    )
61    was_pinned_before: bool | None = Field(
62        default=None,
63        description="Whether a pin was active before the operation",
64    )
65    is_pinned_after: bool | None = Field(
66        default=None,
67        description="Whether a pin is active after the operation",
68    )
69    customer_tier: str | None = Field(
70        default=None,
71        description="Customer tier of the affected entity (TIER_0, TIER_1, TIER_2). "
72        "Included as a guardrail annotation.",
73    )
74    is_eu: bool | None = Field(
75        default=None,
76        description="Whether the affected entity is in the EU region.",
77    )
78    tier_warning: str | None = Field(
79        default=None,
80        description="Warning message if the operation targets a sensitive customer tier.",
81    )
82
83    def __str__(self) -> str:
84        """Return a string representation of the operation result."""
85        if self.success:
86            return f"✓ {self.message}"
87        return f"✗ {self.message}"