✨ Add fastapi cloud integrations resources connect

src/fastapi_cloud_cli/commands/integrations/resources/__init__.py

@@ -1,12 +1,14 @@
 import typer
 
+from fastapi_cloud_cli.commands.integrations.resources.connect import connect_resource
 from fastapi_cloud_cli.commands.integrations.resources.get import get_resource
 from fastapi_cloud_cli.commands.integrations.resources.list import list_resources
 
 resources_app = typer.Typer(
     no_args_is_help=True,
     help="Manage resources connected to an app.",
 )
+resources_app.command("connect")(connect_resource)
 resources_app.command("get")(get_resource)
 resources_app.command("list")(list_resources)
 

src/fastapi_cloud_cli/commands/integrations/resources/connect.py

@@ -0,0 +1,155 @@
+import logging
+from typing import Annotated, Any
+
+import typer
+from pydantic import BaseModel, Field
+from rich.text import Text
+from rich_toolkit import RichToolkit
+
+from fastapi_cloud_cli.commands.apps.list import (
+    App,
+    _get_app,
+    _get_app_dashboard_url,
+    _get_team,
+)
+from fastapi_cloud_cli.config import Settings
+from fastapi_cloud_cli.utils.api import APIClient
+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
+
+logger = logging.getLogger(__name__)
+
+
+class ResourceConnectOutput(BaseModel):
+    app_id: str
+    connect_url: str
+    app_name: Annotated[str, Field(exclude=True)]
+    browser_opened: Annotated[bool, Field(exclude=True)]
+    app_id_was_provided: Annotated[bool, Field(exclude=True)]
+
+
+def _get_resource_connect_url(
+    app: App,
+    *,
+    team_slug: str,
+    settings: Settings,
+) -> str:
+    app_dashboard_url = _get_app_dashboard_url(
+        app,
+        team_slug=team_slug,
+        settings=settings,
+    )
+    return f"{app_dashboard_url}/integrations/connect"
+
+
+def _render_resource_connect_output(
+    data: ResourceConnectOutput,
+    toolkit: RichToolkit,
+) -> None:
+    toolkit.print_title("connect resource")
+    toolkit.print_line()
+
+    if data.browser_opened:
+        toolkit.print(
+            f"Opened the integration setup for [bold]{data.app_name}[/bold] in your browser.",
+            emoji="🔌",
+        )
+    else:
+        toolkit.print(
+            f"Open the integration setup for [bold]{data.app_name}[/bold] in your browser:",
+            emoji="🔌",
+        )
+
+    toolkit.print(Text(data.connect_url, style=f"link {data.connect_url}"))
+    toolkit.print_line()
+
+    list_command = "fastapi cloud integrations resources list"
+    if data.app_id_was_provided:
+        list_command = f"{list_command} --app-id {data.app_id}"
+
+    toolkit.print(
+        f"When the connection is complete, run [bold]{list_command}[/bold] to view it.",
+        emoji="💡",
+    )
+
+
+def connect_resource(
+    app_id: Annotated[
+        str | None,
+        typer.Option(
+            "--app-id",
+            help="ID of the app to connect a provider resource to.",
+        ),
+    ] = None,
+    no_open: Annotated[
+        bool,
+        typer.Option(
+            "--no-open",
+            help="Do not open the browser automatically.",
+        ),
+    ] = False,
+    json_output: JsonOutputOption = False,
+) -> Any:
+    """
+    Open FastAPI Cloud to connect a provider resource to 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` or set FASTAPI_CLOUD_TOKEN.",
+            )
+
+        app_id_was_provided = app_id is not None
+        app_id = resolve_app_id_or_fail(toolkit, app_id=app_id)
+
+        with APIClient() as client:
+            with (
+                toolkit.progress(title="Fetching app", transient=True) as progress,
+                client.handle_http_errors(
+                    progress,
+                    default_message="Error fetching app. Please try again later.",
+                    not_found_message="App not found.",
+                    toolkit=toolkit,
+                ),
+            ):
+                app = _get_app(client, app_id)
+
+            with (
+                toolkit.progress(title="Fetching team", transient=True) as progress,
+                client.handle_http_errors(
+                    progress,
+                    default_message="Error fetching team. Please try again later.",
+                    not_found_message="Team not found.",
+                    toolkit=toolkit,
+                ),
+            ):
+                team = _get_team(client, app.team_id)
+
+        connect_url = _get_resource_connect_url(
+            app,
+            team_slug=team.slug,
+            settings=Settings.get(),
+        )
+        browser_opened = False
+
+        if not json_output and not no_open:
+            launch_result = typer.launch(connect_url)
+            logger.debug("Launch command result: %s", launch_result)
+            browser_opened = launch_result == 0
+
+        toolkit.success(
+            ResourceConnectOutput(
+                app_id=app.id,
+                app_name=app.name,
+                connect_url=connect_url,
+                browser_opened=browser_opened,
+                app_id_was_provided=app_id_was_provided,
+            ),
+            render_output=_render_resource_connect_output,
+        )

tests/integrations/resources/test_connect.py

@@ -0,0 +1,221 @@
+import json
+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, write_app_config
+from tests.utils import changing_dir
+
+runner = CliRunner()
+
+APP_ID = "00000000-0000-4000-8000-000000000001"
+LINKED_APP_ID = "00000000-0000-4000-8000-000000000002"
+TEAM_ID = "00000000-0000-4000-8000-000000000003"
+TEAM_SLUG = "acme"
+APP_SLUG = "api"
+CONNECT_URL = (
+    f"{Settings().dashboard_base_url}/{TEAM_SLUG}/apps/{APP_SLUG}/integrations/connect"
+)
+
+
+def _app(*, app_id: str = APP_ID) -> dict[str, object]:
+    return {
+        "id": app_id,
+        "team_id": TEAM_ID,
+        "slug": APP_SLUG,
+        "name": "API",
+        "directory": None,
+    }
+
+
+def _team() -> dict[str, str]:
+    return {
+        "id": TEAM_ID,
+        "slug": TEAM_SLUG,
+        "name": "Acme",
+    }
+
+
+def _mock_app_and_team(
+    respx_mock: respx.MockRouter,
+    *,
+    app_id: str = APP_ID,
+) -> None:
+    respx_mock.get(f"/apps/{app_id}").mock(
+        return_value=Response(200, json=_app(app_id=app_id))
+    )
+    respx_mock.get(f"/teams/{TEAM_ID}").mock(return_value=Response(200, json=_team()))
+
+
+def test_connect_resource_json_returns_not_logged_in_when_logged_out(
+    logged_out_cli: None,
+) -> None:
+    result = runner.invoke(
+        app,
+        ["integrations", "resources", "connect", "--app-id", APP_ID, "--json"],
+    )
+
+    assert result.exit_code == 1
+    assert json.loads(result.stdout) == {
+        "error": {
+            "code": "not_logged_in",
+            "message": "No credentials found.",
+            "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.",
+        }
+    }
+    assert result.stderr == ""
+
+
+@pytest.mark.respx
+def test_connect_resource_json_returns_url_without_opening_browser(
+    logged_in_cli: None,
+    respx_mock: respx.MockRouter,
+    tmp_path: Path,
+) -> None:
+    write_app_config(
+        tmp_path,
+        AppConfig(app_id=LINKED_APP_ID, team_id=TEAM_ID),
+    )
+    _mock_app_and_team(respx_mock)
+
+    with (
+        changing_dir(tmp_path),
+        patch(
+            "fastapi_cloud_cli.commands.integrations.resources.connect.typer.launch"
+        ) as mock_launch,
+    ):
+        result = runner.invoke(
+            app,
+            [
+                "integrations",
+                "resources",
+                "connect",
+                "--app-id",
+                APP_ID,
+                "--json",
+            ],
+        )
+
+    assert result.exit_code == 0
+    assert json.loads(result.stdout) == {
+        "data": {
+            "app_id": APP_ID,
+            "connect_url": CONNECT_URL,
+        }
+    }
+    assert result.stderr == ""
+    mock_launch.assert_not_called()
+
+
+@pytest.mark.respx
+def test_connect_resource_opens_browser_for_linked_app(
+    logged_in_cli: None,
+    respx_mock: respx.MockRouter,
+    tmp_path: Path,
+) -> None:
+    write_app_config(
+        tmp_path,
+        AppConfig(app_id=APP_ID, team_id=TEAM_ID),
+    )
+    _mock_app_and_team(respx_mock)
+
+    with (
+        changing_dir(tmp_path),
+        patch(
+            "fastapi_cloud_cli.commands.integrations.resources.connect.typer.launch",
+            return_value=0,
+        ) as mock_launch,
+    ):
+        result = runner.invoke(app, ["integrations", "resources", "connect"])
+
+    assert result.exit_code == 0
+    mock_launch.assert_called_once_with(CONNECT_URL)
+    assert "connect resource" in result.output
+    assert "Opened the integration setup for API in your browser." in result.output
+    assert CONNECT_URL in result.output
+    assert "When the connection is complete" in result.output
+    assert "fastapi cloud integrations resources list" in " ".join(
+        result.output.split()
+    )
+    assert APP_ID not in result.output
+
+
+@pytest.mark.respx
+def test_connect_resource_no_open_prints_url_without_opening_browser(
+    logged_in_cli: None,
+    respx_mock: respx.MockRouter,
+) -> None:
+    _mock_app_and_team(respx_mock)
+
+    with patch(
+        "fastapi_cloud_cli.commands.integrations.resources.connect.typer.launch"
+    ) as mock_launch:
+        result = runner.invoke(
+            app,
+            [
+                "integrations",
+                "resources",
+                "connect",
+                "--app-id",
+                APP_ID,
+                "--no-open",
+            ],
+        )
+
+    assert result.exit_code == 0
+    mock_launch.assert_not_called()
+    assert "Open the integration setup for API in your browser:" in result.output
+    assert CONNECT_URL in result.output
+    assert f"fastapi cloud integrations resources list --app-id {APP_ID}" in " ".join(
+        result.output.split()
+    )
+
+
+def test_connect_resource_json_requires_app_without_linked_app(
+    logged_in_cli: None,
+    tmp_path: Path,
+) -> None:
+    with changing_dir(tmp_path):
+        result = runner.invoke(
+            app,
+            ["integrations", "resources", "connect", "--json"],
+        )
+
+    assert result.exit_code == 1
+    assert json.loads(result.stdout) == {
+        "error": {
+            "code": "missing_required_input",
+            "message": "App ID is required.",
+            "hint": "Pass --app-id or run `fastapi cloud apps create --link` first.",
+        }
+    }
+    assert result.stderr == ""
+
+
+@pytest.mark.respx
+def test_connect_resource_json_returns_not_found_for_unknown_app(
+    logged_in_cli: None,
+    respx_mock: respx.MockRouter,
+) -> None:
+    respx_mock.get(f"/apps/{APP_ID}").mock(return_value=Response(404))
+
+    result = runner.invoke(
+        app,
+        ["integrations", "resources", "connect", "--app-id", APP_ID, "--json"],
+    )
+
+    assert result.exit_code == 1
+    assert json.loads(result.stdout) == {
+        "error": {
+            "code": "not_found",
+            "message": "App not found.",
+            "hint": None,
+        }
+    }
+    assert result.stderr == ""