✨ Wait for verification after building

Shortcake-Parent: update-how-we-show-messages

Author: patrick91

pyproject.toml

@@ -32,7 +32,7 @@ dependencies = [
     "uvicorn[standard] >= 0.17.6",
     "rignore >= 0.5.1",
     "httpx >= 0.27.0",
-    "rich-toolkit >= 0.19.6",
+    "rich-toolkit >= 0.19.7",
     "pydantic[email] >= 2.7.4; python_version < '3.13'",
     "pydantic[email] >= 2.8.0; python_version == '3.13'",
     "pydantic[email] >= 2.12.0; python_version >= '3.14'",

src/fastapi_cloud_cli/commands/deploy.py

@@ -4,7 +4,6 @@
 import subprocess
 import tempfile
 import time
-from enum import Enum
 from itertools import cycle
 from pathlib import Path, PurePosixPath
 from textwrap import dedent
@@ -20,7 +19,13 @@
 from rich_toolkit.menu import Option
 
 from fastapi_cloud_cli.commands.login import login
-from fastapi_cloud_cli.utils.api import APIClient, StreamLogError, TooManyRetriesError
+from fastapi_cloud_cli.utils.api import (
+    SUCCESSFUL_STATUSES,
+    APIClient,
+    DeploymentStatus,
+    StreamLogError,
+    TooManyRetriesError,
+)
 from fastapi_cloud_cli.utils.apps import AppConfig, get_app_config, write_app_config
 from fastapi_cloud_cli.utils.auth import Identity
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors
@@ -174,42 +179,6 @@ def _create_app(team_id: str, app_name: str, directory: str | None) -> AppRespon
         return AppResponse.model_validate(response.json())
 
 
-class DeploymentStatus(str, Enum):
-    waiting_upload = "waiting_upload"
-    ready_for_build = "ready_for_build"
-    building = "building"
-    extracting = "extracting"
-    extracting_failed = "extracting_failed"
-    building_image = "building_image"
-    building_image_failed = "building_image_failed"
-    deploying = "deploying"
-    deploying_failed = "deploying_failed"
-    verifying = "verifying"
-    verifying_failed = "verifying_failed"
-    verifying_skipped = "verifying_skipped"
-    success = "success"
-    failed = "failed"
-
-    @classmethod
-    def to_human_readable(cls, status: "DeploymentStatus") -> str:
-        return {
-            cls.waiting_upload: "Waiting for upload",
-            cls.ready_for_build: "Ready for build",
-            cls.building: "Building",
-            cls.extracting: "Extracting",
-            cls.extracting_failed: "Extracting failed",
-            cls.building_image: "Building image",
-            cls.building_image_failed: "Build failed",
-            cls.deploying: "Deploying",
-            cls.deploying_failed: "Deploying failed",
-            cls.verifying: "Verifying",
-            cls.verifying_failed: "Verifying failed",
-            cls.verifying_skipped: "Verification skipped",
-            cls.success: "Success",
-            cls.failed: "Failed",
-        }[status]
-
-
 class CreateDeploymentResponse(BaseModel):
     id: str
     app_id: str
@@ -440,6 +409,42 @@ def _configure_app(toolkit: RichToolkit, path_to_deploy: Path) -> AppConfig:
     return app_config
 
 
+def _verify_deployment(
+    toolkit: RichToolkit,
+    client: APIClient,
+    app_id: str,
+    deployment: CreateDeploymentResponse,
+) -> None:
+    with toolkit.progress(
+        title="Verifying deployment...",
+        inline_logs=True,
+        done_emoji="✅",
+    ) as progress:
+        try:
+            final_status = client.poll_deployment_status(app_id, deployment.id)
+        except TimeoutError:
+            progress.metadata["done_emoji"] = "⚠️"
+            progress.current_message = (
+                f"Could not confirm deployment status. "
+                f"Check the dashboard: [link={deployment.dashboard_url}]{deployment.dashboard_url}[/link]"
+            )
+            return
+
+        if final_status in SUCCESSFUL_STATUSES:
+            progress.current_message = f"Ready the chicken! 🐔 Your app is ready at [link={deployment.url}]{deployment.url}[/link]"
+        else:
+            progress.metadata["done_emoji"] = "❌"
+            progress.current_message = "Deployment failed"
+
+            human_status = DeploymentStatus.to_human_readable(final_status)
+
+            progress.log(
+                f"😔 Oh no! Deployment failed: {human_status}. "
+                f"Check out the logs at [link={deployment.dashboard_url}]{deployment.dashboard_url}[/link]"
+            )
+            raise typer.Exit(1)
+
+
 def _wait_for_deployment(
     toolkit: RichToolkit, app_id: str, deployment: CreateDeploymentResponse
 ) -> None:
@@ -451,73 +456,66 @@ def _wait_for_deployment(
     )
     toolkit.print_line()
 
-    toolkit.print(
-        f"You can also check the status at [link={deployment.dashboard_url}]{deployment.dashboard_url}[/link]",
-    )
-    toolkit.print_line()
-
     time_elapsed = 0.0
 
     started_at = time.monotonic()
 
     last_message_changed_at = time.monotonic()
 
-    with (
-        toolkit.progress(
+    build_complete = False
+
+    with APIClient() as client:
+        with toolkit.progress(
             next(messages),
             inline_logs=True,
             lines_to_show=20,
             done_emoji="🚀",
-        ) as progress,
-        APIClient() as client,
-    ):
-        try:
-            for log in client.stream_build_logs(deployment.id):
-                time_elapsed = time.monotonic() - started_at
-
-                if log.type == "message":
-                    progress.log(Text.from_ansi(log.message.rstrip()))
-
-                if log.type == "complete":
-                    progress.title = "Build complete!"
-                    progress.log("")
-                    progress.log(
-                        f"You can also check the app logs at [link={deployment.dashboard_url}]{deployment.dashboard_url}[/link]"
-                    )
-
-                    progress.log("")
-
-                    progress.log(
-                        f"🐔 Ready the chicken! Your app is ready at [link={deployment.url}]{deployment.url}[/link]"
-                    )
-
-                    break
-
-                if log.type == "failed":
-                    progress.log("")
-                    progress.log(
-                        f"😔 Oh no! Something went wrong. Check out the logs at [link={deployment.dashboard_url}]{deployment.dashboard_url}[/link]"
-                    )
-                    raise typer.Exit(1)
-
-                if time_elapsed > 30:
-                    messages = cycle(LONG_WAIT_MESSAGES)
-
-                if (time.monotonic() - last_message_changed_at) > 2:
-                    progress.title = next(messages)
+        ) as progress:
+            try:
+                for log in client.stream_build_logs(deployment.id):
+                    time_elapsed = time.monotonic() - started_at
+
+                    if log.type == "message":
+                        progress.log(Text.from_ansi(log.message.rstrip()))
+
+                    if log.type == "complete":
+                        build_complete = True
+                        progress.title = "Build complete!"
+                        break
+
+                    if log.type == "failed":
+                        progress.log("")
+                        progress.log(
+                            f"😔 Oh no! Something went wrong. Check out the logs at [link={deployment.dashboard_url}]{deployment.dashboard_url}[/link]"
+                        )
+                        raise typer.Exit(1)
+
+                    if time_elapsed > 30:
+                        messages = cycle(LONG_WAIT_MESSAGES)
+
+                    if (time.monotonic() - last_message_changed_at) > 2:
+                        progress.title = next(messages)
+
+                        last_message_changed_at = time.monotonic()
+
+            except KeyboardInterrupt:
+                progress.title = "Cancelled"
+                raise
+            except (StreamLogError, TooManyRetriesError, TimeoutError) as e:
+                progress.set_error(
+                    dedent(f"""
+                    [error]Build log streaming failed: {e}[/]
 
-                    last_message_changed_at = time.monotonic()
+                    Unable to stream build logs. Check the dashboard for status: [link={deployment.dashboard_url}]{deployment.dashboard_url}[/link]
+                    """).strip()
+                )
 
-        except (StreamLogError, TooManyRetriesError, TimeoutError) as e:
-            progress.set_error(
-                dedent(f"""
-                [error]Build log streaming failed: {e}[/]
+                raise typer.Exit(1) from None
 
-                Unable to stream build logs. Check the dashboard for status: [link={deployment.dashboard_url}]{deployment.dashboard_url}[/link]
-                """).strip()
-            )
+        if build_complete:
+            toolkit.print_line()
 
-            raise typer.Exit(1) from None
+            _verify_deployment(toolkit, client, app_id, deployment)
 
 
 class SignupToWaitingList(BaseModel):

src/fastapi_cloud_cli/utils/api.py

@@ -4,6 +4,7 @@
 from collections.abc import Callable, Generator
 from contextlib import contextmanager
 from datetime import timedelta
+from enum import Enum
 from functools import wraps
 from typing import (
     Annotated,
@@ -138,6 +139,56 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> Generator[T, None, None]:
     return decorator
 
 
+class DeploymentStatus(str, Enum):
+    waiting_upload = "waiting_upload"
+    ready_for_build = "ready_for_build"
+    building = "building"
+    extracting = "extracting"
+    extracting_failed = "extracting_failed"
+    building_image = "building_image"
+    building_image_failed = "building_image_failed"
+    deploying = "deploying"
+    deploying_failed = "deploying_failed"
+    verifying = "verifying"
+    verifying_failed = "verifying_failed"
+    verifying_skipped = "verifying_skipped"
+    success = "success"
+    failed = "failed"
+
+    @classmethod
+    def to_human_readable(cls, status: "DeploymentStatus") -> str:
+        return {
+            cls.waiting_upload: "Waiting for upload",
+            cls.ready_for_build: "Ready for build",
+            cls.building: "Building",
+            cls.extracting: "Extracting",
+            cls.extracting_failed: "Extracting failed",
+            cls.building_image: "Building image",
+            cls.building_image_failed: "Build failed",
+            cls.deploying: "Deploying",
+            cls.deploying_failed: "Deploying failed",
+            cls.verifying: "Verifying",
+            cls.verifying_failed: "Verifying failed",
+            cls.verifying_skipped: "Verification skipped",
+            cls.success: "Success",
+            cls.failed: "Failed",
+        }[status]
+
+
+SUCCESSFUL_STATUSES = {DeploymentStatus.success, DeploymentStatus.verifying_skipped}
+FAILED_STATUSES = {
+    DeploymentStatus.failed,
+    DeploymentStatus.verifying_failed,
+    DeploymentStatus.deploying_failed,
+    DeploymentStatus.building_image_failed,
+    DeploymentStatus.extracting_failed,
+}
+TERMINAL_STATUSES = SUCCESSFUL_STATUSES | FAILED_STATUSES
+
+POLL_INTERVAL = 2.0
+POLL_TIMEOUT = 1200.0
+
+
 class APIClient(httpx.Client):
     def __init__(self) -> None:
         settings = Settings.get()
@@ -241,3 +292,35 @@ def stream_app_logs(
                 except ValidationError as e:  # pragma: no cover
                     logger.debug("Failed to parse log entry: %s - %s", data, e)
                     continue
+
+    def poll_deployment_status(
+        self,
+        app_id: str,
+        deployment_id: str,
+        timeout: float = POLL_TIMEOUT,
+        interval: float = POLL_INTERVAL,
+    ) -> DeploymentStatus:
+        start = time.monotonic()
+        consecutive_errors = 0
+        max_errors = 5
+
+        while time.monotonic() - start < timeout:
+            try:
+                response = self.get(f"/apps/{app_id}/deployments/{deployment_id}")
+                response.raise_for_status()
+                status = DeploymentStatus(response.json()["status"])
+                consecutive_errors = 0
+            except httpx.HTTPError as e:
+                consecutive_errors += 1
+                logger.debug("Error polling deployment status: %s", e)
+                if consecutive_errors >= max_errors:
+                    raise
+                time.sleep(interval)
+                continue
+
+            if status in TERMINAL_STATUSES:
+                return status
+
+            time.sleep(interval)
+
+        raise TimeoutError("Deployment verification timed out")

src/fastapi_cloud_cli/utils/cli.py

@@ -1,7 +1,7 @@
 import contextlib
 import logging
 from collections.abc import Generator
-from typing import Any
+from typing import Any, Literal
 
 import typer
 from httpx import HTTPError, HTTPStatusError, ReadTimeout
@@ -24,9 +24,12 @@ def _get_tag_segments(
         metadata: dict[str, Any],
         is_animated: bool = False,
         done: bool = False,
+        animation_status: Literal["started", "stopped", "error"] | None = None,
     ) -> tuple[list[Segment], int]:
         if not is_animated:
-            return super()._get_tag_segments(metadata, is_animated, done)
+            return super()._get_tag_segments(
+                metadata, is_animated, done, animation_status=animation_status
+            )
 
         emojis = [
             "🥚",
@@ -42,6 +45,9 @@ def _get_tag_segments(
         if done:
             tag = metadata.get("done_emoji", emojis[-1])
 
+        if animation_status == "error":
+            tag = "⚠️"
+
         left_padding = self.tag_width - 1
         left_padding = max(0, left_padding)