From ac55123745a37d631b8dc72e2b5c35b5bcae5c21 Mon Sep 17 00:00:00 2001 From: Patrick Arminio Date: Wed, 26 Aug 2026 19:32:48 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Add=20`fastapi=20cloud=20integratio?= =?UTF-8?q?ns=20resources=20get`?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Shortcake-Parent: main --- .../integrations/resources/__init__.py | 2 + .../commands/integrations/resources/get.py | 230 ++++++++++++++ .../commands/integrations/resources/list.py | 15 +- .../integrations/resources/providers.py | 10 + tests/integrations/resources/test_get.py | 297 ++++++++++++++++++ 5 files changed, 544 insertions(+), 10 deletions(-) create mode 100644 src/fastapi_cloud_cli/commands/integrations/resources/get.py create mode 100644 src/fastapi_cloud_cli/commands/integrations/resources/providers.py create mode 100644 tests/integrations/resources/test_get.py diff --git a/src/fastapi_cloud_cli/commands/integrations/resources/__init__.py b/src/fastapi_cloud_cli/commands/integrations/resources/__init__.py index d6657522..b556c200 100644 --- a/src/fastapi_cloud_cli/commands/integrations/resources/__init__.py +++ b/src/fastapi_cloud_cli/commands/integrations/resources/__init__.py @@ -1,11 +1,13 @@ import typer +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("get")(get_resource) resources_app.command("list")(list_resources) __all__ = ["resources_app"] diff --git a/src/fastapi_cloud_cli/commands/integrations/resources/get.py b/src/fastapi_cloud_cli/commands/integrations/resources/get.py new file mode 100644 index 00000000..13a69d44 --- /dev/null +++ b/src/fastapi_cloud_cli/commands/integrations/resources/get.py @@ -0,0 +1,230 @@ +from typing import Annotated, Any, Literal + +import typer +from pydantic import BaseModel, Field +from rich.console import RenderableType +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.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_details_table, get_rich_toolkit +from fastapi_cloud_cli.utils.dates import format_last_updated +from fastapi_cloud_cli.utils.execution import JsonOutputOption + +DatabaseProvider = Literal["neon", "redis", "supabase"] + + +class DatabaseProviderMetadata(BaseModel): + type: DatabaseProvider + database_name: str + + +class LogfireProviderMetadata(BaseModel): + type: Literal["logfire"] + project_name: str + organization_name: str + + +ProviderMetadata = Annotated[ + DatabaseProviderMetadata | LogfireProviderMetadata, + Field(discriminator="type"), +] + + +class EnvironmentVariable(BaseModel): + name: str + + +class ConnectedResourceAPIResponse(BaseModel): + id: str + name: str + provider_metadata: ProviderMetadata + console_url: str + environment_variables: list[EnvironmentVariable] + created_at: str + updated_at: str + + +class ConnectedResourceBase(BaseModel): + id: str + name: str + provider: Provider + console_url: str + environment_variables: list[str] + created_at: str + updated_at: str + + +class DatabaseConnectedResource(ConnectedResourceBase): + provider: DatabaseProvider + database_name: str + + +class LogfireConnectedResource(ConnectedResourceBase): + provider: Literal["logfire"] + project_name: str + organization_name: str + + +ConnectedResource = DatabaseConnectedResource | LogfireConnectedResource + + +class ResourceGetOutput(BaseModel): + app_id: str + resource: ConnectedResource + + +def _to_connected_resource( + resource: ConnectedResourceAPIResponse, +) -> ConnectedResource: + metadata = resource.provider_metadata + environment_variables = [ + variable.name for variable in resource.environment_variables + ] + + match metadata: + case DatabaseProviderMetadata(): + return DatabaseConnectedResource( + id=resource.id, + name=resource.name, + provider=metadata.type, + database_name=metadata.database_name, + console_url=resource.console_url, + environment_variables=environment_variables, + created_at=resource.created_at, + updated_at=resource.updated_at, + ) + case LogfireProviderMetadata(): + return LogfireConnectedResource( + id=resource.id, + name=resource.name, + provider=metadata.type, + project_name=metadata.project_name, + organization_name=metadata.organization_name, + console_url=resource.console_url, + environment_variables=environment_variables, + created_at=resource.created_at, + updated_at=resource.updated_at, + ) + + +def _get_resource( + client: APIClient, + *, + app_id: str, + resource_id: str, +) -> ConnectedResource: + response = client.get(f"/apps/{app_id}/connected-resources/{resource_id}") + response.raise_for_status() + + resource = ConnectedResourceAPIResponse.model_validate(response.json()) + return _to_connected_resource(resource) + + +def _render_resource_get_output( + data: ResourceGetOutput, + toolkit: RichToolkit, +) -> None: + resource = data.resource + + toolkit.print_title("connected resource") + toolkit.print_line() + toolkit.print(Text(resource.name, style="bold"), emoji="🔌") + toolkit.print_line() + + rows: list[tuple[str, RenderableType]] = [ + ("id", resource.id), + ("provider", PROVIDER_NAMES[resource.provider]), + ] + + match resource: + case DatabaseConnectedResource(): + rows.append(("database", resource.database_name)) + case LogfireConnectedResource(): + rows.extend( + [ + ("project", resource.project_name), + ("organization", resource.organization_name), + ] + ) + + rows.extend( + [ + ( + "console", + Text(resource.console_url, style=f"link {resource.console_url}"), + ), + ( + "environment variables", + Text("\n".join(resource.environment_variables)) + if resource.environment_variables + else Text("-", style="dim"), + ), + ("connected", format_last_updated(resource.created_at)), + ("last updated", format_last_updated(resource.updated_at)), + ] + ) + + toolkit.print(get_details_table(rows)) + + +def get_resource( + resource_id: Annotated[ + str, + typer.Argument(help="ID of the connected resource to return."), + ], + app_id: Annotated[ + str | None, + typer.Option( + "--app-id", + help="ID of the app that owns the connected resource.", + ), + ] = None, + json_output: JsonOutputOption = False, +) -> Any: + """ + Get a resource connected 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 = resolve_app_id_or_fail(toolkit, app_id=app_id) + + with APIClient() as client: + with ( + toolkit.progress( + title="Fetching connected resource", + transient=True, + ) as progress, + client.handle_http_errors( + progress, + default_message=( + "Error fetching connected resource. Please try again later." + ), + not_found_message="Connected resource not found.", + toolkit=toolkit, + ), + ): + resource = _get_resource( + client, + app_id=app_id, + resource_id=resource_id, + ) + + toolkit.success( + ResourceGetOutput(app_id=app_id, resource=resource), + render_output=_render_resource_get_output, + ) diff --git a/src/fastapi_cloud_cli/commands/integrations/resources/list.py b/src/fastapi_cloud_cli/commands/integrations/resources/list.py index a7df6dce..ecae0870 100644 --- a/src/fastapi_cloud_cli/commands/integrations/resources/list.py +++ b/src/fastapi_cloud_cli/commands/integrations/resources/list.py @@ -1,4 +1,4 @@ -from typing import Annotated, Any, Literal +from typing import Annotated, Any import typer from pydantic import AliasPath, BaseModel, Field @@ -6,21 +6,16 @@ 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.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 -Provider = Literal["neon", "redis", "supabase", "logfire"] - -PROVIDER_NAMES: dict[Provider, str] = { - "neon": "Neon", - "redis": "Redis Cloud", - "supabase": "Supabase", - "logfire": "Logfire", -} - class ConnectedResource(BaseModel): id: str diff --git a/src/fastapi_cloud_cli/commands/integrations/resources/providers.py b/src/fastapi_cloud_cli/commands/integrations/resources/providers.py new file mode 100644 index 00000000..73eaff2b --- /dev/null +++ b/src/fastapi_cloud_cli/commands/integrations/resources/providers.py @@ -0,0 +1,10 @@ +from typing import Literal + +Provider = Literal["neon", "redis", "supabase", "logfire"] + +PROVIDER_NAMES: dict[Provider, str] = { + "neon": "Neon", + "redis": "Redis Cloud", + "supabase": "Supabase", + "logfire": "Logfire", +} diff --git a/tests/integrations/resources/test_get.py b/tests/integrations/resources/test_get.py new file mode 100644 index 00000000..ce3234ac --- /dev/null +++ b/tests/integrations/resources/test_get.py @@ -0,0 +1,297 @@ +import json +from pathlib import Path +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 +from fastapi_cloud_cli.utils.apps import AppConfig, write_app_config +from tests.utils import changing_dir + +runner = CliRunner() + +DatabaseProvider = Literal["neon", "redis", "supabase"] + +APP_ID = "00000000-0000-4000-8000-000000000001" +LINKED_APP_ID = "00000000-0000-4000-8000-000000000002" +TEAM_ID = "00000000-0000-4000-8000-000000000003" +RESOURCE_ID = "00000000-0000-4000-8000-000000000004" +INTEGRATION_ID = "00000000-0000-4000-8000-000000000005" +CONSOLE_URL = "https://provider.example/resource" +CREATED_AT = "2026-08-18T10:00:00Z" +UPDATED_AT = "2026-08-19T11:00:00Z" + + +def _database_resource( + *, + provider: DatabaseProvider = "neon", + environment_variables: list[str] | None = None, +) -> dict[str, object]: + if environment_variables is None: + environment_variables = ["DATABASE_URL", "DATABASE_PASSWORD"] + + return { + "id": RESOURCE_ID, + "name": "production", + "app_id": APP_ID, + "integration_id": INTEGRATION_ID, + "provider_metadata": { + "type": provider, + "database_name": "app_db", + "project_id": "unused-project-id", + "branch_id": "unused-branch-id", + }, + "console_url": CONSOLE_URL, + "environment_variables": [{"name": name} for name in environment_variables], + "created_at": CREATED_AT, + "updated_at": UPDATED_AT, + } + + +def _logfire_resource( + *, + environment_variables: list[str] | None = None, +) -> dict[str, object]: + if environment_variables is None: + environment_variables = ["LOGFIRE_TOKEN"] + + return { + "id": RESOURCE_ID, + "name": "observability", + "app_id": APP_ID, + "integration_id": INTEGRATION_ID, + "provider_metadata": { + "type": "logfire", + "project_id": "unused-project-id", + "project_name": "fastapi-api", + "organization_name": "acme", + "write_token_id": "unused-token-id", + }, + "console_url": CONSOLE_URL, + "environment_variables": [{"name": name} for name in environment_variables], + "created_at": CREATED_AT, + "updated_at": UPDATED_AT, + } + + +def test_gets_resource_json_returns_not_logged_in_when_logged_out( + logged_out_cli: None, +) -> None: + result = runner.invoke( + app, + [ + "integrations", + "resources", + "get", + RESOURCE_ID, + "--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_gets_database_resource_as_json_with_explicit_app_overriding_linked_app( + 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), + ) + respx_mock.get(f"/apps/{APP_ID}/connected-resources/{RESOURCE_ID}").mock( + return_value=Response(200, json=_database_resource()) + ) + + with changing_dir(tmp_path): + result = runner.invoke( + app, + [ + "integrations", + "resources", + "get", + RESOURCE_ID, + "--app-id", + APP_ID, + "--json", + ], + ) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == { + "data": { + "app_id": APP_ID, + "resource": { + "id": RESOURCE_ID, + "name": "production", + "provider": "neon", + "console_url": CONSOLE_URL, + "environment_variables": ["DATABASE_URL", "DATABASE_PASSWORD"], + "created_at": CREATED_AT, + "updated_at": UPDATED_AT, + "database_name": "app_db", + }, + } + } + assert result.stderr == "" + + +@pytest.mark.respx +def test_gets_logfire_resource_as_json_using_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), + ) + respx_mock.get(f"/apps/{APP_ID}/connected-resources/{RESOURCE_ID}").mock( + return_value=Response(200, json=_logfire_resource()) + ) + + with changing_dir(tmp_path): + result = runner.invoke( + app, + ["integrations", "resources", "get", RESOURCE_ID, "--json"], + ) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == { + "data": { + "app_id": APP_ID, + "resource": { + "id": RESOURCE_ID, + "name": "observability", + "provider": "logfire", + "console_url": CONSOLE_URL, + "environment_variables": ["LOGFIRE_TOKEN"], + "created_at": CREATED_AT, + "updated_at": UPDATED_AT, + "project_name": "fastapi-api", + "organization_name": "acme", + }, + } + } + + +def test_gets_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", "get", RESOURCE_ID, "--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_gets_database_resource_in_human_output( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + respx_mock.get(f"/apps/{APP_ID}/connected-resources/{RESOURCE_ID}").mock( + return_value=Response(200, json=_database_resource()) + ) + + result = runner.invoke( + app, + ["integrations", "resources", "get", RESOURCE_ID, "--app-id", APP_ID], + ) + + assert result.exit_code == 0 + assert "connected resource" in result.output + assert "production" in result.output + assert f"id {RESOURCE_ID}" in result.output + assert "provider Neon" in result.output + assert "database app_db" in result.output + assert f"console {CONSOLE_URL}" in result.output + assert "environment variables DATABASE_URL" in result.output + assert "DATABASE_PASSWORD" in result.output + assert "connected" in result.output + assert "last updated" in result.output + assert INTEGRATION_ID not in result.output + assert "unused-project-id" not in result.output + + +@pytest.mark.respx +def test_gets_logfire_resource_in_human_output_without_environment_variables( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + respx_mock.get(f"/apps/{APP_ID}/connected-resources/{RESOURCE_ID}").mock( + return_value=Response( + 200, + json=_logfire_resource(environment_variables=[]), + ) + ) + + result = runner.invoke( + app, + ["integrations", "resources", "get", RESOURCE_ID, "--app-id", APP_ID], + ) + + assert result.exit_code == 0 + assert "provider Logfire" in result.output + assert "project fastapi-api" in result.output + assert "organization acme" in result.output + assert "environment variables -" in result.output + + +@pytest.mark.respx +def test_gets_resource_json_returns_not_found( + logged_in_cli: None, + respx_mock: respx.MockRouter, +) -> None: + respx_mock.get(f"/apps/{APP_ID}/connected-resources/{RESOURCE_ID}").mock( + return_value=Response(404) + ) + + result = runner.invoke( + app, + [ + "integrations", + "resources", + "get", + RESOURCE_ID, + "--app-id", + APP_ID, + "--json", + ], + ) + + assert result.exit_code == 1 + assert json.loads(result.stdout) == { + "error": { + "code": "not_found", + "message": "Connected resource not found.", + "hint": None, + } + } + assert result.stderr == ""