✨ Wait for verification after building (#160)

Shortcake-Parent: main

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, TooManyRetriesError, StreamLogError):
+            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,11 +456,6 @@ 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()
@@ -471,6 +471,8 @@ def _wait_for_deployment(
         ) as progress,
         APIClient() as client,
     ):
+        build_complete = False
+
         try:
             for log in client.stream_build_logs(deployment.id):
                 time_elapsed = time.monotonic() - started_at
@@ -479,18 +481,8 @@ def _wait_for_deployment(
                     progress.log(Text.from_ansi(log.message.rstrip()))
 
                 if log.type == "complete":
+                    build_complete = True
                     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":
@@ -519,6 +511,11 @@ def _wait_for_deployment(
 
             raise typer.Exit(1) from None
 
+        if build_complete:
+            toolkit.print_line()
+
+            _verify_deployment(toolkit, client, app_id, deployment)
+
 
 class SignupToWaitingList(BaseModel):
     email: EmailStr

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,57 @@ 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 = timedelta(seconds=120)
+POLL_MAX_RETRIES = 5
+
+
 class APIClient(httpx.Client):
     def __init__(self) -> None:
         settings = Settings.get()
@@ -241,3 +293,33 @@ 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,
+    ) -> DeploymentStatus:
+        start = time.monotonic()
+        error_count = 0
+
+        while True:
+            if time.monotonic() - start > POLL_TIMEOUT.total_seconds():
+                raise TimeoutError("Deployment verification timed out")
+
+            with attempt(error_count):
+                response = self.get(f"/apps/{app_id}/deployments/{deployment_id}")
+                response.raise_for_status()
+                status = DeploymentStatus(response.json()["status"])
+                error_count = 0
+
+                if status in TERMINAL_STATUSES:
+                    return status
+
+                time.sleep(POLL_INTERVAL)
+                continue
+
+            error_count += 1
+            if error_count >= POLL_MAX_RETRIES:
+                raise TooManyRetriesError(
+                    f"Failed after {POLL_MAX_RETRIES} attempts polling deployment status"
+                )

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)
 

tests/test_api_client.py

@@ -11,6 +11,7 @@
     STREAM_LOGS_MAX_RETRIES,
     APIClient,
     BuildLogLineMessage,
+    DeploymentStatus,
     StreamLogError,
     TooManyRetriesError,
 )
@@ -351,3 +352,55 @@ def responses(request: httpx.Request, route: respx.Route) -> Response:
 
     with patch("time.sleep"), pytest.raises(TimeoutError, match="timed out"):
         list(client.stream_build_logs(deployment_id))
+
+
+@pytest.fixture
+def app_id() -> str:
+    return "test-app-456"
+
+
+@pytest.fixture
+def poll_route(
+    respx_mock: respx.MockRouter, app_id: str, deployment_id: str
+) -> respx.Route:
+    return respx_mock.get(f"/apps/{app_id}/deployments/{deployment_id}")
+
+
+def test_poll_deployment_status_recovers_from_transient_errors(
+    poll_route: respx.Route, client: APIClient, app_id: str, deployment_id: str
+) -> None:
+    call_count = 0
+
+    def handler(request: httpx.Request, route: respx.Route) -> Response:
+        nonlocal call_count
+        call_count += 1
+        if call_count <= 2:
+            return Response(500)
+        return Response(200, json={"status": "success"})
+
+    poll_route.mock(side_effect=handler)
+
+    with patch("time.sleep"):
+        status = client.poll_deployment_status(app_id, deployment_id)
+
+    assert status == DeploymentStatus.success
+    assert call_count == 3
+
+
+def test_poll_deployment_status_raises_after_max_consecutive_errors(
+    poll_route: respx.Route, client: APIClient, app_id: str, deployment_id: str
+) -> None:
+    poll_route.mock(return_value=Response(500))
+
+    with patch("time.sleep"), pytest.raises(TooManyRetriesError):
+        client.poll_deployment_status(app_id, deployment_id)
+
+
+def test_poll_deployment_status_timeout(
+    client: APIClient, app_id: str, deployment_id: str
+) -> None:
+    with (
+        patch("fastapi_cloud_cli.utils.api.POLL_TIMEOUT", timedelta(seconds=-1)),
+        pytest.raises(TimeoutError, match="timed out"),
+    ):
+        client.poll_deployment_status(app_id, deployment_id)