From f8baba2b6cf2ff6b48465f4ea6d231339d0ae98a Mon Sep 17 00:00:00 2001 From: Patrick Arminio Date: Wed, 2 Sep 2026 11:03:35 +0200 Subject: [PATCH] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Improve=20how=20we=20handl?= =?UTF-8?q?e=20auth=20on=20each=20command?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Shortcake-Parent: main --- src/fastapi_cloud_cli/_app.py | 35 ++ src/fastapi_cloud_cli/cli.py | 132 +++--- src/fastapi_cloud_cli/commands/_auth.py | 41 ++ .../commands/apps/__init__.py | 22 +- src/fastapi_cloud_cli/commands/apps/_app.py | 6 + src/fastapi_cloud_cli/commands/apps/create.py | 155 +++---- src/fastapi_cloud_cli/commands/apps/get.py | 86 ++-- src/fastapi_cloud_cli/commands/apps/link.py | 6 +- src/fastapi_cloud_cli/commands/apps/list.py | 99 ++-- src/fastapi_cloud_cli/commands/apps/unlink.py | 4 + src/fastapi_cloud_cli/commands/apps/update.py | 99 ++-- .../commands/auth/__init__.py | 13 +- src/fastapi_cloud_cli/commands/auth/_app.py | 6 + src/fastapi_cloud_cli/commands/auth/wait.py | 2 + src/fastapi_cloud_cli/commands/ci/__init__.py | 14 +- src/fastapi_cloud_cli/commands/ci/_app.py | 6 + .../commands/ci/print_workflow.py | 2 + .../commands/deploy/command.py | 3 + src/fastapi_cloud_cli/commands/deployments.py | 199 ++++---- .../commands/domains/__init__.py | 18 +- .../commands/domains/_app.py | 6 + src/fastapi_cloud_cli/commands/domains/add.py | 112 +++-- src/fastapi_cloud_cli/commands/domains/get.py | 130 +++--- .../commands/domains/list.py | 65 ++- .../commands/domains/remove.py | 197 ++++---- .../commands/domains/restart.py | 171 ++++--- .../commands/env/__init__.py | 16 +- src/fastapi_cloud_cli/commands/env/_app.py | 6 + src/fastapi_cloud_cli/commands/env/delete.py | 178 ++++---- src/fastapi_cloud_cli/commands/env/get.py | 124 +++-- src/fastapi_cloud_cli/commands/env/list.py | 54 +-- src/fastapi_cloud_cli/commands/env/set.py | 101 ++--- .../integrations/providers/__init__.py | 10 +- .../commands/integrations/providers/_app.py | 6 + .../commands/integrations/providers/list.py | 71 ++- .../integrations/resources/__init__.py | 18 +- .../commands/integrations/resources/_app.py | 6 + .../integrations/resources/connect.py | 113 +++-- .../integrations/resources/disconnect.py | 236 +++++----- .../commands/integrations/resources/get.py | 70 ++- .../commands/integrations/resources/list.py | 63 ++- src/fastapi_cloud_cli/commands/login.py | 5 + src/fastapi_cloud_cli/commands/logout.py | 2 + src/fastapi_cloud_cli/commands/logs.py | 68 ++- src/fastapi_cloud_cli/commands/setup_ci.py | 427 +++++++++--------- .../commands/teams/__init__.py | 139 +----- src/fastapi_cloud_cli/commands/teams/_app.py | 6 + src/fastapi_cloud_cli/commands/teams/get.py | 50 +- src/fastapi_cloud_cli/commands/teams/list.py | 119 +++++ .../commands/tokens/__init__.py | 14 +- src/fastapi_cloud_cli/commands/tokens/_app.py | 6 + .../commands/tokens/create.py | 90 ++-- .../commands/tokens/delete.py | 87 ++-- src/fastapi_cloud_cli/commands/tokens/list.py | 57 ++- src/fastapi_cloud_cli/commands/whoami.py | 54 ++- tests/integrations/providers/test_list.py | 2 +- tests/integrations/resources/test_connect.py | 2 +- .../integrations/resources/test_disconnect.py | 2 +- tests/integrations/resources/test_get.py | 2 +- tests/integrations/resources/test_list.py | 2 +- tests/test_cli_apps.py | 8 +- tests/test_cli_deployments.py | 6 +- tests/test_cli_link.py | 2 +- tests/test_cli_teams.py | 6 +- tests/test_cli_tokens.py | 6 +- tests/test_cli_whoami.py | 6 +- 66 files changed, 1865 insertions(+), 2004 deletions(-) create mode 100644 src/fastapi_cloud_cli/_app.py create mode 100644 src/fastapi_cloud_cli/commands/_auth.py create mode 100644 src/fastapi_cloud_cli/commands/apps/_app.py create mode 100644 src/fastapi_cloud_cli/commands/auth/_app.py create mode 100644 src/fastapi_cloud_cli/commands/ci/_app.py create mode 100644 src/fastapi_cloud_cli/commands/domains/_app.py create mode 100644 src/fastapi_cloud_cli/commands/env/_app.py create mode 100644 src/fastapi_cloud_cli/commands/integrations/providers/_app.py create mode 100644 src/fastapi_cloud_cli/commands/integrations/resources/_app.py create mode 100644 src/fastapi_cloud_cli/commands/teams/_app.py create mode 100644 src/fastapi_cloud_cli/commands/teams/list.py create mode 100644 src/fastapi_cloud_cli/commands/tokens/_app.py diff --git a/src/fastapi_cloud_cli/_app.py b/src/fastapi_cloud_cli/_app.py new file mode 100644 index 00000000..482b8a8b --- /dev/null +++ b/src/fastapi_cloud_cli/_app.py @@ -0,0 +1,35 @@ +from typing import Annotated + +import typer +from rich import print + +from fastapi_cloud_cli import __version__ + +app = typer.Typer(rich_markup_mode="rich") + + +def version_callback(value: bool) -> None: + if value: + print(f"FastAPI Cloud CLI version: [green]{__version__}[/green]") + raise typer.Exit() + + +cloud_app = typer.Typer( + rich_markup_mode="rich", + help="Manage [bold]FastAPI[/bold] Cloud deployments.", + no_args_is_help=True, +) + + +@cloud_app.callback() +def cloud_main( + version: Annotated[ + bool, + typer.Option( + "--version", + callback=version_callback, + is_eager=True, + help="Show the version and exit.", + ), + ] = False, +) -> None: ... diff --git a/src/fastapi_cloud_cli/cli.py b/src/fastapi_cloud_cli/cli.py index b6e67305..3f4a9cc0 100644 --- a/src/fastapi_cloud_cli/cli.py +++ b/src/fastapi_cloud_cli/cli.py @@ -1,74 +1,64 @@ -from typing import Annotated - -import typer -from rich import print - -from . import __version__ -from .commands.apps import apps_app -from .commands.apps.link import link_app -from .commands.apps.unlink import unlink_app -from .commands.auth import auth_app -from .commands.ci import ci_app -from .commands.deploy import deploy -from .commands.deployments import deployments_app -from .commands.domains import domains_app -from .commands.env import env_app -from .commands.integrations import integrations_app -from .commands.login import login -from .commands.logout import logout -from .commands.logs import logs -from .commands.setup_ci import setup_ci -from .commands.teams import teams_app -from .commands.tokens import tokens_app -from .commands.whoami import whoami -from .logging import setup_logging -from .utils.sentry import init_sentry - -setup_logging() - -app = typer.Typer(rich_markup_mode="rich") - - -def version_callback(value: bool) -> None: - if value: - print(f"FastAPI Cloud CLI version: [green]{__version__}[/green]") - raise typer.Exit() - - -cloud_app = typer.Typer( - rich_markup_mode="rich", - help="Manage [bold]FastAPI[/bold] Cloud deployments.", - no_args_is_help=True, +# Import decorated callbacks so they register themselves with their Typer apps. +from fastapi_cloud_cli._app import app as app +from fastapi_cloud_cli._app import cloud_app as cloud_app +from fastapi_cloud_cli.commands.apps._app import apps_app +from fastapi_cloud_cli.commands.apps.create import create_app as create_app +from fastapi_cloud_cli.commands.apps.get import get_app as get_app +from fastapi_cloud_cli.commands.apps.link import link_app as link_app +from fastapi_cloud_cli.commands.apps.list import list_apps as list_apps +from fastapi_cloud_cli.commands.apps.unlink import unlink_app as unlink_app +from fastapi_cloud_cli.commands.apps.update import update_app as update_app +from fastapi_cloud_cli.commands.auth._app import auth_app +from fastapi_cloud_cli.commands.auth.wait import wait as wait +from fastapi_cloud_cli.commands.ci._app import ci_app +from fastapi_cloud_cli.commands.ci.print_workflow import ( + print_workflow as print_workflow, ) +from fastapi_cloud_cli.commands.deploy.command import deploy as deploy +from fastapi_cloud_cli.commands.deployments import deployments_app +from fastapi_cloud_cli.commands.domains._app import domains_app +from fastapi_cloud_cli.commands.domains.add import add_domain as add_domain +from fastapi_cloud_cli.commands.domains.get import get_domain as get_domain +from fastapi_cloud_cli.commands.domains.list import list_domains as list_domains +from fastapi_cloud_cli.commands.domains.remove import remove_domain as remove_domain +from fastapi_cloud_cli.commands.domains.restart import restart_domain as restart_domain +from fastapi_cloud_cli.commands.env._app import env_app +from fastapi_cloud_cli.commands.env.delete import delete as delete +from fastapi_cloud_cli.commands.env.get import get_variable as get_variable +from fastapi_cloud_cli.commands.env.list import list_variables as list_variables +from fastapi_cloud_cli.commands.env.set import set as set +from fastapi_cloud_cli.commands.integrations import integrations_app +from fastapi_cloud_cli.commands.integrations.providers.list import ( + list_providers as list_providers, +) +from fastapi_cloud_cli.commands.integrations.resources.connect import ( + connect_resource as connect_resource, +) +from fastapi_cloud_cli.commands.integrations.resources.disconnect import ( + disconnect_resource as disconnect_resource, +) +from fastapi_cloud_cli.commands.integrations.resources.get import ( + get_resource as get_resource, +) +from fastapi_cloud_cli.commands.integrations.resources.list import ( + list_resources as list_resources, +) +from fastapi_cloud_cli.commands.login import login as login +from fastapi_cloud_cli.commands.logout import logout as logout +from fastapi_cloud_cli.commands.logs import logs as logs +from fastapi_cloud_cli.commands.setup_ci import setup_ci as setup_ci +from fastapi_cloud_cli.commands.teams._app import teams_app +from fastapi_cloud_cli.commands.teams.get import get_team as get_team +from fastapi_cloud_cli.commands.teams.list import list_teams as list_teams +from fastapi_cloud_cli.commands.tokens._app import tokens_app +from fastapi_cloud_cli.commands.tokens.create import create_token as create_token +from fastapi_cloud_cli.commands.tokens.delete import delete_token as delete_token +from fastapi_cloud_cli.commands.tokens.list import list_tokens as list_tokens +from fastapi_cloud_cli.commands.whoami import whoami as whoami +from fastapi_cloud_cli.logging import setup_logging +from fastapi_cloud_cli.utils.sentry import init_sentry - -@cloud_app.callback() -def cloud_main( - version: Annotated[ - bool, - typer.Option( - "--version", - callback=version_callback, - is_eager=True, - help="Show the version and exit.", - ), - ] = False, -) -> None: ... - - -# TODO: use the app structure - -# Additional commands - -# fastapi cloud [command] -cloud_app.command()(deploy) -cloud_app.command("link")(link_app) -cloud_app.command()(login) -cloud_app.command()(logs) -cloud_app.command()(logout) -cloud_app.command()(whoami) -cloud_app.command("unlink")(unlink_app) -cloud_app.command()(setup_ci) +setup_logging() cloud_app.add_typer(env_app, name="env") cloud_app.add_typer(auth_app, name="auth") @@ -80,10 +70,6 @@ def cloud_main( cloud_app.add_typer(teams_app, name="teams") cloud_app.add_typer(tokens_app, name="tokens") -# fastapi [command] -app.command()(deploy) -app.command()(login) - app.add_typer(cloud_app, name="cloud") diff --git a/src/fastapi_cloud_cli/commands/_auth.py b/src/fastapi_cloud_cli/commands/_auth.py new file mode 100644 index 00000000..b9a750a5 --- /dev/null +++ b/src/fastapi_cloud_cli/commands/_auth.py @@ -0,0 +1,41 @@ +from dataclasses import dataclass +from typing import Any, cast + +from typer._click import Context +from typer.core import TyperCommand + +from fastapi_cloud_cli.utils.auth import Identity +from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit, get_rich_toolkit + +_CONTEXT_KEY = "fastapi_cloud_cli.user_command" + + +@dataclass(frozen=True) +class UserCommandContext: + toolkit: FastAPIRichToolkit + identity: Identity + + +def get_user_command_context(ctx: Context) -> UserCommandContext: + return cast(UserCommandContext, ctx.meta[_CONTEXT_KEY]) + + +class UserCommand(TyperCommand): + def invoke(self, ctx: Context) -> Any: + json_output = bool(ctx.params.get("json_output", False)) + + with get_rich_toolkit(json_output=json_output) as toolkit: + identity = Identity() + + if not identity.is_logged_in(): + toolkit.fail( + "not_logged_in", + "No credentials found.", + hint="Run `fastapi cloud login`.", + ) + + ctx.meta[_CONTEXT_KEY] = UserCommandContext( + toolkit=toolkit, + identity=identity, + ) + return super().invoke(ctx) diff --git a/src/fastapi_cloud_cli/commands/apps/__init__.py b/src/fastapi_cloud_cli/commands/apps/__init__.py index 628e49a5..341c68c5 100644 --- a/src/fastapi_cloud_cli/commands/apps/__init__.py +++ b/src/fastapi_cloud_cli/commands/apps/__init__.py @@ -1,23 +1,3 @@ -import typer - -from fastapi_cloud_cli.commands.apps.create import create_app -from fastapi_cloud_cli.commands.apps.get import get_app -from fastapi_cloud_cli.commands.apps.link import link_app -from fastapi_cloud_cli.commands.apps.list import list_apps -from fastapi_cloud_cli.commands.apps.unlink import unlink_app -from fastapi_cloud_cli.commands.apps.update import update_app -from fastapi_cloud_cli.commands.logs import logs - -apps_app = typer.Typer( - no_args_is_help=True, - help="Manage your FastAPI Cloud apps.", -) -apps_app.command("create")(create_app) -apps_app.command("get")(get_app) -apps_app.command("link")(link_app) -apps_app.command("list")(list_apps) -apps_app.command("logs")(logs) -apps_app.command("unlink")(unlink_app) -apps_app.command("update")(update_app) +from fastapi_cloud_cli.commands.apps._app import apps_app as apps_app __all__ = ["apps_app"] diff --git a/src/fastapi_cloud_cli/commands/apps/_app.py b/src/fastapi_cloud_cli/commands/apps/_app.py new file mode 100644 index 00000000..4aa8e997 --- /dev/null +++ b/src/fastapi_cloud_cli/commands/apps/_app.py @@ -0,0 +1,6 @@ +import typer + +apps_app = typer.Typer( + no_args_is_help=True, + help="Manage your FastAPI Cloud apps.", +) diff --git a/src/fastapi_cloud_cli/commands/apps/create.py b/src/fastapi_cloud_cli/commands/apps/create.py index 0df7cc01..126a1cb2 100644 --- a/src/fastapi_cloud_cli/commands/apps/create.py +++ b/src/fastapi_cloud_cli/commands/apps/create.py @@ -7,13 +7,13 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.apps._app import apps_app from fastapi_cloud_cli.commands.deploy.archive import ( _get_app_name, validate_app_directory, ) from fastapi_cloud_cli.utils.apps import AppConfig, write_app_config -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 from fastapi_cloud_cli.utils.teams import select_team @@ -56,7 +56,9 @@ def _render_apps_create_output(data: AppsCreateOutput, toolkit: RichToolkit) -> ) +@apps_app.command("create", cls=UserCommand) def create_app( + ctx: typer.Context, team_id: Annotated[ str | None, typer.Option( @@ -100,97 +102,90 @@ def create_app( """ Create a FastAPI Cloud app. """ - identity = Identity() path_to_link = path or Path.cwd() # JSON output is non-interactive, so it defaults to create-only unless --link is explicit. link_app = link if link is not None else not json_output - 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.", - ) + toolkit = get_user_command_context(ctx).toolkit - if not link_app and path is not None: - toolkit.fail( - "invalid_input", - "Path can only be used when linking.", - hint="Pass --link or omit --path.", - ) + if not link_app and path is not None: + toolkit.fail( + "invalid_input", + "Path can only be used when linking.", + hint="Pass --link or omit --path.", + ) - with APIClient() as client: - if team_id is None: - if json_output: - toolkit.fail( - "missing_required_input", - "Team ID is required.", - hint="Pass --team-id to choose a team.", - ) - - team = select_team( - toolkit, - client, - empty_hint="Create a team before listing apps.", - ) - team_id = team.id - toolkit.print_line() - - if name is None: - if json_output: - toolkit.fail( - "missing_required_input", - "App name is required.", - hint="Pass --name to choose an app name.", - ) - - name = toolkit.input( - title="What's your app name?", - default=_get_app_name(path_to_link), - bullet=False, + with APIClient() as client: + if team_id is None: + if json_output: + toolkit.fail( + "missing_required_input", + "Team ID is required.", + hint="Pass --team-id to choose a team.", ) - toolkit.print_line() - try: - directory = validate_app_directory(directory) - except ValueError as e: + team = select_team( + toolkit, + client, + empty_hint="Create a team before listing apps.", + ) + team_id = team.id + toolkit.print_line() + + if name is None: + if json_output: toolkit.fail( - "invalid_input", - f"Invalid app directory: {e}", - hint=( - "Pass a relative app directory such as `backend` or `webserver`; " - "use --path with --link to choose a local filesystem path." - ), + "missing_required_input", + "App name is required.", + hint="Pass --name to choose an app name.", ) - with toolkit.progress( - title="Creating app", - transient=True, - ) as progress: - with client.handle_http_errors( - progress, - default_message="Error creating app. Please try again later.", - toolkit=toolkit, - ): - app = _create_app( - client, - team_id=team_id, - name=name, - directory=directory, - ) - - if link_app: - write_app_config( - path_to_link, - AppConfig(app_id=app.id, team_id=app.team_id), + name = toolkit.input( + title="What's your app name?", + default=_get_app_name(path_to_link), + bullet=False, + ) + toolkit.print_line() + + try: + directory = validate_app_directory(directory) + except ValueError as e: + toolkit.fail( + "invalid_input", + f"Invalid app directory: {e}", + hint=( + "Pass a relative app directory such as `backend` or `webserver`; " + "use --path with --link to choose a local filesystem path." + ), ) - result = AppsCreateOutput( - app=app, - linked=link_app, - path_to_link=path_to_link if link_app else None, + with toolkit.progress( + title="Creating app", + transient=True, + ) as progress: + with client.handle_http_errors( + progress, + default_message="Error creating app. Please try again later.", + toolkit=toolkit, + ): + app = _create_app( + client, + team_id=team_id, + name=name, + directory=directory, + ) + + if link_app: + write_app_config( + path_to_link, + AppConfig(app_id=app.id, team_id=app.team_id), ) - toolkit.success(result, render_output=_render_apps_create_output) + result = AppsCreateOutput( + app=app, + linked=link_app, + path_to_link=path_to_link if link_app else None, + ) + + toolkit.success(result, render_output=_render_apps_create_output) diff --git a/src/fastapi_cloud_cli/commands/apps/get.py b/src/fastapi_cloud_cli/commands/apps/get.py index 8f3c6245..07e6b25a 100644 --- a/src/fastapi_cloud_cli/commands/apps/get.py +++ b/src/fastapi_cloud_cli/commands/apps/get.py @@ -7,6 +7,8 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.apps._app import apps_app from fastapi_cloud_cli.commands.apps.list import ( App, _get_app, @@ -15,8 +17,7 @@ ) from fastapi_cloud_cli.config import Settings 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.cli import get_details_table from fastapi_cloud_cli.utils.execution import JsonOutputOption logger = logging.getLogger(__name__) @@ -56,7 +57,9 @@ def _render_app_get_output(data: AppGetOutput, toolkit: RichToolkit) -> None: ) +@apps_app.command("get", cls=UserCommand) def get_app( + ctx: typer.Context, app_id: Annotated[ str | None, typer.Argument( @@ -68,55 +71,48 @@ def get_app( """ Get a FastAPI Cloud app by ID. """ - 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.", - ) - target_app_id = resolve_app_id_or_fail( - toolkit, - app_id=app_id, - hint="Pass an app ID or run `fastapi cloud apps create --link` first.", - ) + toolkit = get_user_command_context(ctx).toolkit + + target_app_id = resolve_app_id_or_fail( + toolkit, + app_id=app_id, + hint="Pass an app ID or run `fastapi cloud apps create --link` first.", + ) - with APIClient() as client: + with APIClient() as client: + with toolkit.progress( + title="Fetching app", + transient=True, + ) as progress: + with 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, target_app_id) + + dashboard_url = None + if not json_output: with toolkit.progress( - title="Fetching app", + title="Fetching team", transient=True, ) as progress: with client.handle_http_errors( progress, - default_message="Error fetching app. Please try again later.", - not_found_message="App not found.", + default_message="Error fetching team. Please try again later.", + not_found_message="Team not found.", toolkit=toolkit, ): - app = _get_app(client, target_app_id) - - dashboard_url = None - if not json_output: - with toolkit.progress( - title="Fetching team", - transient=True, - ) as progress: - with 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) - - dashboard_url = _get_app_dashboard_url( - app, - team_slug=team.slug, - settings=Settings.get(), - ) - - result = AppGetOutput(app=app, dashboard_url=dashboard_url) - - toolkit.success(result, render_output=_render_app_get_output) + team = _get_team(client, app.team_id) + + dashboard_url = _get_app_dashboard_url( + app, + team_slug=team.slug, + settings=Settings.get(), + ) + + result = AppGetOutput(app=app, dashboard_url=dashboard_url) + + toolkit.success(result, render_output=_render_app_get_output) diff --git a/src/fastapi_cloud_cli/commands/apps/link.py b/src/fastapi_cloud_cli/commands/apps/link.py index 61082828..5f52ae72 100644 --- a/src/fastapi_cloud_cli/commands/apps/link.py +++ b/src/fastapi_cloud_cli/commands/apps/link.py @@ -7,7 +7,9 @@ from rich_toolkit import RichToolkit from rich_toolkit.menu import Option +from fastapi_cloud_cli._app import cloud_app from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands.apps._app import apps_app from fastapi_cloud_cli.commands.apps.list import _get_app from fastapi_cloud_cli.utils.apps import AppConfig, get_app_config, write_app_config from fastapi_cloud_cli.utils.auth import Identity @@ -171,6 +173,8 @@ def _link_app_interactively( logger.debug(f"Linked to app: {app['id']} in team: {team['id']}") +@cloud_app.command("link") +@apps_app.command("link") def link_app( app_id: Annotated[ str | None, @@ -216,7 +220,7 @@ def link_app( toolkit.fail( "not_logged_in", "No credentials found.", - hint="Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + hint="Run `fastapi cloud login`.", ) if app_id is not None and app_id_option is not None and app_id != app_id_option: diff --git a/src/fastapi_cloud_cli/commands/apps/list.py b/src/fastapi_cloud_cli/commands/apps/list.py index 3ede78fa..5e9caa3e 100644 --- a/src/fastapi_cloud_cli/commands/apps/list.py +++ b/src/fastapi_cloud_cli/commands/apps/list.py @@ -8,9 +8,9 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.apps._app import apps_app from fastapi_cloud_cli.config import Settings -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 from fastapi_cloud_cli.utils.teams import Team, select_team @@ -134,7 +134,9 @@ def _render_apps_list_output(data: AppsListOutput, toolkit: RichToolkit) -> None ) +@apps_app.command("list", cls=UserCommand) def list_apps( + ctx: typer.Context, team_id: Annotated[ str | None, typer.Option( @@ -163,65 +165,58 @@ def list_apps( """ List FastAPI Cloud apps. """ - 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: - team_slug: str | None = None + toolkit = get_user_command_context(ctx).toolkit - if team_id is None: - if json_output: - toolkit.fail( - "missing_required_input", - "Team ID is required.", - hint="Pass --team-id to choose a team.", - ) + with APIClient() as client: + team_slug: str | None = None - team = select_team( - toolkit, - client, - empty_hint="Create a team before listing apps.", + if team_id is None: + if json_output: + toolkit.fail( + "missing_required_input", + "Team ID is required.", + hint="Pass --team-id to choose a team.", ) - team_id = team.id - team_slug = team.slug - - toolkit.print_line() - else: - with toolkit.progress( - title="Fetching team", - transient=True, - ) as progress: - with 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, team_id) - team_slug = team.slug + team = select_team( + toolkit, + client, + empty_hint="Create a team before listing apps.", + ) + team_id = team.id + team_slug = team.slug + + toolkit.print_line() + else: with toolkit.progress( - title="Fetching apps", + title="Fetching team", transient=True, ) as progress: with client.handle_http_errors( progress, - default_message="Error fetching apps. Please try again later.", + default_message="Error fetching team. Please try again later.", + not_found_message="Team not found.", toolkit=toolkit, ): - result = _get_apps( - client, - team_id=team_id, - limit=limit, - offset=offset, - team_slug=team_slug, - ) - - toolkit.success(result, render_output=_render_apps_list_output) + team = _get_team(client, team_id) + team_slug = team.slug + + with toolkit.progress( + title="Fetching apps", + transient=True, + ) as progress: + with client.handle_http_errors( + progress, + default_message="Error fetching apps. Please try again later.", + toolkit=toolkit, + ): + result = _get_apps( + client, + team_id=team_id, + limit=limit, + offset=offset, + team_slug=team_slug, + ) + + toolkit.success(result, render_output=_render_apps_list_output) diff --git a/src/fastapi_cloud_cli/commands/apps/unlink.py b/src/fastapi_cloud_cli/commands/apps/unlink.py index c6e8e274..c7643a64 100644 --- a/src/fastapi_cloud_cli/commands/apps/unlink.py +++ b/src/fastapi_cloud_cli/commands/apps/unlink.py @@ -7,6 +7,8 @@ from rich.text import Text from rich_toolkit import RichToolkit +from fastapi_cloud_cli._app import cloud_app +from fastapi_cloud_cli.commands.apps._app import apps_app from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit, get_rich_toolkit from fastapi_cloud_cli.utils.execution import JsonOutputOption @@ -40,6 +42,8 @@ def _fail_not_linked(toolkit: FastAPIRichToolkit) -> None: ) +@cloud_app.command("unlink") +@apps_app.command("unlink") def unlink_app( path: Annotated[ Path | None, diff --git a/src/fastapi_cloud_cli/commands/apps/update.py b/src/fastapi_cloud_cli/commands/apps/update.py index 6bdb0f4b..59ceb08a 100644 --- a/src/fastapi_cloud_cli/commands/apps/update.py +++ b/src/fastapi_cloud_cli/commands/apps/update.py @@ -6,10 +6,10 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.apps._app import apps_app from fastapi_cloud_cli.commands.deploy.archive import validate_app_directory 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__) @@ -45,7 +45,9 @@ def _render_apps_update_output(data: AppsUpdateOutput, toolkit: RichToolkit) -> ) +@apps_app.command("update", cls=UserCommand) def update_app( + ctx: typer.Context, app_id: Annotated[ str | None, typer.Argument( @@ -67,56 +69,49 @@ def update_app( """ Update FastAPI Cloud app metadata. """ - 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.", - ) - - if directory is None: - toolkit.fail( - "missing_required_input", - "No updates provided.", - hint="Pass --directory to update the app directory.", - ) - - target_app_id = resolve_app_id_or_fail( - toolkit, - app_id=app_id, - hint="Pass an app ID or run `fastapi cloud apps create --link` first.", + + toolkit = get_user_command_context(ctx).toolkit + + if directory is None: + toolkit.fail( + "missing_required_input", + "No updates provided.", + hint="Pass --directory to update the app directory.", ) - try: - directory = validate_app_directory(directory) - except ValueError as e: - toolkit.fail( - "invalid_input", - f"Invalid app directory: {e}", - hint="Pass a relative app directory such as `src` or `backend`.", - ) - - with APIClient() as client: - with toolkit.progress( - title="Updating app", - transient=True, - ) as progress: - with client.handle_http_errors( - progress, - default_message="Error updating app. Please try again later.", - not_found_message="App not found.", - toolkit=toolkit, - ): - app = _update_app( - client, - app_id=target_app_id, - directory=directory, - ) - - toolkit.success( - AppsUpdateOutput(app=app), - render_output=_render_apps_update_output, + target_app_id = resolve_app_id_or_fail( + toolkit, + app_id=app_id, + hint="Pass an app ID or run `fastapi cloud apps create --link` first.", + ) + + try: + directory = validate_app_directory(directory) + except ValueError as e: + toolkit.fail( + "invalid_input", + f"Invalid app directory: {e}", + hint="Pass a relative app directory such as `src` or `backend`.", ) + + with APIClient() as client: + with toolkit.progress( + title="Updating app", + transient=True, + ) as progress: + with client.handle_http_errors( + progress, + default_message="Error updating app. Please try again later.", + not_found_message="App not found.", + toolkit=toolkit, + ): + app = _update_app( + client, + app_id=target_app_id, + directory=directory, + ) + + toolkit.success( + AppsUpdateOutput(app=app), + render_output=_render_apps_update_output, + ) diff --git a/src/fastapi_cloud_cli/commands/auth/__init__.py b/src/fastapi_cloud_cli/commands/auth/__init__.py index c785c8ee..6667a562 100644 --- a/src/fastapi_cloud_cli/commands/auth/__init__.py +++ b/src/fastapi_cloud_cli/commands/auth/__init__.py @@ -1,14 +1,3 @@ -import typer - -from fastapi_cloud_cli.commands.auth import wait as wait_command -from fastapi_cloud_cli.commands.login import login - -auth_app = typer.Typer( - no_args_is_help=True, - help="Authenticate with FastAPI Cloud.", -) - -auth_app.command()(login) -auth_app.command("wait")(wait_command.wait) +from fastapi_cloud_cli.commands.auth._app import auth_app as auth_app __all__ = ["auth_app"] diff --git a/src/fastapi_cloud_cli/commands/auth/_app.py b/src/fastapi_cloud_cli/commands/auth/_app.py new file mode 100644 index 00000000..12b3040e --- /dev/null +++ b/src/fastapi_cloud_cli/commands/auth/_app.py @@ -0,0 +1,6 @@ +import typer + +auth_app = typer.Typer( + no_args_is_help=True, + help="Authenticate with FastAPI Cloud.", +) diff --git a/src/fastapi_cloud_cli/commands/auth/wait.py b/src/fastapi_cloud_cli/commands/auth/wait.py index a8d8623b..ff7a0c18 100644 --- a/src/fastapi_cloud_cli/commands/auth/wait.py +++ b/src/fastapi_cloud_cli/commands/auth/wait.py @@ -8,10 +8,12 @@ complete_device_login, render_login_output, ) +from fastapi_cloud_cli.commands.auth._app import auth_app from fastapi_cloud_cli.utils.cli import get_rich_toolkit from fastapi_cloud_cli.utils.execution import JsonOutputOption +@auth_app.command("wait") def wait( device_code: Annotated[ str, diff --git a/src/fastapi_cloud_cli/commands/ci/__init__.py b/src/fastapi_cloud_cli/commands/ci/__init__.py index 7c7cf8c2..883ed76f 100644 --- a/src/fastapi_cloud_cli/commands/ci/__init__.py +++ b/src/fastapi_cloud_cli/commands/ci/__init__.py @@ -1,15 +1,3 @@ -import typer - -from fastapi_cloud_cli.commands.ci.print_workflow import ( - print_workflow as print_workflow_command, -) -from fastapi_cloud_cli.commands.setup_ci import setup_ci - -ci_app = typer.Typer( - no_args_is_help=True, - help="Manage CI integration helpers.", -) -ci_app.command("print-workflow")(print_workflow_command) -ci_app.command("setup")(setup_ci) +from fastapi_cloud_cli.commands.ci._app import ci_app as ci_app __all__ = ["ci_app"] diff --git a/src/fastapi_cloud_cli/commands/ci/_app.py b/src/fastapi_cloud_cli/commands/ci/_app.py new file mode 100644 index 00000000..457da122 --- /dev/null +++ b/src/fastapi_cloud_cli/commands/ci/_app.py @@ -0,0 +1,6 @@ +import typer + +ci_app = typer.Typer( + no_args_is_help=True, + help="Manage CI integration helpers.", +) diff --git a/src/fastapi_cloud_cli/commands/ci/print_workflow.py b/src/fastapi_cloud_cli/commands/ci/print_workflow.py index d3f152cd..89d212c2 100644 --- a/src/fastapi_cloud_cli/commands/ci/print_workflow.py +++ b/src/fastapi_cloud_cli/commands/ci/print_workflow.py @@ -4,6 +4,7 @@ from pydantic import BaseModel from rich_toolkit import RichToolkit +from fastapi_cloud_cli.commands.ci._app import ci_app from fastapi_cloud_cli.commands.setup_ci import ( DEFAULT_WORKFLOW_PATH, _get_default_branch, @@ -22,6 +23,7 @@ def _render_workflow_output(data: CIWorkflowOutput, toolkit: RichToolkit) -> Non toolkit.console.print(data.content, markup=False, end="") +@ci_app.command("print-workflow") def print_workflow( branch: Annotated[ str | None, diff --git a/src/fastapi_cloud_cli/commands/deploy/command.py b/src/fastapi_cloud_cli/commands/deploy/command.py index ff3c899c..b5fecd0d 100644 --- a/src/fastapi_cloud_cli/commands/deploy/command.py +++ b/src/fastapi_cloud_cli/commands/deploy/command.py @@ -6,6 +6,7 @@ import typer from pydantic import BaseModel +from fastapi_cloud_cli._app import app, cloud_app from fastapi_cloud_cli.api import APIClient, DeploymentStatus from fastapi_cloud_cli.commands.deploy.archive import _get_large_files, archive from fastapi_cloud_cli.commands.deploy.cloud import ( @@ -98,6 +99,8 @@ def _render_linked_app_not_found( ) +@app.command() +@cloud_app.command() def deploy( path: Annotated[ Path | None, diff --git a/src/fastapi_cloud_cli/commands/deployments.py b/src/fastapi_cloud_cli/commands/deployments.py index 651188cb..92fbd254 100644 --- a/src/fastapi_cloud_cli/commands/deployments.py +++ b/src/fastapi_cloud_cli/commands/deployments.py @@ -19,12 +19,11 @@ get_http_error_hint, handle_http_error, ) +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context 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 ( FastAPIRichToolkit, get_details_table, - get_rich_toolkit, ) from fastapi_cloud_cli.utils.dates import format_last_updated from fastapi_cloud_cli.utils.errors import ErrorCode @@ -312,8 +311,9 @@ def _render_build_log_error( ) -@deployments_app.command("get") +@deployments_app.command("get", cls=UserCommand) def get_deployment( + ctx: typer.Context, deployment_id: Annotated[ str, typer.Argument( @@ -332,39 +332,33 @@ def get_deployment( """ Get a FastAPI Cloud deployment by ID. """ - 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.", - ) - - resolve_app_id_or_fail(toolkit, app_id=app_id) - - with APIClient() as client: - with toolkit.progress( - title="Fetching deployment", - transient=True, - ) as progress: - with client.handle_http_errors( - progress, - default_message="Error fetching deployment. Please try again later.", - not_found_message="Deployment not found.", - toolkit=toolkit, - ): - result = _get_deployment( - client, - deployment_id=deployment_id, - ) + toolkit = get_user_command_context(ctx).toolkit + + resolve_app_id_or_fail(toolkit, app_id=app_id) + + with APIClient() as client: + with toolkit.progress( + title="Fetching deployment", + transient=True, + ) as progress: + with client.handle_http_errors( + progress, + default_message="Error fetching deployment. Please try again later.", + not_found_message="Deployment not found.", + toolkit=toolkit, + ): + result = _get_deployment( + client, + deployment_id=deployment_id, + ) - toolkit.success(result, render_output=_render_deployment_get_output) + toolkit.success(result, render_output=_render_deployment_get_output) -@deployments_app.command("build-logs") +@deployments_app.command("build-logs", cls=UserCommand) def build_logs( + ctx: typer.Context, deployment_id: Annotated[ str, typer.Argument( @@ -384,58 +378,52 @@ def build_logs( """ Stream or fetch build logs for a FastAPI Cloud deployment. """ - 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.", - ) - - if follow: - toolkit.print( - f"Streaming build logs for [bold]{deployment_id}[/bold]...", - emoji="📡", - ) - else: - toolkit.print( - f"Fetching build logs for [bold]{deployment_id}[/bold]...", - emoji="📜", - ) + + toolkit = get_user_command_context(ctx).toolkit + + if follow: + toolkit.print( + f"Streaming build logs for [bold]{deployment_id}[/bold]...", + emoji="📡", + ) + else: + toolkit.print( + f"Fetching build logs for [bold]{deployment_id}[/bold]...", + emoji="📜", + ) + toolkit.print_line() + + try: + with APIClient() as client: + if follow: + failed = _stream_build_logs(toolkit, client, deployment_id) + else: + result = _fetch_build_logs(client, deployment_id) + toolkit.success(result, render_output=_render_build_logs_output) + failed = result.failed + + except KeyboardInterrupt: # pragma: no cover toolkit.print_line() + return + except StreamLogError as e: + _handle_build_log_error(toolkit, e) + + except (TooManyRetriesError, TimeoutError): + message = "Lost connection to build log stream. Please try again later." + toolkit.fail( + "network_error", + message, + hint="Please try again later.", + render_output=_render_build_log_error, + ) + + if failed: + raise typer.Exit(1) - try: - with APIClient() as client: - if follow: - failed = _stream_build_logs(toolkit, client, deployment_id) - else: - result = _fetch_build_logs(client, deployment_id) - toolkit.success(result, render_output=_render_build_logs_output) - failed = result.failed - - except KeyboardInterrupt: # pragma: no cover - toolkit.print_line() - return - except StreamLogError as e: - _handle_build_log_error(toolkit, e) - - except (TooManyRetriesError, TimeoutError): - message = "Lost connection to build log stream. Please try again later." - toolkit.fail( - "network_error", - message, - hint="Please try again later.", - render_output=_render_build_log_error, - ) - - if failed: - raise typer.Exit(1) - - -@deployments_app.command("list") + +@deployments_app.command("list", cls=UserCommand) def list_deployments( + ctx: typer.Context, app_id: Annotated[ str | None, typer.Option( @@ -464,34 +452,27 @@ def list_deployments( """ List FastAPI Cloud deployments for 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.", - ) - - target_app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) - with APIClient() as client: - with toolkit.progress( - title="Fetching deployments", - transient=True, - ) as progress: - with client.handle_http_errors( - progress, - default_message="Error fetching deployments. Please try again later.", - not_found_message="App not found.", - toolkit=toolkit, - ): - result = _get_deployments( - client, - app_id=target_app_id, - limit=limit, - offset=offset, - ) + toolkit = get_user_command_context(ctx).toolkit + + target_app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) + + with APIClient() as client: + with toolkit.progress( + title="Fetching deployments", + transient=True, + ) as progress: + with client.handle_http_errors( + progress, + default_message="Error fetching deployments. Please try again later.", + not_found_message="App not found.", + toolkit=toolkit, + ): + result = _get_deployments( + client, + app_id=target_app_id, + limit=limit, + offset=offset, + ) - toolkit.success(result, render_output=_render_deployments_list_output) + toolkit.success(result, render_output=_render_deployments_list_output) diff --git a/src/fastapi_cloud_cli/commands/domains/__init__.py b/src/fastapi_cloud_cli/commands/domains/__init__.py index 33cabaf9..c03bf5ad 100644 --- a/src/fastapi_cloud_cli/commands/domains/__init__.py +++ b/src/fastapi_cloud_cli/commands/domains/__init__.py @@ -1,19 +1,3 @@ -import typer - -from fastapi_cloud_cli.commands.domains.add import add_domain -from fastapi_cloud_cli.commands.domains.get import get_domain -from fastapi_cloud_cli.commands.domains.list import list_domains -from fastapi_cloud_cli.commands.domains.remove import remove_domain -from fastapi_cloud_cli.commands.domains.restart import restart_domain - -domains_app = typer.Typer( - no_args_is_help=True, - help="Manage the custom domains of your app.", -) -domains_app.command("add")(add_domain) -domains_app.command("get")(get_domain) -domains_app.command("list")(list_domains) -domains_app.command("remove")(remove_domain) -domains_app.command("restart")(restart_domain) +from fastapi_cloud_cli.commands.domains._app import domains_app as domains_app __all__ = ["domains_app"] diff --git a/src/fastapi_cloud_cli/commands/domains/_app.py b/src/fastapi_cloud_cli/commands/domains/_app.py new file mode 100644 index 00000000..bdd0166a --- /dev/null +++ b/src/fastapi_cloud_cli/commands/domains/_app.py @@ -0,0 +1,6 @@ +import typer + +domains_app = typer.Typer( + no_args_is_help=True, + help="Manage the custom domains of your app.", +) diff --git a/src/fastapi_cloud_cli/commands/domains/add.py b/src/fastapi_cloud_cli/commands/domains/add.py index 7ae9f5f6..43408468 100644 --- a/src/fastapi_cloud_cli/commands/domains/add.py +++ b/src/fastapi_cloud_cli/commands/domains/add.py @@ -6,11 +6,12 @@ from rich_toolkit.menu import Option from fastapi_cloud_cli.api import APIClient, CustomDomain +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.domains._app import domains_app from fastapi_cloud_cli.commands.domains._shared import _normalize_domain_name from fastapi_cloud_cli.commands.domains.rendering import render_custom_domain_setup 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 FastAPIRichToolkit, get_rich_toolkit +from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit from fastapi_cloud_cli.utils.execution import JsonOutputOption @@ -95,7 +96,9 @@ def _render_custom_domain_add_output( render_custom_domain_setup(data.domain, toolkit) +@domains_app.command("add", cls=UserCommand) def add_domain( + ctx: typer.Context, domain: Annotated[ str | None, typer.Argument( @@ -128,67 +131,56 @@ def add_domain( """ Add a custom domain to an app. """ - identity = Identity() + toolkit = get_user_command_context(ctx).toolkit + app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) - 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`.", - ) + if standard and zero_downtime: + toolkit.fail( + "invalid_input", + "Setup modes are mutually exclusive.", + hint="Pass either --standard or --zero-downtime.", + ) - app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) + prompts_user = domain is None or not (standard or zero_downtime) + if prompts_user and toolkit.mode != "json": + toolkit.print_title("custom domains") + toolkit.print_line() - if standard and zero_downtime: - toolkit.fail( - "invalid_input", - "Setup modes are mutually exclusive.", - hint="Pass either --standard or --zero-downtime.", - ) + domain = _resolve_domain_name(toolkit, domain=domain) + is_using_pre_validation = _resolve_pre_validation( + toolkit, + domain=domain, + standard=standard, + zero_downtime=zero_downtime, + ) - prompts_user = domain is None or not (standard or zero_downtime) - if prompts_user and toolkit.mode != "json": - toolkit.print_title("custom domains") - toolkit.print_line() - - domain = _resolve_domain_name(toolkit, domain=domain) - is_using_pre_validation = _resolve_pre_validation( - toolkit, - domain=domain, - standard=standard, - zero_downtime=zero_downtime, - ) + if prompts_user: + toolkit.print_line() - if prompts_user: - toolkit.print_line() - - with APIClient() as client: - with ( - toolkit.progress( - title="Adding custom domain", - transient=True, - ) as progress, - client.handle_http_errors( - progress, - default_message=( - "Error adding custom domain. Please try again later." - ), - not_found_message="App not found.", - toolkit=toolkit, - ), - ): - created_domain = client.create_custom_domain( - app_id=app_id, - name=domain, - is_using_pre_validation=is_using_pre_validation, - ) - - toolkit.success( - CustomDomainAddOutput( - app_id=app_id, - domain=created_domain, - show_title=not prompts_user, + with APIClient() as client: + with ( + toolkit.progress( + title="Adding custom domain", + transient=True, + ) as progress, + client.handle_http_errors( + progress, + default_message=("Error adding custom domain. Please try again later."), + not_found_message="App not found.", + toolkit=toolkit, ), - render_output=_render_custom_domain_add_output, - ) + ): + created_domain = client.create_custom_domain( + app_id=app_id, + name=domain, + is_using_pre_validation=is_using_pre_validation, + ) + + toolkit.success( + CustomDomainAddOutput( + app_id=app_id, + domain=created_domain, + show_title=not prompts_user, + ), + render_output=_render_custom_domain_add_output, + ) diff --git a/src/fastapi_cloud_cli/commands/domains/get.py b/src/fastapi_cloud_cli/commands/domains/get.py index 053b4e4f..ee16b3dc 100644 --- a/src/fastapi_cloud_cli/commands/domains/get.py +++ b/src/fastapi_cloud_cli/commands/domains/get.py @@ -5,14 +5,14 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient, CustomDomain +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.domains._app import domains_app from fastapi_cloud_cli.commands.domains._shared import ( _find_custom_domain, _select_custom_domain, ) from fastapi_cloud_cli.commands.domains.rendering import render_custom_domain_details 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 @@ -33,7 +33,9 @@ def _render_custom_domain_get_output( render_custom_domain_details(data.domain, toolkit) +@domains_app.command("get", cls=UserCommand) def get_domain( + ctx: typer.Context, domain: Annotated[ str | None, typer.Argument( @@ -52,77 +54,67 @@ def get_domain( """ Get a custom domain for 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`.", - ) - - app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) - domain_was_provided = domain is not None - - if domain is None and toolkit.mode == "json": - toolkit.fail( - "missing_required_input", - "Custom domain is required.", - hint="Pass DOMAIN to choose a custom domain.", - ) + toolkit = get_user_command_context(ctx).toolkit + app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) + domain_was_provided = domain is not None + + if domain is None and toolkit.mode == "json": + toolkit.fail( + "missing_required_input", + "Custom domain is required.", + hint="Pass DOMAIN to choose a custom domain.", + ) - with APIClient() as client: - with ( - toolkit.progress( - title="Fetching custom domains", - transient=True, - ) as progress, - client.handle_http_errors( - progress, - default_message=( - "Error fetching custom domains. Please try again later." - ), - not_found_message="App not found.", - toolkit=toolkit, + with APIClient() as client: + with ( + toolkit.progress( + title="Fetching custom domains", + transient=True, + ) as progress, + client.handle_http_errors( + progress, + default_message=( + "Error fetching custom domains. Please try again later." ), - ): - domains = client.get_custom_domains(app_id=app_id).data + not_found_message="App not found.", + toolkit=toolkit, + ), + ): + domains = client.get_custom_domains(app_id=app_id).data - selected_domain: CustomDomain | None + selected_domain: CustomDomain | None - if domain is None: - toolkit.print_title("custom domains") - toolkit.print_line() + if domain is None: + toolkit.print_title("custom domains") + toolkit.print_line() - if not domains: - toolkit.print("No custom domains found.", bullet=False) - return + if not domains: + toolkit.print("No custom domains found.", bullet=False) + return - selected_domain = _select_custom_domain( - toolkit, - domains, - prompt="Select the custom domain to get:", - ) - toolkit.print_line() - else: - if (selected_domain := _find_custom_domain(domains, domain)) is None: - toolkit.fail( - "not_found", - f"Custom domain {domain} not found.", - hint=( - "Run `fastapi cloud domains list` to see available custom " - "domains." - ), - ) - - assert selected_domain is not None - - toolkit.success( - CustomDomainGetOutput( - app_id=app_id, - domain=selected_domain, - show_title=domain_was_provided, - ), - render_output=_render_custom_domain_get_output, + selected_domain = _select_custom_domain( + toolkit, + domains, + prompt="Select the custom domain to get:", ) + toolkit.print_line() + else: + if (selected_domain := _find_custom_domain(domains, domain)) is None: + toolkit.fail( + "not_found", + f"Custom domain {domain} not found.", + hint=( + "Run `fastapi cloud domains list` to see available custom domains." + ), + ) + + assert selected_domain is not None + + toolkit.success( + CustomDomainGetOutput( + app_id=app_id, + domain=selected_domain, + show_title=domain_was_provided, + ), + render_output=_render_custom_domain_get_output, + ) diff --git a/src/fastapi_cloud_cli/commands/domains/list.py b/src/fastapi_cloud_cli/commands/domains/list.py index 6dc1e570..5cd178d6 100644 --- a/src/fastapi_cloud_cli/commands/domains/list.py +++ b/src/fastapi_cloud_cli/commands/domains/list.py @@ -5,10 +5,10 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient, CustomDomain +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.domains._app import domains_app from fastapi_cloud_cli.commands.domains.rendering import get_custom_domains_table 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 @@ -32,7 +32,9 @@ def _render_custom_domains_list_output( toolkit.print(get_custom_domains_table(data.domains), bullet=False) +@domains_app.command("list", cls=UserCommand) def list_domains( + ctx: typer.Context, app_id: Annotated[ str | None, typer.Option( @@ -45,40 +47,31 @@ def list_domains( """ List custom domains for an app. """ - identity = Identity() + toolkit = get_user_command_context(ctx).toolkit + app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) - 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`.", - ) - - app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) - - with APIClient() as client: - with ( - toolkit.progress( - title="Fetching custom domains", - transient=True, - ) as progress, - client.handle_http_errors( - progress, - default_message=( - "Error fetching custom domains. Please try again later." - ), - not_found_message="App not found.", - toolkit=toolkit, + with APIClient() as client: + with ( + toolkit.progress( + title="Fetching custom domains", + transient=True, + ) as progress, + client.handle_http_errors( + progress, + default_message=( + "Error fetching custom domains. Please try again later." ), - ): - response = client.get_custom_domains(app_id=app_id) - - toolkit.success( - CustomDomainsListOutput( - app_id=app_id, - domains=response.data, - total_count=response.count, + not_found_message="App not found.", + toolkit=toolkit, ), - render_output=_render_custom_domains_list_output, - ) + ): + response = client.get_custom_domains(app_id=app_id) + + toolkit.success( + CustomDomainsListOutput( + app_id=app_id, + domains=response.data, + total_count=response.count, + ), + render_output=_render_custom_domains_list_output, + ) diff --git a/src/fastapi_cloud_cli/commands/domains/remove.py b/src/fastapi_cloud_cli/commands/domains/remove.py index 30465645..5c38b97e 100644 --- a/src/fastapi_cloud_cli/commands/domains/remove.py +++ b/src/fastapi_cloud_cli/commands/domains/remove.py @@ -5,13 +5,13 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient, CustomDomain +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.domains._app import domains_app from fastapi_cloud_cli.commands.domains._shared import ( _find_custom_domain, _select_custom_domain, ) 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 @@ -37,7 +37,9 @@ def _print_removal_warning(toolkit: RichToolkit, domain: CustomDomain) -> None: ) +@domains_app.command("remove", cls=UserCommand) def remove_domain( + ctx: typer.Context, domain: Annotated[ str | None, typer.Argument( @@ -66,116 +68,107 @@ def remove_domain( DNS records at your provider are not changed. """ - identity = Identity() + toolkit = get_user_command_context(ctx).toolkit + app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) - with get_rich_toolkit(json_output=json_output) as toolkit: - if not identity.is_logged_in(): + if toolkit.mode == "json": + if domain is None: toolkit.fail( - "not_logged_in", - "No credentials found.", - hint="Run `fastapi cloud login`.", + "missing_required_input", + "Custom domain is required.", + hint="Pass DOMAIN to choose a custom domain.", + ) + if not yes: + toolkit.fail( + "missing_required_input", + "Removal confirmation is required.", + hint="Pass --yes to confirm removal.", ) - app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) - - if toolkit.mode == "json": - if domain is None: - toolkit.fail( - "missing_required_input", - "Custom domain is required.", - hint="Pass DOMAIN to choose a custom domain.", - ) - if not yes: + with APIClient() as client: + with ( + toolkit.progress( + title="Fetching custom domains", + transient=True, + ) as progress, + client.handle_http_errors( + progress, + default_message=( + "Error fetching custom domains. Please try again later." + ), + not_found_message="App not found.", + toolkit=toolkit, + ), + ): + domains = client.get_custom_domains(app_id=app_id).data + + toolkit.print_title("custom domains") + toolkit.print_line() + + selected_domain: CustomDomain | None + if domain is None: + if not domains: + toolkit.print("No custom domains found.", bullet=False) + return + + selected_domain = _select_custom_domain( + toolkit, + domains, + prompt="Select the custom domain to remove:", + ) + toolkit.print_line() + else: + selected_domain = _find_custom_domain(domains, domain) + if selected_domain is None: toolkit.fail( - "missing_required_input", - "Removal confirmation is required.", - hint="Pass --yes to confirm removal.", + "not_found", + f"Custom domain {domain} not found.", + hint=( + "Run `fastapi cloud domains list` to see available " + "custom domains." + ), ) - with APIClient() as client: - with ( - toolkit.progress( - title="Fetching custom domains", - transient=True, - ) as progress, - client.handle_http_errors( - progress, - default_message=( - "Error fetching custom domains. Please try again later." - ), - not_found_message="App not found.", - toolkit=toolkit, - ), - ): - domains = client.get_custom_domains(app_id=app_id).data + assert selected_domain is not None + _print_removal_warning(toolkit, selected_domain) - toolkit.print_title("custom domains") + if not yes: toolkit.print_line() - - selected_domain: CustomDomain | None - if domain is None: - if not domains: - toolkit.print("No custom domains found.", bullet=False) - return - - selected_domain = _select_custom_domain( - toolkit, - domains, - prompt="Select the custom domain to remove:", - ) - toolkit.print_line() - else: - selected_domain = _find_custom_domain(domains, domain) - if selected_domain is None: - toolkit.fail( - "not_found", - f"Custom domain {domain} not found.", - hint=( - "Run `fastapi cloud domains list` to see available " - "custom domains." - ), - ) - - assert selected_domain is not None - _print_removal_warning(toolkit, selected_domain) - - if not yes: + should_remove = toolkit.confirm( + f"Remove [bold]{selected_domain.name}[/bold]?", + default=False, + bullet=False, + ) + if not should_remove: toolkit.print_line() - should_remove = toolkit.confirm( - f"Remove [bold]{selected_domain.name}[/bold]?", - default=False, - bullet=False, - ) - if not should_remove: - toolkit.print_line() - toolkit.print("Removal cancelled.", bullet=False) - raise typer.Exit(0) - - toolkit.print_line() - with ( - toolkit.progress( - title="Removing custom domain", - transient=True, - ) as progress, - client.handle_http_errors( - progress, - default_message=( - "Error removing custom domain. Please try again later." - ), - not_found_message="Custom domain not found.", - toolkit=toolkit, + toolkit.print("Removal cancelled.", bullet=False) + raise typer.Exit(0) + + toolkit.print_line() + with ( + toolkit.progress( + title="Removing custom domain", + transient=True, + ) as progress, + client.handle_http_errors( + progress, + default_message=( + "Error removing custom domain. Please try again later." ), - ): - client.remove_custom_domain( - app_id=app_id, - domain_id=selected_domain.id, - ) - - toolkit.success( - CustomDomainRemoveOutput( + not_found_message="Custom domain not found.", + toolkit=toolkit, + ), + ): + client.remove_custom_domain( app_id=app_id, domain_id=selected_domain.id, - name=selected_domain.name, - ), - render_output=_render_custom_domain_remove_output, - ) + ) + + toolkit.success( + CustomDomainRemoveOutput( + app_id=app_id, + domain_id=selected_domain.id, + name=selected_domain.name, + ), + render_output=_render_custom_domain_remove_output, + ) diff --git a/src/fastapi_cloud_cli/commands/domains/restart.py b/src/fastapi_cloud_cli/commands/domains/restart.py index ebe7c4dd..e19bab67 100644 --- a/src/fastapi_cloud_cli/commands/domains/restart.py +++ b/src/fastapi_cloud_cli/commands/domains/restart.py @@ -5,14 +5,14 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient, CustomDomain +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.domains._app import domains_app from fastapi_cloud_cli.commands.domains._shared import ( _find_custom_domain, _select_custom_domain, ) from fastapi_cloud_cli.commands.domains.rendering import render_custom_domain_details 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 @@ -43,7 +43,9 @@ def _render_custom_domain_restart_output( ) +@domains_app.command("restart", cls=UserCommand) def restart_domain( + ctx: typer.Context, domain: Annotated[ str | None, typer.Argument( @@ -62,96 +64,87 @@ def restart_domain( """ Restart failed custom domain setup for 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`.", - ) - - app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) - domain_was_provided = domain is not None - - if domain is None and toolkit.mode == "json": - toolkit.fail( - "missing_required_input", - "Custom domain is required.", - hint="Pass DOMAIN to choose a custom domain.", - ) + toolkit = get_user_command_context(ctx).toolkit + app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) + domain_was_provided = domain is not None + + if domain is None and toolkit.mode == "json": + toolkit.fail( + "missing_required_input", + "Custom domain is required.", + hint="Pass DOMAIN to choose a custom domain.", + ) - with APIClient() as client: - with ( - toolkit.progress( - title="Fetching custom domains", - transient=True, - ) as progress, - client.handle_http_errors( - progress, - default_message=( - "Error fetching custom domains. Please try again later." - ), - not_found_message="App not found.", - toolkit=toolkit, + with APIClient() as client: + with ( + toolkit.progress( + title="Fetching custom domains", + transient=True, + ) as progress, + client.handle_http_errors( + progress, + default_message=( + "Error fetching custom domains. Please try again later." ), - ): - domains = client.get_custom_domains(app_id=app_id).data - - selected_domain: CustomDomain | None - if domain is None: - toolkit.print_title("custom domains") - toolkit.print_line() - failed_domains = [domain for domain in domains if domain.setup_failed] - - if not failed_domains: - toolkit.print("No failed custom domains found.", bullet=False) - return - - selected_domain = _select_custom_domain( - toolkit, - failed_domains, - prompt="Select the custom domain to restart:", - ) - toolkit.print_line() - else: - selected_domain = _find_custom_domain(domains, domain) - if selected_domain is None: - toolkit.fail( - "not_found", - f"Custom domain {domain} not found.", - hint=( - "Run `fastapi cloud domains list` to see available " - "custom domains." - ), - ) - - assert selected_domain is not None - with ( - toolkit.progress( - title="Restarting custom domain setup", - transient=True, - ) as progress, - client.handle_http_errors( - progress, - default_message=( - "Error restarting custom domain setup. Please try again later." + not_found_message="App not found.", + toolkit=toolkit, + ), + ): + domains = client.get_custom_domains(app_id=app_id).data + + selected_domain: CustomDomain | None + if domain is None: + toolkit.print_title("custom domains") + toolkit.print_line() + failed_domains = [domain for domain in domains if domain.setup_failed] + + if not failed_domains: + toolkit.print("No failed custom domains found.", bullet=False) + return + + selected_domain = _select_custom_domain( + toolkit, + failed_domains, + prompt="Select the custom domain to restart:", + ) + toolkit.print_line() + else: + selected_domain = _find_custom_domain(domains, domain) + if selected_domain is None: + toolkit.fail( + "not_found", + f"Custom domain {domain} not found.", + hint=( + "Run `fastapi cloud domains list` to see available " + "custom domains." ), - not_found_message="Custom domain not found.", - toolkit=toolkit, - ), - ): - restarted_domain = client.restart_custom_domain_setup( - app_id=app_id, - domain_id=selected_domain.id, ) - toolkit.success( - CustomDomainRestartOutput( - app_id=app_id, - domain=restarted_domain, - show_title=domain_was_provided, + assert selected_domain is not None + with ( + toolkit.progress( + title="Restarting custom domain setup", + transient=True, + ) as progress, + client.handle_http_errors( + progress, + default_message=( + "Error restarting custom domain setup. Please try again later." + ), + not_found_message="Custom domain not found.", + toolkit=toolkit, ), - render_output=_render_custom_domain_restart_output, - ) + ): + restarted_domain = client.restart_custom_domain_setup( + app_id=app_id, + domain_id=selected_domain.id, + ) + + toolkit.success( + CustomDomainRestartOutput( + app_id=app_id, + domain=restarted_domain, + show_title=domain_was_provided, + ), + render_output=_render_custom_domain_restart_output, + ) diff --git a/src/fastapi_cloud_cli/commands/env/__init__.py b/src/fastapi_cloud_cli/commands/env/__init__.py index 323458db..e6ae8bf6 100644 --- a/src/fastapi_cloud_cli/commands/env/__init__.py +++ b/src/fastapi_cloud_cli/commands/env/__init__.py @@ -1,17 +1,3 @@ -import typer - -from fastapi_cloud_cli.commands.env.delete import delete -from fastapi_cloud_cli.commands.env.get import get_variable -from fastapi_cloud_cli.commands.env.list import list_variables -from fastapi_cloud_cli.commands.env.set import set - -env_app = typer.Typer( - no_args_is_help=True, - help="Manage the environment variables of your app.", -) -env_app.command("list")(list_variables) -env_app.command("get")(get_variable) -env_app.command()(delete) -env_app.command()(set) +from fastapi_cloud_cli.commands.env._app import env_app as env_app __all__ = ["env_app"] diff --git a/src/fastapi_cloud_cli/commands/env/_app.py b/src/fastapi_cloud_cli/commands/env/_app.py new file mode 100644 index 00000000..fd7bcbf0 --- /dev/null +++ b/src/fastapi_cloud_cli/commands/env/_app.py @@ -0,0 +1,6 @@ +import typer + +env_app = typer.Typer( + no_args_is_help=True, + help="Manage the environment variables of your app.", +) diff --git a/src/fastapi_cloud_cli/commands/env/delete.py b/src/fastapi_cloud_cli/commands/env/delete.py index 122bebf1..c395ae72 100644 --- a/src/fastapi_cloud_cli/commands/env/delete.py +++ b/src/fastapi_cloud_cli/commands/env/delete.py @@ -7,10 +7,10 @@ from rich_toolkit.menu import Option from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.env._app import env_app from fastapi_cloud_cli.commands.env._shared import _get_environment_variables 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.env import validate_environment_variable_name from fastapi_cloud_cli.utils.execution import JsonOutputOption @@ -43,7 +43,9 @@ def _delete_environment_variable(client: APIClient, app_id: str, name: str) -> b return True +@env_app.command(cls=UserCommand) def delete( + ctx: typer.Context, name: str | None = typer.Argument( None, help="The name of the environment variable to delete", @@ -88,107 +90,99 @@ def delete( Delete an environment variable from the app. """ - identity = Identity() + toolkit = get_user_command_context(ctx).toolkit - 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.", - ) - - target_app_id = resolve_app_id_or_fail( - toolkit, app_id=app_id, path=path or path_arg - ) - name_provided = name is not None - - with APIClient() as client: - if not name: - if toolkit.mode == "json": - toolkit.fail( - "missing_required_input", - "Environment variable name is required.", - hint="Pass NAME to choose an environment variable.", - ) - - with toolkit.progress( - "Fetching environment variables...", transient=True - ) as progress: - with client.handle_http_errors(progress): - environment_variables = _get_environment_variables( - client=client, app_id=target_app_id - ) - - toolkit.print_title("environment variables") - toolkit.print_line() + target_app_id = resolve_app_id_or_fail( + toolkit, app_id=app_id, path=path or path_arg + ) + name_provided = name is not None - if not environment_variables.data: - toolkit.print("No environment variables found.", bullet=False) - return - - name = toolkit.ask( - "Select the environment variable to delete:", - options=[ - Option({"name": env_var.name, "value": env_var.name}) - for env_var in environment_variables.data - ], - bullet=False, + with APIClient() as client: + if not name: + if toolkit.mode == "json": + toolkit.fail( + "missing_required_input", + "Environment variable name is required.", + hint="Pass NAME to choose an environment variable.", ) - assert name - else: - if not validate_environment_variable_name(name): - toolkit.fail( - "invalid_input", - f"The environment variable name [bold]{name}[/] is invalid.", + with toolkit.progress( + "Fetching environment variables...", transient=True + ) as progress: + with client.handle_http_errors(progress): + environment_variables = _get_environment_variables( + client=client, app_id=target_app_id ) - toolkit.print_line() + toolkit.print_title("environment variables") + toolkit.print_line() - if name_provided and not yes: - if toolkit.mode == "json": - toolkit.fail( - "missing_required_input", - "Deletion confirmation is required.", - hint="Pass --yes to confirm deletion.", - ) + if not environment_variables.data: + toolkit.print("No environment variables found.", bullet=False) + return + + name = toolkit.ask( + "Select the environment variable to delete:", + options=[ + Option({"name": env_var.name, "value": env_var.name}) + for env_var in environment_variables.data + ], + bullet=False, + ) - should_delete = toolkit.confirm( - f"Delete [bold]{name}[/]?", - default=False, - bullet=False, + assert name + else: + if not validate_environment_variable_name(name): + toolkit.fail( + "invalid_input", + f"The environment variable name [bold]{name}[/] is invalid.", ) - if not should_delete: - toolkit.print_title("environment variables") - toolkit.print_line() - toolkit.print("Deletion cancelled.", bullet=False) - raise typer.Exit(0) - toolkit.print_line() - with toolkit.progress( - "Deleting environment variable", transient=True - ) as progress: - with client.handle_http_errors(progress): - deleted = _delete_environment_variable( - client=client, app_id=target_app_id, name=name - ) + toolkit.print_line() - if not deleted: - message = ( - f"Environment variable {name} not found." - if toolkit.mode == "json" - else "Environment variable not found." - ) - toolkit.fail( - "not_found", - message, - hint="Run `fastapi cloud env list` to see available variables.", + if name_provided and not yes: + if toolkit.mode == "json": + toolkit.fail( + "missing_required_input", + "Deletion confirmation is required.", + hint="Pass --yes to confirm deletion.", + ) + + should_delete = toolkit.confirm( + f"Delete [bold]{name}[/]?", + default=False, + bullet=False, ) + if not should_delete: + toolkit.print_title("environment variables") + toolkit.print_line() + toolkit.print("Deletion cancelled.", bullet=False) + raise typer.Exit(0) + toolkit.print_line() + + with toolkit.progress( + "Deleting environment variable", transient=True + ) as progress: + with client.handle_http_errors(progress): + deleted = _delete_environment_variable( + client=client, app_id=target_app_id, name=name + ) - toolkit.success( - EnvironmentVariableDeleteOutput( - app_id=target_app_id, name=name, show_tag=name_provided - ), - render_output=_render_environment_variable_delete_output, + if not deleted: + message = ( + f"Environment variable {name} not found." + if toolkit.mode == "json" + else "Environment variable not found." ) + toolkit.fail( + "not_found", + message, + hint="Run `fastapi cloud env list` to see available variables.", + ) + + toolkit.success( + EnvironmentVariableDeleteOutput( + app_id=target_app_id, name=name, show_tag=name_provided + ), + render_output=_render_environment_variable_delete_output, + ) diff --git a/src/fastapi_cloud_cli/commands/env/get.py b/src/fastapi_cloud_cli/commands/env/get.py index 5f8682be..aaa8adcf 100644 --- a/src/fastapi_cloud_cli/commands/env/get.py +++ b/src/fastapi_cloud_cli/commands/env/get.py @@ -8,6 +8,8 @@ from rich_toolkit.menu import Option from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.env._app import env_app from fastapi_cloud_cli.commands.env._shared import ( EnvironmentVariable, _find_environment_variable, @@ -15,8 +17,6 @@ _get_environment_variables, ) 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 @@ -43,7 +43,9 @@ def _render_environment_variable_get_output( toolkit.print(table, bullet=False) +@env_app.command("get", cls=UserCommand) def get_variable( + ctx: typer.Context, name: Annotated[ str | None, typer.Argument( @@ -73,67 +75,59 @@ def get_variable( Get an environment variable for the 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.", - ) - - target_app_id = resolve_app_id_or_fail(toolkit, app_id=app_id, path=path) - name_provided = name is not None - - if name is None and toolkit.mode == "json": - toolkit.fail( - "missing_required_input", - "Environment variable name is required.", - hint="Pass NAME to choose an environment variable.", - ) - - with APIClient() as client: - with toolkit.progress( - "Fetching environment variables...", transient=True - ) as progress: - with client.handle_http_errors(progress): - environment_variables = _get_environment_variables( - client=client, app_id=target_app_id - ) - - if name is None: - toolkit.print_title("environment variables") - toolkit.print_line() - - if not environment_variables.data: - toolkit.print("No environment variables found.", bullet=False) - return - - name = toolkit.ask( - "Select the environment variable to get:", - options=[ - Option({"name": env_var.name, "value": env_var.name}) - for env_var in environment_variables.data - ], - bullet=False, - ) - - variable = _find_environment_variable(environment_variables.data, name) - - if variable is None: - toolkit.fail( - "not_found", - f"Environment variable {name} not found.", - hint="Run `fastapi cloud env list` to see available variables.", - ) - assert variable is not None - - toolkit.success( - EnvironmentVariableGetOutput( - app_id=target_app_id, - variable=variable, - show_tag=name_provided, - ), - render_output=_render_environment_variable_get_output, + toolkit = get_user_command_context(ctx).toolkit + + target_app_id = resolve_app_id_or_fail(toolkit, app_id=app_id, path=path) + name_provided = name is not None + + if name is None and toolkit.mode == "json": + toolkit.fail( + "missing_required_input", + "Environment variable name is required.", + hint="Pass NAME to choose an environment variable.", + ) + + with APIClient() as client: + with toolkit.progress( + "Fetching environment variables...", transient=True + ) as progress: + with client.handle_http_errors(progress): + environment_variables = _get_environment_variables( + client=client, app_id=target_app_id + ) + + if name is None: + toolkit.print_title("environment variables") + toolkit.print_line() + + if not environment_variables.data: + toolkit.print("No environment variables found.", bullet=False) + return + + name = toolkit.ask( + "Select the environment variable to get:", + options=[ + Option({"name": env_var.name, "value": env_var.name}) + for env_var in environment_variables.data + ], + bullet=False, + ) + + variable = _find_environment_variable(environment_variables.data, name) + + if variable is None: + toolkit.fail( + "not_found", + f"Environment variable {name} not found.", + hint="Run `fastapi cloud env list` to see available variables.", ) + assert variable is not None + + toolkit.success( + EnvironmentVariableGetOutput( + app_id=target_app_id, + variable=variable, + show_tag=name_provided, + ), + render_output=_render_environment_variable_get_output, + ) diff --git a/src/fastapi_cloud_cli/commands/env/list.py b/src/fastapi_cloud_cli/commands/env/list.py index e63c9476..5779b33e 100644 --- a/src/fastapi_cloud_cli/commands/env/list.py +++ b/src/fastapi_cloud_cli/commands/env/list.py @@ -8,6 +8,8 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.env._app import env_app from fastapi_cloud_cli.commands.env._shared import ( ENV_VAR_VALUE_MAX_LENGTH, EnvironmentVariable, @@ -15,8 +17,6 @@ _get_environment_variables, ) 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.dates import format_last_updated from fastapi_cloud_cli.utils.execution import JsonOutputOption @@ -59,7 +59,9 @@ def _render_environment_variables_list_output( toolkit.print(_get_environment_variables_table(data.variables), bullet=False) +@env_app.command("list", cls=UserCommand) def list_variables( + ctx: typer.Context, path: Annotated[ Path | None, typer.Option( @@ -83,31 +85,23 @@ def list_variables( List the environment variables for the 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.", - ) - - target_app_id = resolve_app_id_or_fail(toolkit, app_id=app_id, path=path) - - with APIClient() as client: - with toolkit.progress( - "Fetching environment variables...", transient=True - ) as progress: - with client.handle_http_errors(progress): - environment_variables = _get_environment_variables( - client=client, app_id=target_app_id - ) - - toolkit.success( - EnvironmentVariablesListOutput( - app_id=target_app_id, - variables=environment_variables.data, - ), - render_output=_render_environment_variables_list_output, - ) + toolkit = get_user_command_context(ctx).toolkit + + target_app_id = resolve_app_id_or_fail(toolkit, app_id=app_id, path=path) + + with APIClient() as client: + with toolkit.progress( + "Fetching environment variables...", transient=True + ) as progress: + with client.handle_http_errors(progress): + environment_variables = _get_environment_variables( + client=client, app_id=target_app_id + ) + + toolkit.success( + EnvironmentVariablesListOutput( + app_id=target_app_id, + variables=environment_variables.data, + ), + render_output=_render_environment_variables_list_output, + ) diff --git a/src/fastapi_cloud_cli/commands/env/set.py b/src/fastapi_cloud_cli/commands/env/set.py index 90fa1a2c..3271b3ae 100644 --- a/src/fastapi_cloud_cli/commands/env/set.py +++ b/src/fastapi_cloud_cli/commands/env/set.py @@ -7,9 +7,10 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.env._app import env_app 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 FastAPIRichToolkit, get_rich_toolkit +from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit from fastapi_cloud_cli.utils.execution import JsonOutputOption @@ -104,7 +105,9 @@ def _set_environment_variable( response.raise_for_status() +@env_app.command(cls=UserCommand) def set( + ctx: typer.Context, name: str | None = typer.Argument( None, help="The name of the environment variable to set", @@ -159,57 +162,49 @@ def set( Set an environment variable for the app. """ - identity = Identity() + toolkit = get_user_command_context(ctx).toolkit - 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.", - ) + target_app_id = resolve_app_id_or_fail( + toolkit, app_id=app_id, path=path or path_arg + ) + name_needs_prompt = name is None + value_needs_prompt = value is None and not value_stdin + prompts_user = name_needs_prompt or value_needs_prompt + if prompts_user and toolkit.mode != "json": + toolkit.print_title("environment variables") + toolkit.print_line() - target_app_id = resolve_app_id_or_fail( - toolkit, app_id=app_id, path=path or path_arg - ) - name_needs_prompt = name is None - value_needs_prompt = value is None and not value_stdin - prompts_user = name_needs_prompt or value_needs_prompt - if prompts_user and toolkit.mode != "json": - toolkit.print_title("environment variables") - toolkit.print_line() - - name = _resolve_environment_variable_name( - toolkit, - name=name, - secret=secret, - ) - value = _resolve_environment_variable_value( - toolkit, - value=value, - value_stdin=value_stdin, - secret=secret, - ) + name = _resolve_environment_variable_name( + toolkit, + name=name, + secret=secret, + ) + value = _resolve_environment_variable_value( + toolkit, + value=value, + value_stdin=value_stdin, + secret=secret, + ) - with APIClient() as client: - with toolkit.progress( - "Setting environment variable", transient=True - ) as progress: - with client.handle_http_errors(progress): - _set_environment_variable( - client=client, - app_id=target_app_id, - name=name, - value=value, - is_secret=secret, - ) - - toolkit.success( - EnvironmentVariableSetOutput( - app_id=target_app_id, - name=name, - is_secret=secret, - show_tag=not prompts_user, - ), - render_output=_render_environment_variable_set_output, - ) + with APIClient() as client: + with toolkit.progress( + "Setting environment variable", transient=True + ) as progress: + with client.handle_http_errors(progress): + _set_environment_variable( + client=client, + app_id=target_app_id, + name=name, + value=value, + is_secret=secret, + ) + + toolkit.success( + EnvironmentVariableSetOutput( + app_id=target_app_id, + name=name, + is_secret=secret, + show_tag=not prompts_user, + ), + render_output=_render_environment_variable_set_output, + ) diff --git a/src/fastapi_cloud_cli/commands/integrations/providers/__init__.py b/src/fastapi_cloud_cli/commands/integrations/providers/__init__.py index 9758f6d4..0b2430cc 100644 --- a/src/fastapi_cloud_cli/commands/integrations/providers/__init__.py +++ b/src/fastapi_cloud_cli/commands/integrations/providers/__init__.py @@ -1,11 +1,5 @@ -import typer - -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.", +from fastapi_cloud_cli.commands.integrations.providers._app import ( + providers_app as providers_app, ) -providers_app.command("list")(list_providers) __all__ = ["providers_app"] diff --git a/src/fastapi_cloud_cli/commands/integrations/providers/_app.py b/src/fastapi_cloud_cli/commands/integrations/providers/_app.py new file mode 100644 index 00000000..d6749c78 --- /dev/null +++ b/src/fastapi_cloud_cli/commands/integrations/providers/_app.py @@ -0,0 +1,6 @@ +import typer + +providers_app = typer.Typer( + no_args_is_help=True, + help="Manage integration providers for a team.", +) diff --git a/src/fastapi_cloud_cli/commands/integrations/providers/list.py b/src/fastapi_cloud_cli/commands/integrations/providers/list.py index 2014d0db..eb5f3f41 100644 --- a/src/fastapi_cloud_cli/commands/integrations/providers/list.py +++ b/src/fastapi_cloud_cli/commands/integrations/providers/list.py @@ -7,8 +7,8 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient -from fastapi_cloud_cli.utils.auth import Identity -from fastapi_cloud_cli.utils.cli import get_rich_toolkit +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.integrations.providers._app import providers_app from fastapi_cloud_cli.utils.execution import JsonOutputOption from fastapi_cloud_cli.utils.teams import resolve_team_id @@ -88,7 +88,9 @@ def _render_providers_list_output( toolkit.print(_get_providers_table(data.providers), bullet=False) +@providers_app.command("list", cls=UserCommand) def list_providers( + ctx: typer.Context, team_id: Annotated[ str | None, typer.Option( @@ -104,41 +106,34 @@ def list_providers( """ List integration providers and their connection status for a team. """ - 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: - team_id = resolve_team_id( - toolkit, - client, - team_id=team_id, - empty_hint="Create a team before listing integration providers.", - ) - - with ( - toolkit.progress( - title="Fetching integration providers", - transient=True, - ) as progress, - client.handle_http_errors( - progress, - default_message=( - "Error fetching integration providers. Please try again later." - ), - not_found_message="Team not found.", - toolkit=toolkit, - ), - ): - providers = _get_providers(client, team_id=team_id) - toolkit.success( - ProvidersListOutput(team_id=team_id, providers=providers), - render_output=_render_providers_list_output, + toolkit = get_user_command_context(ctx).toolkit + + with APIClient() as client: + team_id = resolve_team_id( + toolkit, + client, + team_id=team_id, + empty_hint="Create a team before listing integration providers.", ) + + with ( + toolkit.progress( + title="Fetching integration providers", + transient=True, + ) as progress, + client.handle_http_errors( + progress, + default_message=( + "Error fetching integration providers. Please try again later." + ), + not_found_message="Team not found.", + toolkit=toolkit, + ), + ): + providers = _get_providers(client, team_id=team_id) + + toolkit.success( + ProvidersListOutput(team_id=team_id, providers=providers), + render_output=_render_providers_list_output, + ) diff --git a/src/fastapi_cloud_cli/commands/integrations/resources/__init__.py b/src/fastapi_cloud_cli/commands/integrations/resources/__init__.py index df98ea1a..fd47cdbb 100644 --- a/src/fastapi_cloud_cli/commands/integrations/resources/__init__.py +++ b/src/fastapi_cloud_cli/commands/integrations/resources/__init__.py @@ -1,19 +1,5 @@ -import typer - -from fastapi_cloud_cli.commands.integrations.resources.connect import connect_resource -from fastapi_cloud_cli.commands.integrations.resources.disconnect import ( - disconnect_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.", +from fastapi_cloud_cli.commands.integrations.resources._app import ( + resources_app as resources_app, ) -resources_app.command("connect")(connect_resource) -resources_app.command("disconnect")(disconnect_resource) -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/_app.py b/src/fastapi_cloud_cli/commands/integrations/resources/_app.py new file mode 100644 index 00000000..9a879aef --- /dev/null +++ b/src/fastapi_cloud_cli/commands/integrations/resources/_app.py @@ -0,0 +1,6 @@ +import typer + +resources_app = typer.Typer( + no_args_is_help=True, + help="Manage resources connected to an app.", +) diff --git a/src/fastapi_cloud_cli/commands/integrations/resources/connect.py b/src/fastapi_cloud_cli/commands/integrations/resources/connect.py index fd9f37a8..3779c1b8 100644 --- a/src/fastapi_cloud_cli/commands/integrations/resources/connect.py +++ b/src/fastapi_cloud_cli/commands/integrations/resources/connect.py @@ -7,16 +7,16 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context from fastapi_cloud_cli.commands.apps.list import ( App, _get_app, _get_app_dashboard_url, _get_team, ) +from fastapi_cloud_cli.commands.integrations.resources._app import resources_app from fastapi_cloud_cli.config import Settings 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__) @@ -75,7 +75,9 @@ def _render_resource_connect_output( ) +@resources_app.command("connect", cls=UserCommand) def connect_resource( + ctx: typer.Context, app_id: Annotated[ str | None, typer.Option( @@ -95,61 +97,54 @@ def connect_resource( """ 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, + + toolkit = get_user_command_context(ctx).toolkit + + 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, ), - render_output=_render_resource_connect_output, - ) + ): + 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, + ) diff --git a/src/fastapi_cloud_cli/commands/integrations/resources/disconnect.py b/src/fastapi_cloud_cli/commands/integrations/resources/disconnect.py index 8c3e5567..03313cee 100644 --- a/src/fastapi_cloud_cli/commands/integrations/resources/disconnect.py +++ b/src/fastapi_cloud_cli/commands/integrations/resources/disconnect.py @@ -6,12 +6,12 @@ from rich_toolkit.menu import Option from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.integrations.resources._app import resources_app from fastapi_cloud_cli.commands.integrations.resources.get import _get_resource from fastapi_cloud_cli.commands.integrations.resources.list import _get_resources from fastapi_cloud_cli.commands.integrations.resources.providers import PROVIDER_NAMES 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 @@ -75,7 +75,9 @@ def _render_resource_disconnect_output( ) +@resources_app.command("disconnect", cls=UserCommand) def disconnect_resource( + ctx: typer.Context, resource_id: Annotated[ str | None, typer.Argument( @@ -105,146 +107,138 @@ def disconnect_resource( The provider resource is not deleted, but its managed environment variables are removed from the 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.", - ) + toolkit = get_user_command_context(ctx).toolkit - app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) - resource_id_was_provided = resource_id is not None - - if json_output: - if resource_id is None: - toolkit.fail( - "missing_required_input", - "Resource ID is required.", - hint="Pass RESOURCE_ID to choose a connected resource.", - ) - - if not yes: - toolkit.fail( - "missing_required_input", - "Disconnection confirmation is required.", - hint="Pass --yes to confirm disconnection.", - ) - - with APIClient() as client: - if resource_id is None: - with ( - toolkit.progress( - title="Fetching connected resources", - transient=True, - ) as progress, - client.handle_http_errors( - progress, - default_message=( - "Error fetching connected resources. Please try again later." - ), - not_found_message="App not found.", - toolkit=toolkit, - ), - ): - resources = _get_resources(client, app_id=app_id) + app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) + resource_id_was_provided = resource_id is not None - toolkit.print_title("disconnect resource") - toolkit.print_line() + if json_output: + if resource_id is None: + toolkit.fail( + "missing_required_input", + "Resource ID is required.", + hint="Pass RESOURCE_ID to choose a connected resource.", + ) - if not resources: - toolkit.print("No connected resources found.", bullet=False) - return - - resource_id = toolkit.ask( - "Select the resource to disconnect:", - options=[ - Option( - { - "name": ( - f"{resource.name} " - f"({PROVIDER_NAMES[resource.provider]})" - ), - "value": resource.id, - } - ) - for resource in resources - ], - bullet=False, - ) - toolkit.print_line() + if not yes: + toolkit.fail( + "missing_required_input", + "Disconnection confirmation is required.", + hint="Pass --yes to confirm disconnection.", + ) + with APIClient() as client: + if resource_id is None: with ( toolkit.progress( - title="Fetching connected resource", + title="Fetching connected resources", transient=True, ) as progress, client.handle_http_errors( progress, default_message=( - "Error fetching connected resource. Please try again later." + "Error fetching connected resources. Please try again later." ), - not_found_message="Connected resource not found.", + not_found_message="App not found.", toolkit=toolkit, ), ): - resource = _get_resource( - client, - app_id=app_id, - resource_id=resource_id, - ) - - if resource_id_was_provided: - toolkit.print_title("disconnect resource") - toolkit.print_line() + resources = _get_resources(client, app_id=app_id) + + toolkit.print_title("disconnect resource") + toolkit.print_line() - _print_disconnect_warning( - toolkit, - provider_name=PROVIDER_NAMES[resource.provider], - environment_variables=resource.environment_variables, + if not resources: + toolkit.print("No connected resources found.", bullet=False) + return + + resource_id = toolkit.ask( + "Select the resource to disconnect:", + options=[ + Option( + { + "name": ( + f"{resource.name} ({PROVIDER_NAMES[resource.provider]})" + ), + "value": resource.id, + } + ) + for resource in resources + ], + bullet=False, ) + toolkit.print_line() - if not yes: - toolkit.print_line() - should_disconnect = toolkit.confirm( - f"Disconnect [bold]{resource.name}[/bold]?", - default=False, - bullet=False, - ) - if not should_disconnect: - toolkit.print_line() - toolkit.print("Disconnection cancelled.", bullet=False) - raise typer.Exit(0) + 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, + ) + if resource_id_was_provided: + toolkit.print_title("disconnect resource") toolkit.print_line() - with ( - toolkit.progress( - title="Disconnecting connected resource", - transient=True, - ) as progress, - client.handle_http_errors( - progress, - default_message=( - "Error disconnecting connected resource. Please try again later." - ), - not_found_message="Connected resource not found.", - toolkit=toolkit, + + _print_disconnect_warning( + toolkit, + provider_name=PROVIDER_NAMES[resource.provider], + environment_variables=resource.environment_variables, + ) + + if not yes: + toolkit.print_line() + should_disconnect = toolkit.confirm( + f"Disconnect [bold]{resource.name}[/bold]?", + default=False, + bullet=False, + ) + if not should_disconnect: + toolkit.print_line() + toolkit.print("Disconnection cancelled.", bullet=False) + raise typer.Exit(0) + + toolkit.print_line() + with ( + toolkit.progress( + title="Disconnecting connected resource", + transient=True, + ) as progress, + client.handle_http_errors( + progress, + default_message=( + "Error disconnecting connected resource. Please try again later." ), - ): - _disconnect_resource( - client, - app_id=app_id, - resource_id=resource.id, - ) - - toolkit.success( - ResourceDisconnectOutput( + not_found_message="Connected resource not found.", + toolkit=toolkit, + ), + ): + _disconnect_resource( + client, app_id=app_id, resource_id=resource.id, - resource_name=resource.name, - environment_variables=resource.environment_variables, - ), - render_output=_render_resource_disconnect_output, - ) + ) + + toolkit.success( + ResourceDisconnectOutput( + app_id=app_id, + resource_id=resource.id, + resource_name=resource.name, + environment_variables=resource.environment_variables, + ), + render_output=_render_resource_disconnect_output, + ) diff --git a/src/fastapi_cloud_cli/commands/integrations/resources/get.py b/src/fastapi_cloud_cli/commands/integrations/resources/get.py index 431de30a..d43d27d9 100644 --- a/src/fastapi_cloud_cli/commands/integrations/resources/get.py +++ b/src/fastapi_cloud_cli/commands/integrations/resources/get.py @@ -7,13 +7,14 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.integrations.resources._app import resources_app from fastapi_cloud_cli.commands.integrations.resources.providers import ( PROVIDER_NAMES, Provider, ) 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.cli import get_details_table from fastapi_cloud_cli.utils.dates import format_last_updated from fastapi_cloud_cli.utils.execution import JsonOutputOption @@ -174,7 +175,9 @@ def _render_resource_get_output( toolkit.print(get_details_table(rows)) +@resources_app.command("get", cls=UserCommand) def get_resource( + ctx: typer.Context, resource_id: Annotated[ str, typer.Argument(help="ID of the connected resource to return."), @@ -191,40 +194,33 @@ def get_resource( """ 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, + toolkit = get_user_command_context(ctx).toolkit + + 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." ), - ): - 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, - ) + 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 9c2f7ccc..ac06d0b6 100644 --- a/src/fastapi_cloud_cli/commands/integrations/resources/list.py +++ b/src/fastapi_cloud_cli/commands/integrations/resources/list.py @@ -7,13 +7,13 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.integrations.resources._app import resources_app from fastapi_cloud_cli.commands.integrations.resources.providers import ( PROVIDER_NAMES, Provider, ) 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 @@ -75,7 +75,9 @@ def _render_resources_list_output( toolkit.print(_get_resources_table(data.resources), bullet=False) +@resources_app.command("list", cls=UserCommand) def list_resources( + ctx: typer.Context, app_id: Annotated[ str | None, typer.Option( @@ -88,36 +90,29 @@ def list_resources( """ List resources 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 resources", - transient=True, - ) as progress, - client.handle_http_errors( - progress, - default_message=( - "Error fetching connected resources. Please try again later." - ), - not_found_message="App not found.", - toolkit=toolkit, - ), - ): - resources = _get_resources(client, app_id=app_id) - toolkit.success( - ResourcesListOutput(app_id=app_id, resources=resources), - render_output=_render_resources_list_output, - ) + toolkit = get_user_command_context(ctx).toolkit + + app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) + + with APIClient() as client: + with ( + toolkit.progress( + title="Fetching connected resources", + transient=True, + ) as progress, + client.handle_http_errors( + progress, + default_message=( + "Error fetching connected resources. Please try again later." + ), + not_found_message="App not found.", + toolkit=toolkit, + ), + ): + resources = _get_resources(client, app_id=app_id) + + toolkit.success( + ResourcesListOutput(app_id=app_id, resources=resources), + render_output=_render_resources_list_output, + ) diff --git a/src/fastapi_cloud_cli/commands/login.py b/src/fastapi_cloud_cli/commands/login.py index 3cbdf96a..9493ef8e 100644 --- a/src/fastapi_cloud_cli/commands/login.py +++ b/src/fastapi_cloud_cli/commands/login.py @@ -3,6 +3,7 @@ import typer +from fastapi_cloud_cli._app import app, cloud_app from fastapi_cloud_cli.api import APIClient from fastapi_cloud_cli.commands._flow import ( DEFAULT_LOGIN_TIMEOUT_SECONDS, @@ -11,6 +12,7 @@ render_login_output, start_device_authorization, ) +from fastapi_cloud_cli.commands.auth._app import auth_app from fastapi_cloud_cli.utils.auth import Identity from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit, get_rich_toolkit from fastapi_cloud_cli.utils.execution import JsonOutputOption @@ -59,6 +61,9 @@ def _interactive_login( ) +@app.command() +@cloud_app.command() +@auth_app.command() def login( no_open: Annotated[ bool, diff --git a/src/fastapi_cloud_cli/commands/logout.py b/src/fastapi_cloud_cli/commands/logout.py index 977f0ccb..e6811d08 100644 --- a/src/fastapi_cloud_cli/commands/logout.py +++ b/src/fastapi_cloud_cli/commands/logout.py @@ -3,6 +3,7 @@ from pydantic import BaseModel from rich_toolkit import RichToolkit +from fastapi_cloud_cli._app import cloud_app from fastapi_cloud_cli.utils.auth import delete_auth_config from fastapi_cloud_cli.utils.cli import get_rich_toolkit from fastapi_cloud_cli.utils.execution import JsonOutputOption @@ -18,6 +19,7 @@ def _render_logout_output(data: LogoutOutput, toolkit: RichToolkit) -> None: toolkit.print("You are now logged out!", emoji="👋") +@cloud_app.command() def logout(json_output: JsonOutputOption = False) -> Any: """ Logout from FastAPI Cloud. diff --git a/src/fastapi_cloud_cli/commands/logs.py b/src/fastapi_cloud_cli/commands/logs.py index 84f925ad..336ed625 100644 --- a/src/fastapi_cloud_cli/commands/logs.py +++ b/src/fastapi_cloud_cli/commands/logs.py @@ -11,6 +11,7 @@ from rich.markup import escape from rich_toolkit import RichToolkit +from fastapi_cloud_cli._app import cloud_app from fastapi_cloud_cli.api import ( APIClient, AppLogEntry, @@ -20,9 +21,10 @@ get_http_error_hint, handle_http_error, ) +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.apps._app import apps_app 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 FastAPIRichToolkit, get_rich_toolkit +from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit from fastapi_cloud_cli.utils.errors import ErrorCode from fastapi_cloud_cli.utils.execution import JsonOutputOption @@ -204,7 +206,10 @@ def _process_log_stream( ) +@cloud_app.command(cls=UserCommand) +@apps_app.command("logs", cls=UserCommand) def logs( + ctx: typer.Context, path: Annotated[ Path | None, typer.Argument( @@ -255,40 +260,33 @@ def logs( fastapi cloud logs --no-follow # Fetch recent logs and exit fastapi cloud logs --tail 50 --since 1h # Last 50 logs from the past hour """ - 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.", - ) - - target_app_id = resolve_app_id_or_fail( - toolkit, - app_id=app_id, - path=path, - hint="Pass --app-id or run `fastapi cloud link` to link an app.", - ) + toolkit = get_user_command_context(ctx).toolkit - logger.debug("Fetching logs for app ID: %s", target_app_id) + target_app_id = resolve_app_id_or_fail( + toolkit, + app_id=app_id, + path=path, + hint="Pass --app-id or run `fastapi cloud link` to link an app.", + ) - if follow: - toolkit.print( - f"Streaming logs for [bold]{target_app_id}[/bold] (Ctrl+C to exit)...", - emoji="📡", - ) - else: - toolkit.print( - f"Fetching logs for [bold]{target_app_id}[/bold]...", - emoji="📜", - ) - toolkit.print_line() + logger.debug("Fetching logs for app ID: %s", target_app_id) - _process_log_stream( - toolkit=toolkit, - app_id=target_app_id, - tail=tail, - since=since, - follow=follow, + if follow: + toolkit.print( + f"Streaming logs for [bold]{target_app_id}[/bold] (Ctrl+C to exit)...", + emoji="📡", ) + else: + toolkit.print( + f"Fetching logs for [bold]{target_app_id}[/bold]...", + emoji="📜", + ) + toolkit.print_line() + + _process_log_stream( + toolkit=toolkit, + app_id=target_app_id, + tail=tail, + since=since, + follow=follow, + ) diff --git a/src/fastapi_cloud_cli/commands/setup_ci.py b/src/fastapi_cloud_cli/commands/setup_ci.py index b5d4268b..caf9eac0 100644 --- a/src/fastapi_cloud_cli/commands/setup_ci.py +++ b/src/fastapi_cloud_cli/commands/setup_ci.py @@ -9,10 +9,12 @@ import typer from pydantic import BaseModel +from fastapi_cloud_cli._app import cloud_app from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.ci._app import ci_app 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 FastAPIRichToolkit, get_rich_toolkit +from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit from fastapi_cloud_cli.utils.execution import JsonOutputOption logger = logging.getLogger(__name__) @@ -223,7 +225,10 @@ def _resolve_existing_workflow_path( return None +@cloud_app.command(cls=UserCommand) +@ci_app.command("setup", cls=UserCommand) def setup_ci( + ctx: typer.Context, path: Annotated[ Path | None, typer.Argument( @@ -283,254 +288,238 @@ def setup_ci( fastapi cloud setup-ci --file ci.yml # Writes workflow to .github/workflows/ci.yml """ - identity = Identity() + toolkit = get_user_command_context(ctx).toolkit - 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.", - ) + if secrets_only and workflow_only: + toolkit.fail( + "invalid_input", + "--secrets-only and --workflow-only cannot be used together.", + ) - if secrets_only and workflow_only: - toolkit.fail( - "invalid_input", - "--secrets-only and --workflow-only cannot be used together.", - ) + target_app_id = resolve_app_id_or_fail( + toolkit, + app_id=app_id, + path=path, + hint="Pass --app-id or run `fastapi deploy` first.", + ) - target_app_id = resolve_app_id_or_fail( - toolkit, - app_id=app_id, - path=path, - hint="Pass --app-id or run `fastapi deploy` first.", + if not _check_git_installed(): + toolkit.fail( + "not_found", + "git is not installed. Please install git to use this command.", ) - if not _check_git_installed(): - toolkit.fail( - "not_found", - "git is not installed. Please install git to use this command.", - ) - - try: - origin = _get_remote_origin() - except subprocess.CalledProcessError: - toolkit.fail( - "not_found", - "Could not retrieve the git remote origin URL. Make sure you're in a git repository with a remote origin set.", - ) + try: + origin = _get_remote_origin() + except subprocess.CalledProcessError: + toolkit.fail( + "not_found", + "Could not retrieve the git remote origin URL. Make sure you're in a git repository with a remote origin set.", + ) - # Check if it's a GitHub host (github.com or GitHub Enterprise) - if "github" not in origin.lower(): - toolkit.fail( - "invalid_input", - "Remote origin is not a GitHub repository. Please set up a GitHub repo and add it as the remote origin.", - ) + # Check if it's a GitHub host (github.com or GitHub Enterprise) + if "github" not in origin.lower(): + toolkit.fail( + "invalid_input", + "Remote origin is not a GitHub repository. Please set up a GitHub repo and add it as the remote origin.", + ) - repo_slug = _repo_slug_from_origin(origin) or origin - - if not branch: - branch = _get_default_branch() - - workflow_path = _get_workflow_path(file) - needs_secrets = not workflow_only - needs_workflow = not secrets_only - has_gh = _check_gh_cli_installed() if needs_secrets and not dry_run else True - - if ( - toolkit.mode == "json" - and needs_workflow - and not dry_run - and not file - and workflow_path.exists() - ): - toolkit.fail( - "invalid_input", - f"Workflow file {_format_workflow_path(workflow_path)} already exists.", - hint="Pass --file to choose another workflow file or remove the existing file.", - ) + repo_slug = _repo_slug_from_origin(origin) or origin - if needs_secrets and not dry_run and toolkit.mode == "json" and not has_gh: - toolkit.fail( - "dependency_missing", - "GitHub CLI (`gh`) is required to set GitHub Actions secrets.", - hint="Install gh or use --workflow-only to write only the workflow file.", - ) + if not branch: + branch = _get_default_branch() - if dry_run: - toolkit.print( - "[yellow]This is a dry run — no changes will be made[/yellow]" - ) - toolkit.print_line() + workflow_path = _get_workflow_path(file) + needs_secrets = not workflow_only + needs_workflow = not secrets_only + has_gh = _check_gh_cli_installed() if needs_secrets and not dry_run else True - toolkit.print_title("Configuring CI") - toolkit.print_line() + if ( + toolkit.mode == "json" + and needs_workflow + and not dry_run + and not file + and workflow_path.exists() + ): + toolkit.fail( + "invalid_input", + f"Workflow file {_format_workflow_path(workflow_path)} already exists.", + hint="Pass --file to choose another workflow file or remove the existing file.", + ) - toolkit.print( - f"Setting up CI for [bold]{repo_slug}[/bold] (branch: {branch})", - emoji="⚙️", + if needs_secrets and not dry_run and toolkit.mode == "json" and not has_gh: + toolkit.fail( + "dependency_missing", + "GitHub CLI (`gh`) is required to set GitHub Actions secrets.", + hint="Install gh or use --workflow-only to write only the workflow file.", ) + + if dry_run: + toolkit.print("[yellow]This is a dry run — no changes will be made[/yellow]") toolkit.print_line() - msg_token = "Created deploy token" - msg_secrets = ( - "Set GitHub Actions secrets [bold blue]FASTAPI_CLOUD_TOKEN[/] " - "and [bold blue]FASTAPI_CLOUD_APP_ID[/]" - ) - msg_workflow = f"Wrote [bold]{workflow_path}[/bold] (branch: {branch})" - - if dry_run: - if needs_secrets: - toolkit.print(msg_token) - toolkit.print(msg_secrets) - - if needs_workflow: - toolkit.print(msg_workflow) - - toolkit.success( - CISetupOutput( - app_id=target_app_id, - repo=repo_slug, - branch=branch, - workflow_path=_format_workflow_path(workflow_path), - created_token=False, - set_github_secrets=False, - wrote_workflow=False, - ), - render_output=lambda _data, _toolkit: None, - ) - return + toolkit.print_title("Configuring CI") + toolkit.print_line() + + toolkit.print( + f"Setting up CI for [bold]{repo_slug}[/bold] (branch: {branch})", + emoji="⚙️", + ) + toolkit.print_line() - token_expired_at: str | None = None - created_token = False - set_github_secrets = False - wrote_workflow = False + msg_token = "Created deploy token" + msg_secrets = ( + "Set GitHub Actions secrets [bold blue]FASTAPI_CLOUD_TOKEN[/] " + "and [bold blue]FASTAPI_CLOUD_APP_ID[/]" + ) + msg_workflow = f"Wrote [bold]{workflow_path}[/bold] (branch: {branch})" + if dry_run: if needs_secrets: - should_create_token = ( - True - if toolkit.mode == "json" - else toolkit.confirm( - "Create a FastAPI Cloud deploy token for GitHub Actions?", - default=True, - ) + toolkit.print(msg_token) + toolkit.print(msg_secrets) + + if needs_workflow: + toolkit.print(msg_workflow) + + toolkit.success( + CISetupOutput( + app_id=target_app_id, + repo=repo_slug, + branch=branch, + workflow_path=_format_workflow_path(workflow_path), + created_token=False, + set_github_secrets=False, + wrote_workflow=False, + ), + render_output=lambda _data, _toolkit: None, + ) + return + + token_expired_at: str | None = None + created_token = False + set_github_secrets = False + wrote_workflow = False + + if needs_secrets: + should_create_token = ( + True + if toolkit.mode == "json" + else toolkit.confirm( + "Create a FastAPI Cloud deploy token for GitHub Actions?", + default=True, ) - if toolkit.mode != "json": - toolkit.print_line() + ) + if toolkit.mode != "json": + toolkit.print_line() - if should_create_token: - # Create unique token name with timestamp to avoid duplicates - timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") - token_name = f"GitHub Actions — {repo_slug} ({timestamp})" - - with ( - APIClient() as client, - toolkit.progress( - title="Generating deploy token...", done_emoji="🔑" - ) as progress, - client.handle_http_errors( - progress, default_message="Error creating deploy token." - ), - ): - token_data = _create_token( - client=client, app_id=target_app_id, token_name=token_name - ) - token_expired_at = token_data["expired_at"] - created_token = True - progress.log(msg_token) + if should_create_token: + # Create unique token name with timestamp to avoid duplicates + timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") + token_name = f"GitHub Actions — {repo_slug} ({timestamp})" + + with ( + APIClient() as client, + toolkit.progress( + title="Generating deploy token...", done_emoji="🔑" + ) as progress, + client.handle_http_errors( + progress, default_message="Error creating deploy token." + ), + ): + token_data = _create_token( + client=client, app_id=target_app_id, token_name=token_name + ) + token_expired_at = token_data["expired_at"] + created_token = True + progress.log(msg_token) - toolkit.print_line() + toolkit.print_line() - if has_gh: - should_set_secrets = ( - True - if toolkit.mode == "json" - else toolkit.confirm( - "Set GitHub Actions secrets " - "[bold blue]FASTAPI_CLOUD_TOKEN[/] and " - "[bold blue]FASTAPI_CLOUD_APP_ID[/] via gh?", - default=True, - ) - ) - if toolkit.mode != "json": - toolkit.print_line() - else: - should_set_secrets = False - secrets_url = ( - f"https://{_get_github_host(origin)}/{repo_slug}" - "/settings/secrets/actions" + if has_gh: + should_set_secrets = ( + True + if toolkit.mode == "json" + else toolkit.confirm( + "Set GitHub Actions secrets " + "[bold blue]FASTAPI_CLOUD_TOKEN[/] and " + "[bold blue]FASTAPI_CLOUD_APP_ID[/] via gh?", + default=True, ) - toolkit.print( - "[yellow]gh CLI not found. Set these secrets manually:[/yellow]", - ) - toolkit.print_line() - toolkit.print(f"Repository: [blue]{secrets_url}[/]") + ) + if toolkit.mode != "json": toolkit.print_line() - toolkit.print( - f"[bold blue]FASTAPI_CLOUD_TOKEN[/] = {token_data['value']}" - ) - toolkit.print( - f"[bold blue]FASTAPI_CLOUD_APP_ID[/] = {target_app_id}" - ) - - if should_set_secrets: - with toolkit.progress( - title="Setting repo secrets...", done_emoji="🔒" - ) as progress: - try: - _set_github_secret( - "FASTAPI_CLOUD_TOKEN", token_data["value"] - ) - _set_github_secret("FASTAPI_CLOUD_APP_ID", target_app_id) - - progress.log(msg_secrets) - except GitHubSecretError: - progress.set_error( - "Failed to set GitHub secrets via gh CLI." - ) - toolkit.fail( - "api_error", - "Failed to set GitHub secrets via gh CLI.", - ) - set_github_secrets = True - else: - toolkit.print("Skipped setting GitHub Actions secrets.", emoji="⏭️") else: + should_set_secrets = False + secrets_url = ( + f"https://{_get_github_host(origin)}/{repo_slug}" + "/settings/secrets/actions" + ) toolkit.print( - "Skipped creating deploy token and GitHub secrets.", emoji="⏭️" + "[yellow]gh CLI not found. Set these secrets manually:[/yellow]", ) + toolkit.print_line() + toolkit.print(f"Repository: [blue]{secrets_url}[/]") + toolkit.print_line() + toolkit.print( + f"[bold blue]FASTAPI_CLOUD_TOKEN[/] = {token_data['value']}" + ) + toolkit.print(f"[bold blue]FASTAPI_CLOUD_APP_ID[/] = {target_app_id}") + + if should_set_secrets: + with toolkit.progress( + title="Setting repo secrets...", done_emoji="🔒" + ) as progress: + try: + _set_github_secret("FASTAPI_CLOUD_TOKEN", token_data["value"]) + _set_github_secret("FASTAPI_CLOUD_APP_ID", target_app_id) + + progress.log(msg_secrets) + except GitHubSecretError: + progress.set_error("Failed to set GitHub secrets via gh CLI.") + toolkit.fail( + "api_error", + "Failed to set GitHub secrets via gh CLI.", + ) + set_github_secrets = True + else: + toolkit.print("Skipped setting GitHub Actions secrets.", emoji="⏭️") + else: + toolkit.print( + "Skipped creating deploy token and GitHub secrets.", emoji="⏭️" + ) - toolkit.print_line() + toolkit.print_line() - if needs_workflow: - if not file and workflow_path.exists(): - resolved_workflow_path = _resolve_existing_workflow_path( - toolkit, workflow_path - ) + if needs_workflow: + if not file and workflow_path.exists(): + resolved_workflow_path = _resolve_existing_workflow_path( + toolkit, workflow_path + ) - if resolved_workflow_path is None: - needs_workflow = False - else: - workflow_path = resolved_workflow_path + if resolved_workflow_path is None: + needs_workflow = False + else: + workflow_path = resolved_workflow_path - if needs_workflow: - msg_workflow = f"Wrote [bold]{workflow_path}[/bold] (branch: {branch})" + if needs_workflow: + msg_workflow = f"Wrote [bold]{workflow_path}[/bold] (branch: {branch})" - _write_workflow_file(branch, workflow_path) - wrote_workflow = True + _write_workflow_file(branch, workflow_path) + wrote_workflow = True - toolkit.print(msg_workflow) - toolkit.print_line() + toolkit.print(msg_workflow) + toolkit.print_line() - output = CISetupOutput( - app_id=target_app_id, - repo=repo_slug, - branch=branch, - workflow_path=_format_workflow_path(workflow_path), - created_token=created_token, - set_github_secrets=set_github_secrets, - wrote_workflow=wrote_workflow, - token_expired_at=token_expired_at, - ) + output = CISetupOutput( + app_id=target_app_id, + repo=repo_slug, + branch=branch, + workflow_path=_format_workflow_path(workflow_path), + created_token=created_token, + set_github_secrets=set_github_secrets, + wrote_workflow=wrote_workflow, + token_expired_at=token_expired_at, + ) - toolkit.success(output, render_output=_render_ci_setup_output) + toolkit.success(output, render_output=_render_ci_setup_output) diff --git a/src/fastapi_cloud_cli/commands/teams/__init__.py b/src/fastapi_cloud_cli/commands/teams/__init__.py index da4cf7b4..dc1bddb9 100644 --- a/src/fastapi_cloud_cli/commands/teams/__init__.py +++ b/src/fastapi_cloud_cli/commands/teams/__init__.py @@ -1,138 +1,3 @@ -import logging -from typing import Annotated, Any +from fastapi_cloud_cli.commands.teams._app import teams_app as teams_app -import typer -from pydantic import BaseModel -from rich.markup import escape -from rich.table import Table -from rich_toolkit import RichToolkit - -from fastapi_cloud_cli.api import APIClient -from fastapi_cloud_cli.commands.teams.get import ( - Team, - _get_team_dashboard_url, - get_team, -) -from fastapi_cloud_cli.config import Settings -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__) - -DEFAULT_LIMIT = 100 -DEFAULT_OFFSET = 0 - - -class TeamsListAPIResponse(BaseModel): - data: list[Team] - count: int - - -class TeamsListOutput(BaseModel): - teams: list[Team] - total_count: int - limit: int - offset: int - - -def _get_teams(client: APIClient, *, limit: int, offset: int) -> TeamsListOutput: - response = client.get( - "/teams/", - params={ - "limit": limit, - "skip": offset, - }, - ) - response.raise_for_status() - - data = TeamsListAPIResponse.model_validate(response.json()) - - return TeamsListOutput( - teams=data.data, - total_count=data.count, - limit=limit, - offset=offset, - ) - - -def _render_teams_list_output(data: TeamsListOutput, toolkit: RichToolkit) -> None: - toolkit.print_title("teams") - toolkit.print_line() - - if not data.teams: - toolkit.print("No teams found.", bullet=False) - return - - settings = Settings.get() - - table = Table.grid(padding=(0, 2), pad_edge=False) - table.add_column("Name") - table.add_column("ID") - table.add_row("[bold]Name[/bold]", "[bold]ID[/bold]") - table.add_row("", "") - - for team in data.teams: - table.add_row( - f"[link={_get_team_dashboard_url(team, settings=settings)}]{escape(team.name)}[/link]", - team.id, - ) - - toolkit.print(table, bullet=False) - - -teams_app = typer.Typer( - no_args_is_help=True, - help="Manage your FastAPI Cloud teams.", -) - - -@teams_app.command("list") -def list_teams( - limit: Annotated[ - int, - typer.Option( - "--limit", - help="Maximum number of teams to return.", - min=1, - ), - ] = DEFAULT_LIMIT, - offset: Annotated[ - int, - typer.Option( - "--offset", - help="Offset into the team result set.", - min=0, - ), - ] = DEFAULT_OFFSET, - json_output: JsonOutputOption = False, -) -> Any: - """ - List FastAPI Cloud teams. - """ - 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 teams", - transient=True, - ) as progress: - with client.handle_http_errors( - progress, - default_message="Error fetching teams. Please try again later.", - toolkit=toolkit, - ): - result = _get_teams(client, limit=limit, offset=offset) - - toolkit.success(result, render_output=_render_teams_list_output) - - -teams_app.command("get")(get_team) +__all__ = ["teams_app"] diff --git a/src/fastapi_cloud_cli/commands/teams/_app.py b/src/fastapi_cloud_cli/commands/teams/_app.py new file mode 100644 index 00000000..1101d95d --- /dev/null +++ b/src/fastapi_cloud_cli/commands/teams/_app.py @@ -0,0 +1,6 @@ +import typer + +teams_app = typer.Typer( + no_args_is_help=True, + help="Manage your FastAPI Cloud teams.", +) diff --git a/src/fastapi_cloud_cli/commands/teams/get.py b/src/fastapi_cloud_cli/commands/teams/get.py index 71a88984..56efdec0 100644 --- a/src/fastapi_cloud_cli/commands/teams/get.py +++ b/src/fastapi_cloud_cli/commands/teams/get.py @@ -6,9 +6,10 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.teams._app import teams_app from fastapi_cloud_cli.config import Settings -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.cli import get_details_table from fastapi_cloud_cli.utils.execution import JsonOutputOption logger = logging.getLogger(__name__) @@ -51,7 +52,9 @@ def _render_team_get_output(data: TeamGetOutput, toolkit: RichToolkit) -> None: ) +@teams_app.command("get", cls=UserCommand) def get_team( + ctx: typer.Context, team_id: Annotated[ str, typer.Argument( @@ -63,29 +66,22 @@ def get_team( """ Get a FastAPI Cloud team by ID. """ - 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, - toolkit.progress( - title="Fetching team", - transient=True, - ) as progress, + + toolkit = get_user_command_context(ctx).toolkit + + with ( + APIClient() as client, + toolkit.progress( + title="Fetching team", + transient=True, + ) as progress, + ): + with client.handle_http_errors( + progress, + default_message="Error fetching team. Please try again later.", + not_found_message="Team not found.", + toolkit=toolkit, ): - with client.handle_http_errors( - progress, - default_message="Error fetching team. Please try again later.", - not_found_message="Team not found.", - toolkit=toolkit, - ): - result = _get_team(client, team_id) - - toolkit.success(result, render_output=_render_team_get_output) + result = _get_team(client, team_id) + + toolkit.success(result, render_output=_render_team_get_output) diff --git a/src/fastapi_cloud_cli/commands/teams/list.py b/src/fastapi_cloud_cli/commands/teams/list.py new file mode 100644 index 00000000..f00d0c91 --- /dev/null +++ b/src/fastapi_cloud_cli/commands/teams/list.py @@ -0,0 +1,119 @@ +import logging +from typing import Annotated, Any + +import typer +from pydantic import BaseModel +from rich.markup import escape +from rich.table import Table +from rich_toolkit import RichToolkit + +from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.teams._app import teams_app +from fastapi_cloud_cli.commands.teams.get import Team, _get_team_dashboard_url +from fastapi_cloud_cli.config import Settings +from fastapi_cloud_cli.utils.execution import JsonOutputOption + +logger = logging.getLogger(__name__) + +DEFAULT_LIMIT = 100 +DEFAULT_OFFSET = 0 + + +class TeamsListAPIResponse(BaseModel): + data: list[Team] + count: int + + +class TeamsListOutput(BaseModel): + teams: list[Team] + total_count: int + limit: int + offset: int + + +def _get_teams(client: APIClient, *, limit: int, offset: int) -> TeamsListOutput: + response = client.get( + "/teams/", + params={ + "limit": limit, + "skip": offset, + }, + ) + response.raise_for_status() + + data = TeamsListAPIResponse.model_validate(response.json()) + + return TeamsListOutput( + teams=data.data, + total_count=data.count, + limit=limit, + offset=offset, + ) + + +def _render_teams_list_output(data: TeamsListOutput, toolkit: RichToolkit) -> None: + toolkit.print_title("teams") + toolkit.print_line() + + if not data.teams: + toolkit.print("No teams found.", bullet=False) + return + + settings = Settings.get() + + table = Table.grid(padding=(0, 2), pad_edge=False) + table.add_column("Name") + table.add_column("ID") + table.add_row("[bold]Name[/bold]", "[bold]ID[/bold]") + table.add_row("", "") + + for team in data.teams: + table.add_row( + f"[link={_get_team_dashboard_url(team, settings=settings)}]{escape(team.name)}[/link]", + team.id, + ) + + toolkit.print(table, bullet=False) + + +@teams_app.command("list", cls=UserCommand) +def list_teams( + ctx: typer.Context, + limit: Annotated[ + int, + typer.Option( + "--limit", + help="Maximum number of teams to return.", + min=1, + ), + ] = DEFAULT_LIMIT, + offset: Annotated[ + int, + typer.Option( + "--offset", + help="Offset into the team result set.", + min=0, + ), + ] = DEFAULT_OFFSET, + json_output: JsonOutputOption = False, +) -> Any: + """ + List FastAPI Cloud teams. + """ + + toolkit = get_user_command_context(ctx).toolkit + + with APIClient() as client: + with toolkit.progress( + title="Fetching teams", + transient=True, + ) as progress: + with client.handle_http_errors( + progress, + default_message="Error fetching teams. Please try again later.", + toolkit=toolkit, + ): + result = _get_teams(client, limit=limit, offset=offset) + + toolkit.success(result, render_output=_render_teams_list_output) diff --git a/src/fastapi_cloud_cli/commands/tokens/__init__.py b/src/fastapi_cloud_cli/commands/tokens/__init__.py index 40a0f16b..d8f5e579 100644 --- a/src/fastapi_cloud_cli/commands/tokens/__init__.py +++ b/src/fastapi_cloud_cli/commands/tokens/__init__.py @@ -1,15 +1,3 @@ -import typer - -from fastapi_cloud_cli.commands.tokens.create import create_token -from fastapi_cloud_cli.commands.tokens.delete import delete_token -from fastapi_cloud_cli.commands.tokens.list import list_tokens - -tokens_app = typer.Typer( - no_args_is_help=True, - help="Manage deploy tokens for your app.", -) -tokens_app.command("create")(create_token) -tokens_app.command("delete")(delete_token) -tokens_app.command("list")(list_tokens) +from fastapi_cloud_cli.commands.tokens._app import tokens_app as tokens_app __all__ = ["tokens_app"] diff --git a/src/fastapi_cloud_cli/commands/tokens/_app.py b/src/fastapi_cloud_cli/commands/tokens/_app.py new file mode 100644 index 00000000..1480ccf0 --- /dev/null +++ b/src/fastapi_cloud_cli/commands/tokens/_app.py @@ -0,0 +1,6 @@ +import typer + +tokens_app = typer.Typer( + no_args_is_help=True, + help="Manage deploy tokens for your app.", +) diff --git a/src/fastapi_cloud_cli/commands/tokens/create.py b/src/fastapi_cloud_cli/commands/tokens/create.py index 19cf7639..56d4d58d 100644 --- a/src/fastapi_cloud_cli/commands/tokens/create.py +++ b/src/fastapi_cloud_cli/commands/tokens/create.py @@ -6,9 +6,10 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.tokens._app import tokens_app 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 FastAPIRichToolkit, get_rich_toolkit +from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit from fastapi_cloud_cli.utils.execution import JsonOutputOption DEFAULT_EXPIRES_IN_DAYS = 365 @@ -95,7 +96,9 @@ def _render_deploy_token_create_output( ) +@tokens_app.command("create", cls=UserCommand) def create_token( + ctx: typer.Context, app_id: Annotated[ str | None, typer.Option( @@ -130,53 +133,46 @@ def create_token( """ Create a deploy token for 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.", - ) + toolkit = get_user_command_context(ctx).toolkit - target_app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) + target_app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) - toolkit.print_title("deploy tokens") + toolkit.print_title("deploy tokens") + toolkit.print_line() + + output_file = _resolve_output_file(toolkit, output_file=output_file) + name_needs_prompt = name is None + name = _resolve_token_name(toolkit, name=name) + if name_needs_prompt: toolkit.print_line() - output_file = _resolve_output_file(toolkit, output_file=output_file) - name_needs_prompt = name is None - name = _resolve_token_name(toolkit, name=name) - if name_needs_prompt: - toolkit.print_line() - - with APIClient() as client: - with toolkit.progress( - title="Creating deploy token", - transient=True, - ) as progress: - with client.handle_http_errors( - progress, - default_message="Error creating deploy token. Please try again later.", - not_found_message="App not found.", - toolkit=toolkit, - ): - token = _create_deploy_token( - client, - app_id=target_app_id, - name=name, - expires_in_days=expires_in_days, - ) - - _write_token_value(output_file, token.value) - - toolkit.success( - DeployTokenCreateOutput( - app_id=target_app_id, - token=CreatedDeployToken.model_validate(token), - stored_secret=StoredDeployTokenSecret(path=output_file), - output_file=output_file, - ), - render_output=_render_deploy_token_create_output, - ) + with APIClient() as client: + with toolkit.progress( + title="Creating deploy token", + transient=True, + ) as progress: + with client.handle_http_errors( + progress, + default_message="Error creating deploy token. Please try again later.", + not_found_message="App not found.", + toolkit=toolkit, + ): + token = _create_deploy_token( + client, + app_id=target_app_id, + name=name, + expires_in_days=expires_in_days, + ) + + _write_token_value(output_file, token.value) + + toolkit.success( + DeployTokenCreateOutput( + app_id=target_app_id, + token=CreatedDeployToken.model_validate(token), + stored_secret=StoredDeployTokenSecret(path=output_file), + output_file=output_file, + ), + render_output=_render_deploy_token_create_output, + ) diff --git a/src/fastapi_cloud_cli/commands/tokens/delete.py b/src/fastapi_cloud_cli/commands/tokens/delete.py index 66464a06..e3e90b2c 100644 --- a/src/fastapi_cloud_cli/commands/tokens/delete.py +++ b/src/fastapi_cloud_cli/commands/tokens/delete.py @@ -5,9 +5,9 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.tokens._app import tokens_app 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 @@ -36,7 +36,9 @@ def _render_deploy_token_delete_output( ) +@tokens_app.command("delete", cls=UserCommand) def delete_token( + ctx: typer.Context, token_id: Annotated[ str, typer.Argument( @@ -55,48 +57,41 @@ def delete_token( """ Delete a deploy token for 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.", - ) - - target_app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) - - with APIClient() as client: - with toolkit.progress( - title="Deleting deploy token", - transient=True, - ) as progress: - with client.handle_http_errors( - progress, - default_message="Error deleting deploy token. Please try again later.", - not_found_message="Deploy token not found.", - toolkit=toolkit, - ): - deleted = _delete_deploy_token( - client, - app_id=target_app_id, - token_id=token_id, - ) - - if not deleted: - message = ( - f"Deploy token {token_id} not found." - if toolkit.mode == "json" - else "Deploy token not found." - ) - toolkit.fail( - "not_found", - message, - hint="Run `fastapi cloud tokens list` to see available deploy tokens.", - ) - - toolkit.success( - DeployTokenDeleteOutput(token_id=token_id), - render_output=_render_deploy_token_delete_output, + + toolkit = get_user_command_context(ctx).toolkit + + target_app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) + + with APIClient() as client: + with toolkit.progress( + title="Deleting deploy token", + transient=True, + ) as progress: + with client.handle_http_errors( + progress, + default_message="Error deleting deploy token. Please try again later.", + not_found_message="Deploy token not found.", + toolkit=toolkit, + ): + deleted = _delete_deploy_token( + client, + app_id=target_app_id, + token_id=token_id, + ) + + if not deleted: + message = ( + f"Deploy token {token_id} not found." + if toolkit.mode == "json" + else "Deploy token not found." + ) + toolkit.fail( + "not_found", + message, + hint="Run `fastapi cloud tokens list` to see available deploy tokens.", ) + + toolkit.success( + DeployTokenDeleteOutput(token_id=token_id), + render_output=_render_deploy_token_delete_output, + ) diff --git a/src/fastapi_cloud_cli/commands/tokens/list.py b/src/fastapi_cloud_cli/commands/tokens/list.py index 3916befc..ec4ff4ba 100644 --- a/src/fastapi_cloud_cli/commands/tokens/list.py +++ b/src/fastapi_cloud_cli/commands/tokens/list.py @@ -7,9 +7,9 @@ from rich_toolkit import RichToolkit from fastapi_cloud_cli.api import APIClient +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context +from fastapi_cloud_cli.commands.tokens._app import tokens_app 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 @@ -71,7 +71,9 @@ def _render_deploy_tokens_list_output( toolkit.print(_get_deploy_tokens_table(data.tokens), bullet=False) +@tokens_app.command("list", cls=UserCommand) def list_tokens( + ctx: typer.Context, app_id: Annotated[ str | None, typer.Option( @@ -84,32 +86,25 @@ def list_tokens( """ List deploy tokens for 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.", - ) - - target_app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) - - with APIClient() as client: - with toolkit.progress( - title="Fetching deploy tokens", - transient=True, - ) as progress: - with client.handle_http_errors( - progress, - default_message="Error fetching deploy tokens. Please try again later.", - not_found_message="App not found.", - toolkit=toolkit, - ): - tokens = _get_deploy_tokens(client=client, app_id=target_app_id) - - toolkit.success( - DeployTokensListOutput(app_id=target_app_id, tokens=tokens.data), - render_output=_render_deploy_tokens_list_output, - ) + + toolkit = get_user_command_context(ctx).toolkit + + target_app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) + + with APIClient() as client: + with toolkit.progress( + title="Fetching deploy tokens", + transient=True, + ) as progress: + with client.handle_http_errors( + progress, + default_message="Error fetching deploy tokens. Please try again later.", + not_found_message="App not found.", + toolkit=toolkit, + ): + tokens = _get_deploy_tokens(client=client, app_id=target_app_id) + + toolkit.success( + DeployTokensListOutput(app_id=target_app_id, tokens=tokens.data), + render_output=_render_deploy_tokens_list_output, + ) diff --git a/src/fastapi_cloud_cli/commands/whoami.py b/src/fastapi_cloud_cli/commands/whoami.py index 669f6480..25193a71 100644 --- a/src/fastapi_cloud_cli/commands/whoami.py +++ b/src/fastapi_cloud_cli/commands/whoami.py @@ -1,12 +1,13 @@ import logging from typing import Any +import typer from pydantic import BaseModel from rich_toolkit import RichToolkit +from fastapi_cloud_cli._app import cloud_app from fastapi_cloud_cli.api import APIClient -from fastapi_cloud_cli.utils.auth import Identity -from fastapi_cloud_cli.utils.cli import get_rich_toolkit +from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context from fastapi_cloud_cli.utils.execution import JsonOutputOption logger = logging.getLogger(__name__) @@ -28,41 +29,38 @@ def _render_whoami_output(data: WhoAmIOutput, toolkit: RichToolkit) -> None: ) +@cloud_app.command(cls=UserCommand) def whoami( + ctx: typer.Context, json_output: JsonOutputOption = False, ) -> Any: """ Show the currently logged in user. """ - 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 [blue]`fastapi login`[/] or set [blue]FASTAPI_CLOUD_TOKEN.[/]", - ) + command_context = get_user_command_context(ctx) + toolkit = command_context.toolkit + identity = command_context.identity - with ( - APIClient() as client, - toolkit.progress( - title="Fetching profile", - transient=True, - ) as progress, + with ( + APIClient() as client, + toolkit.progress( + title="Fetching profile", + transient=True, + ) as progress, + ): + with client.handle_http_errors( + progress, + default_message="", + toolkit=toolkit, ): - with client.handle_http_errors( - progress, - default_message="", - toolkit=toolkit, - ): - response = client.get("/users/me") - response.raise_for_status() + response = client.get("/users/me") + response.raise_for_status() - data = response.json() + data = response.json() - result = WhoAmIOutput( - has_deploy_token=identity.has_deploy_token(), email=data["email"] - ) + result = WhoAmIOutput( + has_deploy_token=identity.has_deploy_token(), email=data["email"] + ) - toolkit.success(result, render_output=_render_whoami_output) + toolkit.success(result, render_output=_render_whoami_output) diff --git a/tests/integrations/providers/test_list.py b/tests/integrations/providers/test_list.py index 7ce2beca..ac86ad63 100644 --- a/tests/integrations/providers/test_list.py +++ b/tests/integrations/providers/test_list.py @@ -86,7 +86,7 @@ def test_lists_providers_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" diff --git a/tests/integrations/resources/test_connect.py b/tests/integrations/resources/test_connect.py index 43d39958..933cc5db 100644 --- a/tests/integrations/resources/test_connect.py +++ b/tests/integrations/resources/test_connect.py @@ -66,7 +66,7 @@ def test_connect_resource_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" diff --git a/tests/integrations/resources/test_disconnect.py b/tests/integrations/resources/test_disconnect.py index 21f633b3..d5bd15f7 100644 --- a/tests/integrations/resources/test_disconnect.py +++ b/tests/integrations/resources/test_disconnect.py @@ -80,7 +80,7 @@ def test_disconnect_resource_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" diff --git a/tests/integrations/resources/test_get.py b/tests/integrations/resources/test_get.py index 7171647a..f18f2c8e 100644 --- a/tests/integrations/resources/test_get.py +++ b/tests/integrations/resources/test_get.py @@ -100,7 +100,7 @@ def test_gets_resource_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" diff --git a/tests/integrations/resources/test_list.py b/tests/integrations/resources/test_list.py index 1f79c736..a15ac1cd 100644 --- a/tests/integrations/resources/test_list.py +++ b/tests/integrations/resources/test_list.py @@ -55,7 +55,7 @@ def test_lists_resources_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" diff --git a/tests/test_cli_apps.py b/tests/test_cli_apps.py index fc46f77d..286cb1c5 100644 --- a/tests/test_cli_apps.py +++ b/tests/test_cli_apps.py @@ -43,7 +43,7 @@ def test_creates_app_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" @@ -379,7 +379,7 @@ def test_updates_app_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" @@ -716,7 +716,7 @@ def test_gets_app_human_returns_not_logged_in_when_logged_out( assert result.output == snapshot("""\ ✗ error: No credentials found. - hint: Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.\ + hint: Run `fastapi cloud login`.\ """) @@ -1018,5 +1018,5 @@ def test_lists_apps_human_returns_not_logged_in_when_logged_out( assert result.output == snapshot("""\ ✗ error: No credentials found. - hint: Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.\ + hint: Run `fastapi cloud login`.\ """) diff --git a/tests/test_cli_deployments.py b/tests/test_cli_deployments.py index 417dd764..c3c970d1 100644 --- a/tests/test_cli_deployments.py +++ b/tests/test_cli_deployments.py @@ -172,7 +172,7 @@ def test_lists_deployments_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" @@ -214,7 +214,7 @@ def test_gets_deployment_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" @@ -631,7 +631,7 @@ def test_build_logs_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" diff --git a/tests/test_cli_link.py b/tests/test_cli_link.py index 63409944..5d54caeb 100644 --- a/tests/test_cli_link.py +++ b/tests/test_cli_link.py @@ -37,7 +37,7 @@ def test_link_json_returns_not_logged_in_when_logged_out(logged_out_cli: None) - "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" diff --git a/tests/test_cli_teams.py b/tests/test_cli_teams.py index 3c7dddf2..9302a801 100644 --- a/tests/test_cli_teams.py +++ b/tests/test_cli_teams.py @@ -52,7 +52,7 @@ def test_lists_teams_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" @@ -135,7 +135,7 @@ def test_lists_teams_human_returns_not_logged_in_when_logged_out( assert result.output == snapshot("""\ ✗ error: No credentials found. - hint: Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.\ + hint: Run `fastapi cloud login`.\ """) @@ -151,7 +151,7 @@ def test_gets_team_human_returns_not_logged_in_when_logged_out( assert result.output == snapshot("""\ ✗ error: No credentials found. - hint: Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.\ + hint: Run `fastapi cloud login`.\ """) diff --git a/tests/test_cli_tokens.py b/tests/test_cli_tokens.py index 23579005..dc915607 100644 --- a/tests/test_cli_tokens.py +++ b/tests/test_cli_tokens.py @@ -37,7 +37,7 @@ def test_creates_token_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" @@ -63,7 +63,7 @@ def test_deletes_token_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" @@ -564,7 +564,7 @@ def test_lists_tokens_json_returns_not_logged_in_when_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi cloud login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" diff --git a/tests/test_cli_whoami.py b/tests/test_cli_whoami.py index aca5d94e..d664cc37 100644 --- a/tests/test_cli_whoami.py +++ b/tests/test_cli_whoami.py @@ -154,7 +154,7 @@ def test_prints_json_error_when_json_env_is_enabled_and_logged_out( "error": { "code": "not_logged_in", "message": "No credentials found.", - "hint": "Run `fastapi login` or set FASTAPI_CLOUD_TOKEN.", + "hint": "Run `fastapi cloud login`.", } } assert result.stderr == "" @@ -258,7 +258,7 @@ def test_prints_not_logged_in(logged_out_cli: None) -> None: assert result.exit_code == 1 assert "No credentials found." in result.output - assert "Run `fastapi login` or set FASTAPI_CLOUD_TOKEN." in result.output + assert "Run `fastapi cloud login`." in result.output def test_prints_not_logged_in_with_deploy_token(logged_out_cli: None) -> None: @@ -266,7 +266,7 @@ def test_prints_not_logged_in_with_deploy_token(logged_out_cli: None) -> None: assert result.exit_code == 1 assert "No credentials found." in result.output - assert "Run `fastapi login` or set FASTAPI_CLOUD_TOKEN." in result.output + assert "Run `fastapi cloud login`." in result.output @pytest.mark.respx