✨ Add fastapi cloud domains get

src/fastapi_cloud_cli/commands/domains/__init__.py

@@ -1,11 +1,13 @@
 import typer
 
+from fastapi_cloud_cli.commands.domains.get import get_domain
 from fastapi_cloud_cli.commands.domains.list import list_domains
 
 domains_app = typer.Typer(
     no_args_is_help=True,
     help="Manage the custom domains of your app.",
 )
+domains_app.command("get")(get_domain)
 domains_app.command("list")(list_domains)
 
 __all__ = ["domains_app"]

src/fastapi_cloud_cli/commands/domains/_setup.py

@@ -0,0 +1,211 @@
+from dataclasses import dataclass
+from typing import Literal
+
+from fastapi_cloud_cli.api import (
+    CustomDomain,
+    CustomDomainRecord,
+    CustomDomainStatus,
+)
+
+StepStatus = Literal["verified", "in_progress", "attention", "locked", "failed"]
+StepID = Literal["ownership", "certificate", "traffic", "combined"]
+PhasedStepID = Literal["ownership", "certificate", "traffic"]
+RecordGroup = PhasedStepID
+Phase = Literal["internal", "external", "origin"]
+
+
+@dataclass(frozen=True)
+class SetupStep:
+    id: StepID
+    title: str
+    description: str
+    status: StepStatus
+    records: list[CustomDomainRecord]
+
+
+ATTENTION_STATUSES = frozenset(
+    {
+        CustomDomainStatus.internal_dcv_invalid,
+        CustomDomainStatus.origin_setup_invalid,
+    }
+)
+
+FAILED_STATUSES = frozenset(
+    {
+        CustomDomainStatus.internal_dcv_timeout,
+        CustomDomainStatus.internal_dcv_revoked,
+        CustomDomainStatus.external_dcv_blocked,
+        CustomDomainStatus.external_dcv_timeout,
+        CustomDomainStatus.origin_setup_timeout,
+        CustomDomainStatus.origin_setup_removed,
+    }
+)
+
+PHASE_ORDER: dict[Phase, int] = {
+    "internal": 0,
+    "external": 1,
+    "origin": 2,
+}
+
+
+def _phase_from_status(status: CustomDomainStatus) -> Phase:
+    if status.value.startswith("internal_dcv"):
+        return "internal"
+    if status.value.startswith("external_dcv"):
+        return "external"
+    return "origin"
+
+
+def _record_group(record: CustomDomainRecord) -> RecordGroup:
+    if record.type == "A":
+        return "traffic"
+    if record.type == "TXT":
+        return "ownership" if "_fc-dcv" in (record.name or "") else "certificate"
+    return "certificate" if "_acme-challenge" in (record.name or "") else "traffic"
+
+
+def _step_status(status: CustomDomainStatus, phase: Phase) -> StepStatus:
+    current_phase = PHASE_ORDER[_phase_from_status(status)]
+    step_phase = PHASE_ORDER[phase]
+    if current_phase > step_phase:
+        return "verified"
+    if current_phase < step_phase:
+        return "locked"
+    if status == CustomDomainStatus.origin_setup_success:
+        return "verified"
+    if status in ATTENTION_STATUSES:
+        return "attention"
+    if status in FAILED_STATUSES:
+        return "failed"
+    return "in_progress"
+
+
+def _describe_step(
+    step_id: StepID,
+    status: StepStatus,
+    *,
+    is_apex: bool = False,
+) -> str:
+    if step_id == "ownership":
+        if status == "attention":
+            return (
+                "We found this record, but its value doesn't match. Update it to "
+                "the value below. We re-check every minute, no restart needed."
+            )
+        if status == "failed":
+            return (
+                "We couldn't confirm ownership. Re-check the value below, then "
+                "restart verification."
+            )
+        if status == "verified":
+            return "Ownership confirmed."
+        return (
+            "Add this TXT record. We check every minute and unlock the next step "
+            "automatically."
+        )
+
+    if step_id == "certificate":
+        if status == "locked":
+            return "Unlocks once ownership is verified."
+        if status == "failed":
+            return (
+                "We couldn't secure your domain. Re-check the records below, then "
+                "restart verification."
+            )
+        if status == "verified":
+            return "Domain verified and TLS certificate issued."
+        return (
+            "Add both records so we can issue your TLS certificate. Your live site "
+            "doesn't change yet."
+        )
+
+    if step_id == "traffic":
+        if status == "locked":
+            return (
+                "Unlocks once your domain is secured. Your live site won't change "
+                "until you add the records here."
+            )
+        if status == "attention":
+            return (
+                "We found these records, but they don't match. Update them to the "
+                "values below. We re-check automatically."
+            )
+        if status == "failed":
+            return (
+                "We couldn't confirm your traffic records. Re-check them below, then "
+                "restart verification."
+            )
+        if status == "verified":
+            return "Your domain is live on FastAPI Cloud."
+        return "Add these records to move traffic to FastAPI Cloud with no downtime."
+
+    if status == "attention":
+        return (
+            "We found your records, but some values don't match. Update them to the "
+            "values below. We re-check automatically."
+        )
+    if status == "failed":
+        return (
+            "We couldn't complete setup. Re-check the records below, then restart "
+            "verification."
+        )
+    if status == "verified":
+        return "Your domain is live on FastAPI Cloud."
+    if is_apex:
+        return (
+            "Add the records below. We'll verify ownership, issue your TLS "
+            "certificate, and route traffic automatically."
+        )
+    return (
+        "Add the record below. We'll verify ownership, issue your TLS certificate, "
+        "and route traffic automatically."
+    )
+
+
+def get_setup_steps(domain: CustomDomain) -> list[SetupStep]:
+    is_apex = any(record.type == "A" for record in domain.dns_records)
+
+    if not domain.is_using_pre_validation:
+        if domain.setup_successful:
+            status: StepStatus = "verified"
+        elif domain.setup_failed:
+            status = "failed"
+        elif domain.status in ATTENTION_STATUSES:
+            status = "attention"
+        else:
+            status = "in_progress"
+
+        return [
+            SetupStep(
+                id="combined",
+                title="Verify ownership and route traffic",
+                description=_describe_step("combined", status, is_apex=is_apex),
+                status=status,
+                records=domain.dns_records,
+            )
+        ]
+
+    def records_for(group: RecordGroup) -> list[CustomDomainRecord]:
+        return [
+            record for record in domain.dns_records if _record_group(record) == group
+        ]
+
+    def build_step(
+        step_id: PhasedStepID,
+        title: str,
+        phase: Phase,
+    ) -> SetupStep:
+        status = _step_status(domain.status, phase)
+        return SetupStep(
+            id=step_id,
+            title=title,
+            description=_describe_step(step_id, status),
+            status=status,
+            records=records_for(step_id),
+        )
+
+    return [
+        build_step("ownership", "Prove ownership", "internal"),
+        build_step("certificate", "Secure your domain", "external"),
+        build_step("traffic", "Switch traffic", "origin"),
+    ]

