✨ Add link command (#148)

Author: patrick91

src/fastapi_cloud_cli/cli.py

@@ -2,6 +2,7 @@
 
 from .commands.deploy import deploy
 from .commands.env import env_app
+from .commands.link import link
 from .commands.login import login
 from .commands.logout import logout
 from .commands.logs import logs
@@ -25,6 +26,7 @@
 
 # fastapi cloud [command]
 cloud_app.command()(deploy)
+cloud_app.command()(link)
 cloud_app.command()(login)
 cloud_app.command()(logs)
 cloud_app.command()(logout)

src/fastapi_cloud_cli/commands/link.py

@@ -0,0 +1,117 @@
+import logging
+from pathlib import Path
+from typing import Any
+
+import typer
+from rich_toolkit.menu import Option
+
+from fastapi_cloud_cli.utils.api import APIClient
+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
+
+logger = logging.getLogger(__name__)
+
+
+def link() -> Any:
+    """
+    Link a local directory to an existing FastAPI Cloud app.
+    """
+    identity = Identity()
+
+    with get_rich_toolkit() as toolkit:
+        if not identity.is_logged_in():
+            toolkit.print(
+                "[error]You need to be logged in to link an app.[/]",
+            )
+            toolkit.print_line()
+            toolkit.print(
+                "Run [bold]fastapi cloud login[/] to authenticate.",
+                tag="tip",
+            )
+            raise typer.Exit(1)
+
+        path_to_link = Path.cwd()
+
+        if get_app_config(path_to_link):
+            toolkit.print(
+                "[error]This directory is already linked to an app.[/]",
+            )
+            toolkit.print_line()
+            toolkit.print(
+                "Run [bold]fastapi cloud unlink[/] first to remove the existing configuration.",
+                tag="tip",
+            )
+            raise typer.Exit(1)
+
+        toolkit.print_title("Link to FastAPI Cloud", tag="FastAPI")
+        toolkit.print_line()
+
+        with toolkit.progress("Fetching teams...") as progress:
+            with handle_http_errors(
+                progress, message="Error fetching teams. Please try again later."
+            ):
+                with APIClient() as client:
+                    response = client.get("/teams/")
+                    response.raise_for_status()
+                    teams_data = response.json()["data"]
+
+        if not teams_data:
+            toolkit.print(
+                "[error]No teams found. Please create a team first.[/]",
+            )
+            raise typer.Exit(1)
+
+        toolkit.print_line()
+
+        team = toolkit.ask(
+            "Select the team:",
+            tag="team",
+            options=[
+                Option({"name": t["name"], "value": {"id": t["id"], "name": t["name"]}})
+                for t in teams_data
+            ],
+        )
+
+        toolkit.print_line()
+
+        with toolkit.progress("Fetching apps...") as progress:
+            with handle_http_errors(
+                progress, message="Error fetching apps. Please try again later."
+            ):
+                with APIClient() as client:
+                    response = client.get("/apps/", params={"team_id": team["id"]})
+                    response.raise_for_status()
+                    apps_data = response.json()["data"]
+
+        if not apps_data:
+            toolkit.print(
+                "[error]No apps found in this team.[/]",
+            )
+            toolkit.print_line()
+            toolkit.print(
+                "Run [bold]fastapi cloud deploy[/] to create and deploy a new app.",
+                tag="tip",
+            )
+            raise typer.Exit(1)
+
+        toolkit.print_line()
+
+        app = toolkit.ask(
+            "Select the app to link:",
+            tag="app",
+            options=[
+                Option({"name": a["slug"], "value": {"id": a["id"], "slug": a["slug"]}})
+                for a in apps_data
+            ],
+        )
+
+        toolkit.print_line()
+
+        app_config = AppConfig(app_id=app["id"], team_id=team["id"])
+        write_app_config(path_to_link, app_config)
+
+        toolkit.print(
+            f"Successfully linked to app [bold]{app['slug']}[/bold]! 🔗",
+        )
+        logger.debug(f"Linked to app: {app['id']} in team: {team['id']}")

tests/test_cli_link.py

@@ -0,0 +1,191 @@
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+import respx
+from httpx import Response
+from typer.testing import CliRunner
+
+from fastapi_cloud_cli.cli import cloud_app as app
+from fastapi_cloud_cli.config import Settings
+from fastapi_cloud_cli.utils.apps import AppConfig
+from tests.conftest import ConfiguredApp
+from tests.utils import Keys, changing_dir
+
+runner = CliRunner()
+settings = Settings.get()
+
+
+def test_shows_a_message_if_not_logged_in(logged_out_cli: None) -> None:
+    result = runner.invoke(app, ["link"])
+
+    assert result.exit_code == 1
+    assert "You need to be logged in to link an app." in result.output
+
+
+def test_shows_a_message_if_already_linked(
+    logged_in_cli: None, configured_app: ConfiguredApp
+) -> None:
+    with changing_dir(configured_app.path):
+        result = runner.invoke(app, ["link"])
+
+    assert result.exit_code == 1
+    assert "This directory is already linked to an app." in result.output
+
+
+@pytest.mark.respx(base_url=settings.base_api_url)
+def test_shows_a_message_if_no_teams(
+    logged_in_cli: None, respx_mock: respx.MockRouter, tmp_path: Path
+) -> None:
+    respx_mock.get("/teams/").mock(return_value=Response(200, json={"data": []}))
+
+    with changing_dir(tmp_path):
+        result = runner.invoke(app, ["link"])
+
+    assert result.exit_code == 1
+    assert "No teams found" in result.output
+
+
+@pytest.mark.respx(base_url=settings.base_api_url)
+def test_shows_a_message_if_no_apps(
+    logged_in_cli: None, respx_mock: respx.MockRouter, tmp_path: Path
+) -> None:
+    steps = [Keys.ENTER]
+
+    respx_mock.get("/teams/").mock(
+        return_value=Response(
+            200, json={"data": [{"id": "team-1", "name": "My Team", "slug": "my-team"}]}
+        )
+    )
+    respx_mock.get("/apps/", params={"team_id": "team-1"}).mock(
+        return_value=Response(200, json={"data": []})
+    )
+
+    with (
+        changing_dir(tmp_path),
+        patch("rich_toolkit.container.getchar") as mock_getchar,
+    ):
+        mock_getchar.side_effect = steps
+        result = runner.invoke(app, ["link"])
+
+    assert result.exit_code == 1
+    assert "No apps found in this team." in result.output
+
+
+@pytest.mark.respx(base_url=settings.base_api_url)
+def test_links_successfully(
+    logged_in_cli: None, respx_mock: respx.MockRouter, tmp_path: Path
+) -> None:
+    steps = [Keys.ENTER, Keys.ENTER]
+
+    respx_mock.get("/teams/").mock(
+        return_value=Response(
+            200, json={"data": [{"id": "team-1", "name": "My Team", "slug": "my-team"}]}
+        )
+    )
+    respx_mock.get("/apps/", params={"team_id": "team-1"}).mock(
+        return_value=Response(200, json={"data": [{"id": "app-1", "slug": "my-app"}]})
+    )
+
+    with (
+        changing_dir(tmp_path),
+        patch("rich_toolkit.container.getchar") as mock_getchar,
+    ):
+        mock_getchar.side_effect = steps
+        result = runner.invoke(app, ["link"])
+
+    assert result.exit_code == 0
+    assert "Successfully linked to app" in result.output
+    assert "my-app" in result.output
+
+    config_path = tmp_path / ".fastapicloud" / "cloud.json"
+    assert config_path.exists()
+    config = AppConfig.model_validate_json(config_path.read_text())
+    assert config.app_id == "app-1"
+    assert config.team_id == "team-1"
+
+
+@pytest.mark.respx(base_url=settings.base_api_url)
+def test_shows_error_on_teams_api_failure(
+    logged_in_cli: None, respx_mock: respx.MockRouter, tmp_path: Path
+) -> None:
+    respx_mock.get("/teams/").mock(return_value=Response(500))
+
+    with changing_dir(tmp_path):
+        result = runner.invoke(app, ["link"])
+
+    assert result.exit_code == 1
+    assert "Error fetching teams" in result.output
+
+
+@pytest.mark.respx(base_url=settings.base_api_url)
+def test_shows_error_on_apps_api_failure(
+    logged_in_cli: None, respx_mock: respx.MockRouter, tmp_path: Path
+) -> None:
+    steps = [Keys.ENTER]
+
+    respx_mock.get("/teams/").mock(
+        return_value=Response(
+            200, json={"data": [{"id": "team-1", "name": "My Team", "slug": "my-team"}]}
+        )
+    )
+    respx_mock.get("/apps/", params={"team_id": "team-1"}).mock(
+        return_value=Response(500)
+    )
+
+    with (
+        changing_dir(tmp_path),
+        patch("rich_toolkit.container.getchar") as mock_getchar,
+    ):
+        mock_getchar.side_effect = steps
+        result = runner.invoke(app, ["link"])
+
+    assert result.exit_code == 1
+    assert "Error fetching apps" in result.output
+
+
+@pytest.mark.respx(base_url=settings.base_api_url)
+def test_links_with_multiple_teams_and_apps(
+    logged_in_cli: None, respx_mock: respx.MockRouter, tmp_path: Path
+) -> None:
+    steps = [Keys.DOWN_ARROW, Keys.ENTER, Keys.DOWN_ARROW, Keys.ENTER]
+
+    respx_mock.get("/teams/").mock(
+        return_value=Response(
+            200,
+            json={
+                "data": [
+                    {"id": "team-1", "name": "Team One", "slug": "team-one"},
+                    {"id": "team-2", "name": "Team Two", "slug": "team-two"},
+                ]
+            },
+        )
+    )
+    respx_mock.get("/apps/", params={"team_id": "team-2"}).mock(
+        return_value=Response(
+            200,
+            json={
+                "data": [
+                    {"id": "app-1", "slug": "first-app"},
+                    {"id": "app-2", "slug": "second-app"},
+                ]
+            },
+        )
+    )
+
+    with (
+        changing_dir(tmp_path),
+        patch("rich_toolkit.container.getchar") as mock_getchar,
+    ):
+        mock_getchar.side_effect = steps
+        result = runner.invoke(app, ["link"])
+
+    assert result.exit_code == 0
+    assert "Successfully linked to app" in result.output
+    assert "second-app" in result.output
+
+    config_path = tmp_path / ".fastapicloud" / "cloud.json"
+    assert config_path.exists()
+    config = AppConfig.model_validate_json(config_path.read_text())
+    assert config.app_id == "app-2"
+    assert config.team_id == "team-2"