✨ Show build logs in real time

pyproject.toml

@@ -2,12 +2,10 @@
 name = "fastapi-cloud-cli"
 dynamic = ["version"]
 description = "Deploy and manage FastAPI Cloud apps from the command line 🚀"
-authors = [
-    {name = "Patrick Arminio", email = "patrick@fastapilabs.com"},
-]
+authors = [{ name = "Patrick Arminio", email = "patrick@fastapilabs.com" }]
 requires-python = ">=3.8"
 readme = "README.md"
-license = {text = "MIT"}
+license = { text = "MIT" }
 classifiers = [
     "Intended Audience :: Information Technology",
     "Intended Audience :: System Administrators",
@@ -37,14 +35,12 @@ dependencies = [
     "uvicorn[standard] >= 0.15.0",
     "rignore >= 0.5.1",
     "httpx >= 0.27.0,< 0.28.0",
-    "rich-toolkit >= 0.12.0",
+    "rich-toolkit >= 0.14.3",
     "pydantic >= 1.6.1",
 ]
 
 [project.optional-dependencies]
-standard = [
-    "uvicorn[standard] >= 0.15.0",
-]
+standard = ["uvicorn[standard] >= 0.15.0"]
 
 [project.urls]
 Homepage = "https://github.com/fastapi/fastapi-cloud-cli"
@@ -62,32 +58,20 @@ version = { source = "file", path = "src/fastapi_cloud_cli/__init__.py" }
 distribution = true
 
 [tool.pdm.build]
-source-includes = [
-    "tests/",
-    "requirements*.txt",
-    "scripts/",
-]
+source-includes = ["tests/", "requirements*.txt", "scripts/"]
 
 [tool.pytest.ini_options]
-addopts = [
-  "--strict-config",
-  "--strict-markers",
-]
+addopts = ["--strict-config", "--strict-markers"]
 xfail_strict = true
 junit_family = "xunit2"
 
 [tool.coverage.run]
 parallel = true
 data_file = "coverage/.coverage"
-source = [
-    "src",
-    "tests",
-]
+source = ["src", "tests"]
 context = '${CONTEXT}'
 dynamic_context = "test_function"
-omit = [
-    "tests/assets/*",
-]
+omit = ["tests/assets/*"]
 
 [tool.coverage.report]
 show_missing = true
@@ -104,9 +88,7 @@ show_contexts = true
 
 [tool.mypy]
 strict = true
-exclude = [
-    "tests/assets/*",
-]
+exclude = ["tests/assets/*"]
 
 [tool.ruff.lint]
 select = [
@@ -115,13 +97,13 @@ select = [
     "F",  # pyflakes
     "I",  # isort
     "B",  # flake8-bugbear
-    "C4",  # flake8-comprehensions
+    "C4", # flake8-comprehensions
     "UP", # pyupgrade
 ]
 ignore = [
-    "E501",  # line too long, handled by black
-    "B008",  # do not perform function calls in argument defaults
-    "C901",  # too complex
+    "E501", # line too long, handled by black
+    "B008", # do not perform function calls in argument defaults
+    "C901", # too complex
     "W191", # indentation contains tabs
 ]
 

src/fastapi_cloud_cli/commands/deploy.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import tarfile
 import tempfile
@@ -6,12 +7,13 @@
 from enum import Enum
 from itertools import cycle
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, Generator, List, Optional, Union
 
 import rignore
 import typer
 from httpx import Client
 from pydantic import BaseModel
+from rich.text import Text
 from rich_toolkit import RichToolkit
 from rich_toolkit.menu import Option
 from typing_extensions import Annotated
@@ -212,6 +214,16 @@ def _create_environment_variables(app_id: str, env_vars: Dict[str, str]) -> None
         response.raise_for_status()
 
 
+def _stream_build_logs(deployment_id: str) -> Generator[str, None, None]:
+    with APIClient() as client:
+        with client.stream(
+            "GET", f"/deployments/{deployment_id}/build-logs", timeout=60
+        ) as response:
+            response.raise_for_status()
+
+            yield from response.iter_lines()
+
+
 WAITING_MESSAGES = [
     "🚀 Preparing for liftoff! Almost there...",
     "👹 Sneaking past the dependency gremlins... Don't wake them up!",
@@ -314,40 +326,35 @@ def _wait_for_deployment(
     toolkit.print_line()
 
     toolkit.print(
-        f"You can also check the status at [link]{check_deployment_url}[/link]",
+        f"You can also check the status at [link={check_deployment_url}]{check_deployment_url}[/link]",
     )
     toolkit.print_line()
 
-    time_elapsed = 0
+    time_elapsed = 0.0
 
-    with toolkit.progress("Deploying...") as progress:
-        while True:
-            with handle_http_errors(progress):
-                deployment = _get_deployment(app_id, deployment_id)
+    started_at = time.monotonic()
 
-            if deployment.status == DeploymentStatus.success:
-                progress.log(
-                    f"🐔 Ready the chicken! Your app is ready at {deployment.url}"
-                )
-                break
-            elif deployment.status == DeploymentStatus.failed:
-                progress.set_error(
-                    f"Deployment failed. Please check the logs for more information.\n\n[link={check_deployment_url}]{check_deployment_url}[/link]"
-                )
+    last_message_changed_at = time.monotonic()
 
-                raise typer.Exit(1)
-            else:
-                message = next(messages)
-                progress.log(
-                    f"{message} ({DeploymentStatus.to_human_readable(deployment.status)})"
-                )
+    with toolkit.progress(
+        next(messages), inline_logs=True, lines_to_show=20
+    ) as progress:
+        for line in _stream_build_logs(deployment_id):
+            time_elapsed = time.monotonic() - started_at
 
-            time.sleep(4)
-            time_elapsed += 4
+            data = json.loads(line)
 
-            if time_elapsed == len(WAITING_MESSAGES) * 4:
+            if "message" in data:
+                progress.log(Text.from_ansi(data["message"].rstrip()))
+
+            if time_elapsed > 10:
                 messages = cycle(LONG_WAIT_MESSAGES)
 
+            if (time.monotonic() - last_message_changed_at) > 2:
+                progress.title = next(messages)
+
+                last_message_changed_at = time.monotonic()
+
 
 def _setup_environment_variables(toolkit: RichToolkit, app_id: str) -> None:
     if not toolkit.confirm("Do you want to setup environment variables?", tag="env"):
@@ -358,7 +365,9 @@ def _setup_environment_variables(toolkit: RichToolkit, app_id: str) -> None:
     env_vars = {}
 
     while True:
-        key = toolkit.input("Enter the environment variable name: [ENTER to skip]")
+        key = toolkit.input(
+            "Enter the environment variable name: [ENTER to skip]", required=False
+        )
 
         if key.strip() == "":
             break
@@ -462,5 +471,5 @@ def deploy(
             _wait_for_deployment(toolkit, app.id, deployment.id, check_deployment_url)
         else:
             toolkit.print(
-                f"Check the status of your deployment at [link]{check_deployment_url}[/link]"
+                f"Check the status of your deployment at [link={check_deployment_url}]{check_deployment_url}[/link]"
             )

src/fastapi_cloud_cli/utils/cli.py

@@ -1,18 +1,52 @@
 import contextlib
 import logging
-from typing import Generator, Optional
+from typing import Any, Dict, Generator, List, Optional, Tuple
 
 import typer
 from httpx import HTTPError, HTTPStatusError, ReadTimeout
+from rich.segment import Segment
 from rich_toolkit import RichToolkit, RichToolkitTheme
 from rich_toolkit.progress import Progress
 from rich_toolkit.styles import MinimalStyle, TaggedStyle
 
 logger = logging.getLogger(__name__)
 
 
+class FastAPIStyle(TaggedStyle):
+    def __init__(self, tag_width: int = 11):
+        super().__init__(tag_width=tag_width)
+
+    def _get_tag_segments(
+        self,
+        metadata: Dict[str, Any],
+        is_animated: bool = False,
+        done: bool = False,
+    ) -> Tuple[List[Segment], int]:
+        if not is_animated:
+            return super()._get_tag_segments(metadata, is_animated, done)
+
+        emojis = [
+            "🥚",
+            "🐣",
+            "🐤",
+            "🐥",
+            "🐓",
+            "🐔",
+        ]
+
+        tag = emojis[self.animation_counter % len(emojis)]
+
+        if done:
+            tag = emojis[-1]
+
+        left_padding = self.tag_width - 1
+        left_padding = max(0, left_padding)
+
+        return [Segment(tag)], left_padding
+
+
 def get_rich_toolkit(minimal: bool = False) -> RichToolkit:
-    style = MinimalStyle() if minimal else TaggedStyle(tag_width=11)
+    style = MinimalStyle() if minimal else FastAPIStyle(tag_width=11)
 
     theme = RichToolkitTheme(
         style=style,

tests/test_cli_deploy.py

@@ -221,78 +221,6 @@ def test_uses_existing_app(
         assert app_data["slug"] in result.output
 
 
-@pytest.mark.respx(base_url=settings.base_api_url)
-def test_creates_and_uploads_deployment_then_fails(
-    logged_in_cli: None, tmp_path: Path, respx_mock: respx.MockRouter
-) -> None:
-    steps = [
-        Keys.ENTER,
-        Keys.ENTER,
-        Keys.ENTER,
-        *"demo",
-        Keys.ENTER,
-        Keys.RIGHT_ARROW,
-        Keys.ENTER,
-    ]
-
-    team = _get_random_team()
-    app_data = _get_random_app(team_id=team["id"])
-
-    respx_mock.get("/teams/").mock(return_value=Response(200, json={"data": [team]}))
-
-    respx_mock.post("/apps/", json={"name": "demo", "team_id": team["id"]}).mock(
-        return_value=Response(201, json=app_data)
-    )
-
-    respx_mock.get(f"/apps/{app_data['id']}").mock(
-        return_value=Response(200, json=app_data)
-    )
-
-    deployment_data = _get_random_deployment(app_id=app_data["id"])
-
-    respx_mock.post(f"/apps/{app_data['id']}/deployments/").mock(
-        return_value=Response(201, json=deployment_data)
-    )
-    respx_mock.post(
-        f"/deployments/{deployment_data['id']}/upload",
-    ).mock(
-        return_value=Response(
-            200,
-            json={
-                "url": "http://test.com",
-                "fields": {"key": "value"},
-            },
-        )
-    )
-
-    respx_mock.post(
-        "http://test.com",
-        data={"key": "value"},
-    ).mock(return_value=Response(200))
-
-    respx_mock.get(
-        f"/apps/{app_data['id']}/deployments/{deployment_data['id']}",
-    ).mock(
-        return_value=Response(
-            200,
-            json=_get_random_deployment(app_id=app_data["id"], status="failed"),
-        )
-    )
-
-    respx_mock.post(
-        f"/deployments/{deployment_data['id']}/upload-complete",
-    ).mock(return_value=Response(200))
-
-    with changing_dir(tmp_path), patch("click.getchar") as mock_getchar:
-        mock_getchar.side_effect = steps
-
-        result = runner.invoke(app, ["deploy"])
-
-        assert result.exit_code == 1
-
-        assert "Checking the status of your deployment" in result.output
-
-
 @pytest.mark.respx(base_url=settings.base_api_url)
 def test_exits_successfully_when_deployment_is_done(
     logged_in_cli: None, tmp_path: Path, respx_mock: respx.MockRouter
@@ -346,9 +274,12 @@ def test_exits_successfully_when_deployment_is_done(
         data={"key": "value"},
     ).mock(return_value=Response(200))
 
-    respx_mock.get(f"/apps/{app_data['id']}/deployments/{deployment_data['id']}").mock(
+    respx_mock.get(f"/deployments/{deployment_data['id']}/build-logs").mock(
         return_value=Response(
-            200, json=_get_random_deployment(app_id=app_data["id"], status="success")
+            200,
+            json={
+                "message": "Hello, world!",
+            },
         )
     )
 
@@ -359,7 +290,7 @@ def test_exits_successfully_when_deployment_is_done(
 
         assert result.exit_code == 0
 
-        assert "Ready the chicken! Your app is ready at" in result.output
+        # TODO: show a message when the deployment is done (based on the status)
 
 
 @pytest.mark.respx(base_url=settings.base_api_url)
@@ -394,9 +325,12 @@ def test_exists_successfully_when_deployment_is_done_when_app_is_configured(
         return_value=Response(200)
     )
 
-    respx_mock.get(f"/apps/{app_id}/deployments/{deployment_data['id']}").mock(
+    respx_mock.get(f"/deployments/{deployment_data['id']}/build-logs").mock(
         return_value=Response(
-            200, json=_get_random_deployment(app_id=app_id, status="success")
+            200,
+            json={
+                "message": "Hello, world!",
+            },
         )
     )
 
@@ -409,7 +343,7 @@ def test_exists_successfully_when_deployment_is_done_when_app_is_configured(
 
         assert result.exit_code == 0
 
-        assert "Ready the chicken! Your app is ready at" in result.output
+        # TODO: show a message when the deployment is done (based on the status)
 
 
 @pytest.mark.respx(base_url=settings.base_api_url)