src/fastapi_cloud_cli/commands/domains/_shared.py

@@ -0,0 +1,40 @@
+import string
+
+from rich_toolkit.menu import Option
+
+from fastapi_cloud_cli.api import CustomDomain
+from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit
+
+
+def _normalize_domain_name(name: str) -> str:
+    return name.strip(string.whitespace + ".").lower()
+
+
+def _find_custom_domain(
+    domains: list[CustomDomain],
+    name_or_id: str,
+) -> CustomDomain | None:
+    normalized = _normalize_domain_name(name_or_id)
+
+    return next(
+        (
+            domain
+            for domain in domains
+            if domain.id.lower() == normalized
+            or _normalize_domain_name(domain.name) == normalized
+        ),
+        None,
+    )
+
+
+def _select_custom_domain(
+    toolkit: FastAPIRichToolkit,
+    domains: list[CustomDomain],
+    *,
+    prompt: str,
+) -> CustomDomain:
+    return toolkit.ask(
+        prompt,
+        options=[Option({"name": domain.name, "value": domain}) for domain in domains],
+        bullet=False,
+    )

src/fastapi_cloud_cli/commands/domains/get.py

@@ -0,0 +1,128 @@
+from typing import Annotated, Any
+
+import typer
+from pydantic import BaseModel, Field
+from rich_toolkit import RichToolkit
+
+from fastapi_cloud_cli.api import APIClient, CustomDomain
+from fastapi_cloud_cli.commands.domains._shared import (
+    _find_custom_domain,
+    _select_custom_domain,
+)
+from fastapi_cloud_cli.commands.domains.rendering import render_custom_domain_details
+from fastapi_cloud_cli.utils.apps import resolve_app_id_or_fail
+from fastapi_cloud_cli.utils.auth import Identity
+from fastapi_cloud_cli.utils.cli import get_rich_toolkit
+from fastapi_cloud_cli.utils.execution import JsonOutputOption
+
+
+class CustomDomainGetOutput(BaseModel):
+    app_id: str
+    domain: CustomDomain
+    show_title: Annotated[bool, Field(exclude=True)] = True
+
+
+def _render_custom_domain_get_output(
+    data: CustomDomainGetOutput,
+    toolkit: RichToolkit,
+) -> None:
+    if data.show_title:
+        toolkit.print_title("custom domains")
+        toolkit.print_line()
+
+    render_custom_domain_details(data.domain, toolkit)
+
+
+def get_domain(
+    domain: Annotated[
+        str | None,
+        typer.Argument(
+            help="Hostname or ID of the custom domain to return.",
+        ),
+    ] = None,
+    app_id: Annotated[
+        str | None,
+        typer.Option(
+            "--app-id",
+            help="ID of the app that owns the custom domain.",
+        ),
+    ] = None,
+    json_output: JsonOutputOption = False,
+) -> Any:
+    """
+    Get a custom domain for an app.
+    """
+    identity = Identity()
+
+    with get_rich_toolkit(json_output=json_output) as toolkit:
+        if not identity.is_logged_in():
+            toolkit.fail(
+                "not_logged_in",
+                "No credentials found.",
+                hint="Run `fastapi cloud login`.",
+            )
+
+        app_id = resolve_app_id_or_fail(toolkit, app_id=app_id)
+        domain_was_provided = domain is not None
+
+        if domain is None and toolkit.mode == "json":
+            toolkit.fail(
+                "missing_required_input",
+                "Custom domain is required.",
+                hint="Pass DOMAIN to choose a custom domain.",
+            )
+
+        with APIClient() as client:
+            with (
+                toolkit.progress(
+                    title="Fetching custom domains",
+                    transient=True,
+                ) as progress,
+                client.handle_http_errors(
+                    progress,
+                    default_message=(
+                        "Error fetching custom domains. Please try again later."
+                    ),
+                    not_found_message="App not found.",
+                    toolkit=toolkit,
+                ),
+            ):
+                domains = client.get_custom_domains(app_id=app_id).data
+
+        selected_domain: CustomDomain | None
+
+        if domain is None:
+            toolkit.print_title("custom domains")
+            toolkit.print_line()
+
+            if not domains:
+                toolkit.print("No custom domains found.", bullet=False)
+                return
+
+            selected_domain = _select_custom_domain(
+                toolkit,
+                domains,
+                prompt="Select the custom domain to get:",
+            )
+            toolkit.print_line()
+        else:
+            if (selected_domain := _find_custom_domain(domains, domain)) is None:
+                toolkit.fail(
+                    "not_found",
+                    f"Custom domain {domain} not found.",
+                    hint=(
+                        "Run `fastapi cloud domains list` to see available custom "
+                        "domains."
+                    ),
+                )
+
+        assert selected_domain is not None
+
+        toolkit.success(
+            CustomDomainGetOutput(
+                app_id=app_id,
+                domain=selected_domain,
+                show_title=domain_was_provided,
+            ),
+            render_output=_render_custom_domain_get_output,
+        )