diff --git a/src/fastapi_cloud_cli/commands/integrations/resources/providers.py b/src/fastapi_cloud_cli/commands/integrations/models.py similarity index 100% rename from src/fastapi_cloud_cli/commands/integrations/resources/providers.py rename to src/fastapi_cloud_cli/commands/integrations/models.py diff --git a/src/fastapi_cloud_cli/commands/integrations/providers/__init__.py b/src/fastapi_cloud_cli/commands/integrations/providers/__init__.py index 9758f6d4..1646f3c5 100644 --- a/src/fastapi_cloud_cli/commands/integrations/providers/__init__.py +++ b/src/fastapi_cloud_cli/commands/integrations/providers/__init__.py @@ -1,11 +1,13 @@ import typer +from fastapi_cloud_cli.commands.integrations.providers.get import get_provider from fastapi_cloud_cli.commands.integrations.providers.list import list_providers providers_app = typer.Typer( no_args_is_help=True, help="Manage integration providers for a team.", ) +providers_app.command("get")(get_provider) providers_app.command("list")(list_providers) __all__ = ["providers_app"] diff --git a/src/fastapi_cloud_cli/commands/integrations/providers/get.py b/src/fastapi_cloud_cli/commands/integrations/providers/get.py new file mode 100644 index 00000000..1f205a50 --- /dev/null +++ b/src/fastapi_cloud_cli/commands/integrations/providers/get.py @@ -0,0 +1,174 @@ +from typing import Annotated, Any + +import typer +from pydantic import BaseModel, Field +from rich.table import Table +from rich.text import Text +from rich_toolkit import RichToolkit + +from fastapi_cloud_cli.commands.integrations.models import PROVIDER_NAMES, Provider +from fastapi_cloud_cli.utils.api import APIClient +from fastapi_cloud_cli.utils.auth import Identity +from fastapi_cloud_cli.utils.cli import get_details_table, get_rich_toolkit +from fastapi_cloud_cli.utils.dates import format_last_updated +from fastapi_cloud_cli.utils.execution import JsonOutputOption + + +class ConnectedIntegration(BaseModel): + id: str + provider: Provider = Field(validation_alias="type") + created_at: str + + +class ConnectedResourceApp(BaseModel): + id: str + name: str + + +class ConnectedResource(BaseModel): + id: str + name: str + app: ConnectedResourceApp + created_at: str + + +class ConnectedResourcesAPIResponse(BaseModel): + data: list[ConnectedResource] + + +class ProviderGetOutput(BaseModel): + integration: ConnectedIntegration + resources: list[ConnectedResource] + + +def _get_integration( + client: APIClient, + *, + integration_id: str, +) -> ConnectedIntegration: + response = client.get(f"/integrations/{integration_id}") + response.raise_for_status() + + return ConnectedIntegration.model_validate(response.json()) + + +def _get_connected_resources( + client: APIClient, + *, + integration_id: str, +) -> list[ConnectedResource]: + response = client.get(f"/integrations/{integration_id}/connected-resources") + response.raise_for_status() + + return ConnectedResourcesAPIResponse.model_validate(response.json()).data + + +def _get_connected_resources_table(resources: list[ConnectedResource]) -> Table: + table = Table.grid(padding=(0, 2), pad_edge=False) + table.add_column("Name", no_wrap=True) + table.add_column("App", no_wrap=True) + table.add_column("Resource ID", no_wrap=True, overflow="ignore") + table.add_column("Connected", no_wrap=True) + table.add_row( + Text("Name", style="bold"), + Text("App", style="bold"), + Text("Resource ID", style="bold"), + Text("Connected", style="bold"), + ) + table.add_row("", "", "", "") + + for resource in resources: + table.add_row( + Text(resource.name), + Text(resource.app.name, style="dim"), + Text(resource.id), + Text(format_last_updated(resource.created_at), style="dim"), + ) + + return table + + +def _render_provider_get_output( + data: ProviderGetOutput, + toolkit: RichToolkit, +) -> None: + integration = data.integration + + toolkit.print_title("integration") + toolkit.print_line() + toolkit.print( + Text(PROVIDER_NAMES[integration.provider], style="bold"), + emoji="🔌", + ) + toolkit.print_line() + toolkit.print( + get_details_table( + [ + ("id", integration.id), + ("provider", PROVIDER_NAMES[integration.provider]), + ("connected", format_last_updated(integration.created_at)), + ] + ) + ) + toolkit.print_line() + toolkit.print_title("connected resources") + toolkit.print_line() + + if not data.resources: + toolkit.print("No connected resources found.", bullet=False) + return + + toolkit.print(_get_connected_resources_table(data.resources), bullet=False) + + +def get_provider( + integration_id: Annotated[ + str, + typer.Argument(help="ID of the connected integration to return."), + ], + json_output: JsonOutputOption = False, +) -> Any: + """ + Get a connected integration and its resources. + """ + 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.", + ) + + with APIClient() as client: + with ( + toolkit.progress( + title="Fetching integration details", + transient=True, + ) as progress, + client.handle_http_errors( + progress, + default_message=( + "Error fetching integration details. Please try again later." + ), + not_found_message="Integration not found.", + toolkit=toolkit, + ), + ): + integration = _get_integration( + client, + integration_id=integration_id, + ) + resources = _get_connected_resources( + client, + integration_id=integration_id, + ) + + toolkit.success( + ProviderGetOutput( + integration=integration, + resources=resources, + ), + render_output=_render_provider_get_output, + ) diff --git a/src/fastapi_cloud_cli/commands/integrations/resources/get.py b/src/fastapi_cloud_cli/commands/integrations/resources/get.py index 13a69d44..7a4ad734 100644 --- a/src/fastapi_cloud_cli/commands/integrations/resources/get.py +++ b/src/fastapi_cloud_cli/commands/integrations/resources/get.py @@ -6,10 +6,7 @@ from rich.text import Text from rich_toolkit import RichToolkit -from fastapi_cloud_cli.commands.integrations.resources.providers import ( - PROVIDER_NAMES, - Provider, -) +from fastapi_cloud_cli.commands.integrations.models import PROVIDER_NAMES, Provider 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 diff --git a/src/fastapi_cloud_cli/commands/integrations/resources/list.py b/src/fastapi_cloud_cli/commands/integrations/resources/list.py index ecae0870..70a9af4a 100644 --- a/src/fastapi_cloud_cli/commands/integrations/resources/list.py +++ b/src/fastapi_cloud_cli/commands/integrations/resources/list.py @@ -6,10 +6,7 @@ from rich.text import Text from rich_toolkit import RichToolkit -from fastapi_cloud_cli.commands.integrations.resources.providers import ( - PROVIDER_NAMES, - Provider, -) +from fastapi_cloud_cli.commands.integrations.models import PROVIDER_NAMES, Provider 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 diff --git a/tests/integrations/providers/test_get.py b/tests/integrations/providers/test_get.py new file mode 100644 index 00000000..e94a9df2 --- /dev/null +++ b/tests/integrations/providers/test_get.py @@ -0,0 +1,212 @@ +import json +from typing import Literal + +import pytest +import respx +from httpx import Response +from typer.testing import CliRunner + +from fastapi_cloud_cli.cli import cloud_app as app + +runner = CliRunner() + +Provider = Literal["neon", "redis", "supabase", "logfire"] + +INTEGRATION_ID = "00000000-0000-4000-8000-000000000001" +APP_ID = "00000000-0000-4000-8000-000000000002" +RESOURCE_ID = "00000000-0000-4000-8000-000000000003" +SECOND_RESOURCE_ID = "00000000-0000-4000-8000-000000000004" +CREATED_AT = "2026-08-18T10:00:00Z" +RESOURCE_CREATED_AT = "2026-08-19T11:00:00Z" +SECOND_RESOURCE_CREATED_AT = "2026-08-18T11:00:00Z" + + +def _integration(*, provider: Provider = "neon") -> dict[str, str]: + return { + "id": INTEGRATION_ID, + "type": provider, + "created_at": CREATED_AT, + } + + +def _resource( + *, + resource_id: str = RESOURCE_ID, + name: str = "primary", + created_at: str = RESOURCE_CREATED_AT, +) -> dict[str, object]: + return { + "id": resource_id, + "name": name, + "app": { + "id": APP_ID, + "name": "API", + "slug": "api", + "team_id": "unused-team-id", + }, + "provider_metadata": { + "type": "neon", + "project_id": "unused-project-id", + "database_name": "unused-database-name", + }, + "created_at": created_at, + "updated_at": "2026-08-20T12:00:00Z", + } + + +def test_gets_provider_json_returns_not_logged_in_when_logged_out( + logged_out_cli: None, +) -> None: + result = runner.invoke( + app, + ["integrations", "providers", "get", INTEGRATION_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_gets_provider_and_resources_as_json( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + resources = [ + _resource(), + _resource( + resource_id=SECOND_RESOURCE_ID, + name="analytics", + created_at=SECOND_RESOURCE_CREATED_AT, + ), + ] + respx_mock.get(f"/integrations/{INTEGRATION_ID}").mock( + return_value=Response(200, json=_integration()) + ) + respx_mock.get(f"/integrations/{INTEGRATION_ID}/connected-resources").mock( + return_value=Response(200, json={"data": resources, "count": 2}) + ) + + result = runner.invoke( + app, + ["integrations", "providers", "get", INTEGRATION_ID, "--json"], + ) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == { + "data": { + "integration": { + "id": INTEGRATION_ID, + "provider": "neon", + "created_at": CREATED_AT, + }, + "resources": [ + { + "id": RESOURCE_ID, + "name": "primary", + "app": {"id": APP_ID, "name": "API"}, + "created_at": RESOURCE_CREATED_AT, + }, + { + "id": SECOND_RESOURCE_ID, + "name": "analytics", + "app": {"id": APP_ID, "name": "API"}, + "created_at": SECOND_RESOURCE_CREATED_AT, + }, + ], + } + } + assert result.stderr == "" + + +@pytest.mark.respx +def test_gets_provider_and_resources_in_human_output( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + resources = [ + _resource(), + _resource( + resource_id=SECOND_RESOURCE_ID, + name="analytics", + created_at=SECOND_RESOURCE_CREATED_AT, + ), + ] + respx_mock.get(f"/integrations/{INTEGRATION_ID}").mock( + return_value=Response(200, json=_integration(provider="redis")) + ) + respx_mock.get(f"/integrations/{INTEGRATION_ID}/connected-resources").mock( + return_value=Response(200, json={"data": resources, "count": 2}) + ) + + result = runner.invoke( + app, + ["integrations", "providers", "get", INTEGRATION_ID], + ) + + assert result.exit_code == 0 + assert "integration" in result.output + assert "Redis Cloud" in result.output + assert f"id {INTEGRATION_ID}" in result.output + assert "provider Redis Cloud" in result.output + assert "connected resources" in result.output + assert "Name" in result.output + assert "App" in result.output + assert "Resource ID" in result.output + assert "Connected" in result.output + assert "primary" in result.output + assert "API" in result.output + assert RESOURCE_ID in result.output + assert result.output.index("primary") < result.output.index("analytics") + assert APP_ID not in result.output + assert "unused-project-id" not in result.output + + +@pytest.mark.respx +def test_gets_provider_in_human_output_without_resources( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + respx_mock.get(f"/integrations/{INTEGRATION_ID}").mock( + return_value=Response(200, json=_integration()) + ) + respx_mock.get(f"/integrations/{INTEGRATION_ID}/connected-resources").mock( + return_value=Response(200, json={"data": [], "count": 0}) + ) + + result = runner.invoke( + app, + ["integrations", "providers", "get", INTEGRATION_ID], + ) + + assert result.exit_code == 0 + assert "No connected resources found." in result.output + + +@pytest.mark.respx +def test_gets_provider_json_returns_not_found( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + respx_mock.get(f"/integrations/{INTEGRATION_ID}").mock(return_value=Response(404)) + + result = runner.invoke( + app, + ["integrations", "providers", "get", INTEGRATION_ID, "--json"], + ) + + assert result.exit_code == 1 + assert json.loads(result.stdout) == { + "error": { + "code": "not_found", + "message": "Integration not found.", + "hint": None, + } + } + assert result.stderr == ""