From 72044b96148e601f0d595d03ec7f8586e981e665 Mon Sep 17 00:00:00 2001 From: Marco Burro Date: Fri, 9 Jan 2026 11:47:59 +0100 Subject: [PATCH 1/2] =?UTF-8?q?=E2=9C=A8=20Add=20support=20for=20app=20ID?= =?UTF-8?q?=20in=20`fastapi=20deploy`?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/fastapi_cloud_cli/commands/deploy.py | 54 ++++- src/fastapi_cloud_cli/commands/logs.py | 249 +++++++++++++++++++++++ tests/test_cli_deploy.py | 185 +++++++++++++++++ 3 files changed, 478 insertions(+), 10 deletions(-) create mode 100644 src/fastapi_cloud_cli/commands/logs.py diff --git a/src/fastapi_cloud_cli/commands/deploy.py b/src/fastapi_cloud_cli/commands/deploy.py index 89347e2..1b51f36 100644 --- a/src/fastapi_cloud_cli/commands/deploy.py +++ b/src/fastapi_cloud_cli/commands/deploy.py @@ -557,12 +557,20 @@ def deploy( skip_wait: Annotated[ bool, typer.Option("--no-wait", help="Skip waiting for deployment status") ] = False, + app_id: Annotated[ + Union[str, None], + typer.Option( + "--app-id", + help="Application ID to deploy to", + envvar="FASTAPI_CLOUD_APP_ID", + ), + ] = None, ) -> Any: """ Deploy a [bold]FastAPI[/bold] app to FastAPI Cloud. 🚀 """ logger.debug("Deploy command started") - logger.debug("Deploy path: %s, skip_wait: %s", path, skip_wait) + logger.debug("Deploy path: %s, skip_wait: %s, app_id: %s", path, skip_wait, app_id) identity = Identity() @@ -604,19 +612,44 @@ def deploy( app_config = get_app_config(path_to_deploy) - if not app_config: + if app_id and app_config and app_id != app_config.app_id: + logger.debug( + "Provided app_id (%s) differs from local config (%s)", + app_id, + app_config.app_id, + ) + + toolkit.print( + f"[error]Error: Provided app ID ({app_id}) does not match the local " + f"config ({app_config.app_id}).[/]" + ) + toolkit.print_line() + toolkit.print( + "Run [bold]fastapi cloud unlink[/] to remove the local config, " + "or remove --app-id / unset FASTAPI_CLOUD_APP_ID to use the configured app.", + tag="tip", + ) + raise typer.Exit(1) from None + + target_app_id = app_id or (app_config.app_id if app_config else None) + + if not target_app_id: logger.debug("No app config found, configuring new app") app_config = _configure_app(toolkit, path_to_deploy=path_to_deploy) toolkit.print_line() + target_app_id = app_config.app_id + + if app_id: + toolkit.print(f"Deploying to app [blue]{target_app_id}[/blue]...") else: - logger.debug("Existing app config found, proceeding with deployment") toolkit.print("Deploying app...") - toolkit.print_line() + + toolkit.print_line() with toolkit.progress("Checking app...", transient=True) as progress: with handle_http_errors(progress): - logger.debug("Checking app with ID: %s", app_config.app_id) - app = _get_app(app_config.app_id) + logger.debug("Checking app with ID: %s", target_app_id) + app = _get_app(target_app_id) if not app: logger.debug("App not found in API") @@ -626,10 +659,11 @@ def deploy( if not app: toolkit.print_line() - toolkit.print( - "If you deleted this app, you can run [bold]fastapi cloud unlink[/] to unlink the local configuration.", - tag="tip", - ) + if not app_id: + toolkit.print( + "If you deleted this app, you can run [bold]fastapi cloud unlink[/] to unlink the local configuration.", + tag="tip", + ) raise typer.Exit(1) with tempfile.TemporaryDirectory() as temp_dir: diff --git a/src/fastapi_cloud_cli/commands/logs.py b/src/fastapi_cloud_cli/commands/logs.py new file mode 100644 index 0000000..877ab58 --- /dev/null +++ b/src/fastapi_cloud_cli/commands/logs.py @@ -0,0 +1,249 @@ +import json +import logging +import time +from collections.abc import Generator +from datetime import datetime +from pathlib import Path +from typing import Annotated, Optional + +import typer +from httpx import HTTPError, HTTPStatusError, ReadTimeout +from pydantic import BaseModel, ValidationError +from rich.markup import escape +from rich_toolkit import RichToolkit + +from fastapi_cloud_cli.utils.api import APIClient +from fastapi_cloud_cli.utils.apps import AppConfig, get_app_config +from fastapi_cloud_cli.utils.cli import get_rich_toolkit + +logger = logging.getLogger(__name__) + +MAX_RECONNECT_ATTEMPTS = 10 +RECONNECT_DELAY_SECONDS = 1 +LOG_LEVEL_COLORS = { + "debug": "blue", + "info": "cyan", + "warning": "yellow", + "warn": "yellow", + "error": "red", + "critical": "magenta", + "fatal": "magenta", +} + + +class LogEntry(BaseModel): + timestamp: datetime + message: str + level: str = "unknown" + + +def _stream_logs( + app_id: str, + tail: int, + since: str, + follow: bool, +) -> Generator[str, None, None]: + with APIClient() as client: + timeout = 120 if follow else 30 + with client.stream( + "GET", + f"/apps/{app_id}/logs/stream", + params={ + "tail": tail, + "since": since, + "follow": follow, + }, + timeout=timeout, + ) as response: + response.raise_for_status() + + yield from response.iter_lines() + + +def _format_log_line(log: LogEntry) -> str: + """Format a log entry for display with a colored indicator""" + timestamp_str = log.timestamp.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" + color = LOG_LEVEL_COLORS.get(log.level.lower()) + + message = escape(log.message) + + if color: + return f"[{color}]┃[/{color}] [dim]{timestamp_str}[/dim] {message}" + + return f"[dim]┃[/dim] [dim]{timestamp_str}[/dim] {message}" + + +def _process_log_stream( + toolkit: RichToolkit, + app_config: AppConfig, + tail: int, + since: str, + follow: bool, +) -> None: + log_count = 0 + last_timestamp: datetime | None = None + current_since = since + current_tail = tail + reconnect_attempts = 0 + + while True: + try: + for line in _stream_logs( + app_id=app_config.app_id, + tail=current_tail, + since=current_since, + follow=follow, + ): + if not line: # pragma: no cover + continue + + try: + data = json.loads(line) + except json.JSONDecodeError: + logger.debug("Failed to parse log line: %s", line) + continue + + # Skip heartbeat messages + if data.get("type") == "heartbeat": # pragma: no cover + continue + + if data.get("type") == "error": + toolkit.print( + f"Error: {data.get('message', 'Unknown error')}", + ) + raise typer.Exit(1) + + # Parse and display log entry + try: + log_entry = LogEntry.model_validate(data) + toolkit.print(_format_log_line(log_entry)) + log_count += 1 + last_timestamp = log_entry.timestamp + # Reset reconnect attempts on successful log receipt + reconnect_attempts = 0 + except ValidationError as e: # pragma: no cover + logger.debug("Failed to parse log entry: %s - %s", data, e) + continue + + # Stream ended normally (only happens with --no-follow) + if not follow and log_count == 0: + toolkit.print("No logs found for the specified time range.") + break + + except KeyboardInterrupt: # pragma: no cover + toolkit.print_line() + break + except (ReadTimeout, HTTPError) as e: + # In follow mode, try to reconnect on connection issues + if follow and not isinstance(e, HTTPStatusError): + reconnect_attempts += 1 + if reconnect_attempts >= MAX_RECONNECT_ATTEMPTS: + toolkit.print( + "Lost connection to log stream. Please try again later.", + ) + raise typer.Exit(1) from None + + logger.debug( + "Connection lost, reconnecting (attempt %d/%d)...", + reconnect_attempts, + MAX_RECONNECT_ATTEMPTS, + ) + + # On reconnect, resume from last seen timestamp + # The API uses strict > comparison, so logs with the same timestamp + # as last_timestamp will be filtered out (no duplicates) + if last_timestamp: # pragma: no cover + current_since = last_timestamp.isoformat() + current_tail = 0 # Don't fetch historical logs again + + time.sleep(RECONNECT_DELAY_SECONDS) + continue + + if isinstance(e, HTTPStatusError) and e.response.status_code in (401, 403): + toolkit.print( + "The specified token is not valid. Use [blue]`fastapi login`[/] to generate a new token.", + ) + if isinstance(e, HTTPStatusError) and e.response.status_code == 404: + toolkit.print( + "App not found. Make sure to use the correct account.", + ) + elif isinstance(e, ReadTimeout): + toolkit.print( + "The request timed out. Please try again later.", + ) + else: + logger.exception("Failed to fetch logs") + + toolkit.print( + "Failed to fetch logs. Please try again later.", + ) + raise typer.Exit(1) from None + + +def logs( + path: Annotated[ + Optional[Path], + typer.Argument( + help="Path to the folder containing the app (defaults to current directory)" + ), + ] = None, + tail: int = typer.Option( + 100, + "--tail", + "-t", + help="Number of log lines to show before streaming.", + show_default=True, + ), + since: str = typer.Option( + "5m", + "--since", + "-s", + help="Show logs since a specific time (e.g., '5m', '1h', '2d').", + show_default=True, + ), + follow: bool = typer.Option( + True, + "--follow/--no-follow", + "-f", + help="Stream logs in real-time (use --no-follow to fetch and exit).", + ), +) -> None: + """Stream or fetch logs from your deployed app.""" + with get_rich_toolkit(minimal=True) as toolkit: + if not is_logged_in(): + toolkit.print( + "No credentials found. Use [blue]`fastapi login`[/] to login.", + tag="auth", + ) + raise typer.Exit(1) + + app_path = path or Path.cwd() + app_config = get_app_config(app_path) + + if not app_config: + toolkit.print( + "No app linked to this directory. Run [blue]`fastapi deploy`[/] first.", + ) + raise typer.Exit(1) + + logger.debug("Fetching logs for app ID: %s", app_config.app_id) + + if follow: + toolkit.print( + f"Streaming logs for [bold]{app_config.app_id}[/bold] (Ctrl+C to exit)...", + tag="logs", + ) + else: + toolkit.print( + f"Fetching logs for [bold]{app_config.app_id}[/bold]...", + tag="logs", + ) + toolkit.print_line() + + _process_log_stream( + toolkit=toolkit, + app_config=app_config, + tail=tail, + since=since, + follow=follow, + ) diff --git a/tests/test_cli_deploy.py b/tests/test_cli_deploy.py index 8f8382b..72ea419 100644 --- a/tests/test_cli_deploy.py +++ b/tests/test_cli_deploy.py @@ -1216,3 +1216,188 @@ def test_deploy_with_token_fails( "The specified token is not valid. Make sure to use a valid token." in result.output ) + + +@pytest.mark.respx(base_url=settings.base_api_url) +def test_deploy_with_app_id_arg( + logged_in_cli: None, tmp_path: Path, respx_mock: respx.MockRouter +) -> None: + app_data = _get_random_app() + app_id = app_data["id"] + deployment_data = _get_random_deployment(app_id=app_id) + + respx_mock.get(f"/apps/{app_id}").mock(return_value=Response(200, json=app_data)) + + respx_mock.post(f"/apps/{app_id}/deployments/").mock( + return_value=Response(201, json=deployment_data) + ) + + respx_mock.post(f"/deployments/{deployment_data['id']}/upload").mock( + return_value=Response( + 200, + json={"url": "http://test.com", "fields": {"key": "value"}}, + ) + ) + + respx_mock.post("http://test.com", data={"key": "value"}).mock( + return_value=Response(200) + ) + + respx_mock.post(f"/deployments/{deployment_data['id']}/upload-complete").mock( + return_value=Response(200) + ) + + respx_mock.get(f"/deployments/{deployment_data['id']}/build-logs").mock( + return_value=Response( + 200, + content=build_logs_response( + {"type": "message", "message": "Building...", "id": "1"}, + {"type": "complete"}, + ), + ) + ) + + with changing_dir(tmp_path): + result = runner.invoke(app, ["deploy", "--app-id", app_id]) + + assert result.exit_code == 0 + assert f"Deploying to app {app_id}" in result.output + + +@pytest.mark.respx(base_url=settings.base_api_url) +def test_deploy_with_app_id_from_env_var( + logged_in_cli: None, tmp_path: Path, respx_mock: respx.MockRouter +) -> None: + app_data = _get_random_app() + app_id = app_data["id"] + deployment_data = _get_random_deployment(app_id=app_id) + + respx_mock.get(f"/apps/{app_id}").mock(return_value=Response(200, json=app_data)) + + respx_mock.post(f"/apps/{app_id}/deployments/").mock( + return_value=Response(201, json=deployment_data) + ) + + respx_mock.post(f"/deployments/{deployment_data['id']}/upload").mock( + return_value=Response( + 200, + json={"url": "http://test.com", "fields": {"key": "value"}}, + ) + ) + + respx_mock.post("http://test.com", data={"key": "value"}).mock( + return_value=Response(200) + ) + + respx_mock.post(f"/deployments/{deployment_data['id']}/upload-complete").mock( + return_value=Response(200) + ) + + respx_mock.get(f"/deployments/{deployment_data['id']}/build-logs").mock( + return_value=Response( + 200, + content=build_logs_response( + {"type": "message", "message": "Building...", "id": "1"}, + {"type": "complete"}, + ), + ) + ) + + with changing_dir(tmp_path): + result = runner.invoke(app, ["deploy"], env={"FASTAPI_CLOUD_APP_ID": app_id}) + + assert result.exit_code == 0 + assert f"Deploying to app {app_id}" in result.output + + +@pytest.mark.respx(base_url=settings.base_api_url) +def test_deploy_with_app_id_matching_local_config( + logged_in_cli: None, tmp_path: Path, respx_mock: respx.MockRouter +) -> None: + app_data = _get_random_app() + app_id = app_data["id"] + team_id = "some-team-id" + deployment_data = _get_random_deployment(app_id=app_id) + + config_path = tmp_path / ".fastapicloud" / "cloud.json" + config_path.parent.mkdir(parents=True, exist_ok=True) + config_path.write_text(f'{{"app_id": "{app_id}", "team_id": "{team_id}"}}') + + respx_mock.get(f"/apps/{app_id}").mock(return_value=Response(200, json=app_data)) + + respx_mock.post(f"/apps/{app_id}/deployments/").mock( + return_value=Response(201, json=deployment_data) + ) + + respx_mock.post(f"/deployments/{deployment_data['id']}/upload").mock( + return_value=Response( + 200, + json={"url": "http://test.com", "fields": {"key": "value"}}, + ) + ) + + respx_mock.post("http://test.com", data={"key": "value"}).mock( + return_value=Response(200) + ) + + respx_mock.post(f"/deployments/{deployment_data['id']}/upload-complete").mock( + return_value=Response(200) + ) + + respx_mock.get(f"/deployments/{deployment_data['id']}/build-logs").mock( + return_value=Response( + 200, + content=build_logs_response( + {"type": "message", "message": "Building...", "id": "1"}, + {"type": "complete"}, + ), + ) + ) + + with changing_dir(tmp_path): + result = runner.invoke(app, ["deploy", "--app-id", app_id]) + + assert result.exit_code == 0 + # Should NOT show mismatch warning + assert "does not match" not in result.output + assert f"Deploying to app {app_id}" in result.output + + +@pytest.mark.respx(base_url=settings.base_api_url) +def test_deploy_with_app_id_mismatch_fails( + logged_in_cli: None, tmp_path: Path, respx_mock: respx.MockRouter +) -> None: + local_app_data = _get_random_app() + local_app_id = local_app_data["id"] + team_id = "some-team-id" + + config_path = tmp_path / ".fastapicloud" / "cloud.json" + config_path.parent.mkdir(parents=True, exist_ok=True) + config_path.write_text(f'{{"app_id": "{local_app_id}", "team_id": "{team_id}"}}') + + cli_app_id = "different-app-id" + + with changing_dir(tmp_path): + result = runner.invoke(app, ["deploy", "--app-id", cli_app_id]) + + assert result.exit_code == 1 + assert "does not match" in result.output + assert "fastapi cloud unlink" in result.output + assert "FASTAPI_CLOUD_APP_ID" in result.output + + +@pytest.mark.respx(base_url=settings.base_api_url) +def test_deploy_with_app_id_arg_app_not_found( + logged_in_cli: None, tmp_path: Path, respx_mock: respx.MockRouter +) -> None: + app_id = "nonexistent-app-id" + + respx_mock.get(f"/apps/{app_id}").mock(return_value=Response(404)) + + with changing_dir(tmp_path): + result = runner.invoke(app, ["deploy", "--app-id", app_id]) + + assert result.exit_code == 1 + assert "App not found" in result.output + # Should NOT show unlink tip when using --app-id + assert "unlink" not in result.output From 8fc86c7cbd711d89692a581a3a5bcdfc3a97c9aa Mon Sep 17 00:00:00 2001 From: Patrick Arminio Date: Fri, 9 Jan 2026 16:13:19 +0000 Subject: [PATCH 2/2] If --- src/fastapi_cloud_cli/commands/deploy.py | 32 +-- src/fastapi_cloud_cli/commands/logs.py | 249 ----------------------- 2 files changed, 17 insertions(+), 264 deletions(-) delete mode 100644 src/fastapi_cloud_cli/commands/logs.py diff --git a/src/fastapi_cloud_cli/commands/deploy.py b/src/fastapi_cloud_cli/commands/deploy.py index 1b51f36..a3c8a88 100644 --- a/src/fastapi_cloud_cli/commands/deploy.py +++ b/src/fastapi_cloud_cli/commands/deploy.py @@ -557,7 +557,7 @@ def deploy( skip_wait: Annotated[ bool, typer.Option("--no-wait", help="Skip waiting for deployment status") ] = False, - app_id: Annotated[ + provided_app_id: Annotated[ Union[str, None], typer.Option( "--app-id", @@ -570,7 +570,9 @@ def deploy( Deploy a [bold]FastAPI[/bold] app to FastAPI Cloud. 🚀 """ logger.debug("Deploy command started") - logger.debug("Deploy path: %s, skip_wait: %s, app_id: %s", path, skip_wait, app_id) + logger.debug( + "Deploy path: %s, skip_wait: %s, app_id: %s", path, skip_wait, provided_app_id + ) identity = Identity() @@ -612,15 +614,9 @@ def deploy( app_config = get_app_config(path_to_deploy) - if app_id and app_config and app_id != app_config.app_id: - logger.debug( - "Provided app_id (%s) differs from local config (%s)", - app_id, - app_config.app_id, - ) - + if app_config and provided_app_id and app_config.app_id != provided_app_id: toolkit.print( - f"[error]Error: Provided app ID ({app_id}) does not match the local " + f"[error]Error: Provided app ID ({provided_app_id}) does not match the local " f"config ({app_config.app_id}).[/]" ) toolkit.print_line() @@ -629,17 +625,22 @@ def deploy( "or remove --app-id / unset FASTAPI_CLOUD_APP_ID to use the configured app.", tag="tip", ) - raise typer.Exit(1) from None - target_app_id = app_id or (app_config.app_id if app_config else None) + raise typer.Exit(1) from None - if not target_app_id: + if provided_app_id: + target_app_id = provided_app_id + elif app_config: + target_app_id = app_config.app_id + else: logger.debug("No app config found, configuring new app") + app_config = _configure_app(toolkit, path_to_deploy=path_to_deploy) toolkit.print_line() + target_app_id = app_config.app_id - if app_id: + if provided_app_id: toolkit.print(f"Deploying to app [blue]{target_app_id}[/blue]...") else: toolkit.print("Deploying app...") @@ -659,7 +660,8 @@ def deploy( if not app: toolkit.print_line() - if not app_id: + + if not provided_app_id: toolkit.print( "If you deleted this app, you can run [bold]fastapi cloud unlink[/] to unlink the local configuration.", tag="tip", diff --git a/src/fastapi_cloud_cli/commands/logs.py b/src/fastapi_cloud_cli/commands/logs.py deleted file mode 100644 index 877ab58..0000000 --- a/src/fastapi_cloud_cli/commands/logs.py +++ /dev/null @@ -1,249 +0,0 @@ -import json -import logging -import time -from collections.abc import Generator -from datetime import datetime -from pathlib import Path -from typing import Annotated, Optional - -import typer -from httpx import HTTPError, HTTPStatusError, ReadTimeout -from pydantic import BaseModel, ValidationError -from rich.markup import escape -from rich_toolkit import RichToolkit - -from fastapi_cloud_cli.utils.api import APIClient -from fastapi_cloud_cli.utils.apps import AppConfig, get_app_config -from fastapi_cloud_cli.utils.cli import get_rich_toolkit - -logger = logging.getLogger(__name__) - -MAX_RECONNECT_ATTEMPTS = 10 -RECONNECT_DELAY_SECONDS = 1 -LOG_LEVEL_COLORS = { - "debug": "blue", - "info": "cyan", - "warning": "yellow", - "warn": "yellow", - "error": "red", - "critical": "magenta", - "fatal": "magenta", -} - - -class LogEntry(BaseModel): - timestamp: datetime - message: str - level: str = "unknown" - - -def _stream_logs( - app_id: str, - tail: int, - since: str, - follow: bool, -) -> Generator[str, None, None]: - with APIClient() as client: - timeout = 120 if follow else 30 - with client.stream( - "GET", - f"/apps/{app_id}/logs/stream", - params={ - "tail": tail, - "since": since, - "follow": follow, - }, - timeout=timeout, - ) as response: - response.raise_for_status() - - yield from response.iter_lines() - - -def _format_log_line(log: LogEntry) -> str: - """Format a log entry for display with a colored indicator""" - timestamp_str = log.timestamp.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" - color = LOG_LEVEL_COLORS.get(log.level.lower()) - - message = escape(log.message) - - if color: - return f"[{color}]┃[/{color}] [dim]{timestamp_str}[/dim] {message}" - - return f"[dim]┃[/dim] [dim]{timestamp_str}[/dim] {message}" - - -def _process_log_stream( - toolkit: RichToolkit, - app_config: AppConfig, - tail: int, - since: str, - follow: bool, -) -> None: - log_count = 0 - last_timestamp: datetime | None = None - current_since = since - current_tail = tail - reconnect_attempts = 0 - - while True: - try: - for line in _stream_logs( - app_id=app_config.app_id, - tail=current_tail, - since=current_since, - follow=follow, - ): - if not line: # pragma: no cover - continue - - try: - data = json.loads(line) - except json.JSONDecodeError: - logger.debug("Failed to parse log line: %s", line) - continue - - # Skip heartbeat messages - if data.get("type") == "heartbeat": # pragma: no cover - continue - - if data.get("type") == "error": - toolkit.print( - f"Error: {data.get('message', 'Unknown error')}", - ) - raise typer.Exit(1) - - # Parse and display log entry - try: - log_entry = LogEntry.model_validate(data) - toolkit.print(_format_log_line(log_entry)) - log_count += 1 - last_timestamp = log_entry.timestamp - # Reset reconnect attempts on successful log receipt - reconnect_attempts = 0 - except ValidationError as e: # pragma: no cover - logger.debug("Failed to parse log entry: %s - %s", data, e) - continue - - # Stream ended normally (only happens with --no-follow) - if not follow and log_count == 0: - toolkit.print("No logs found for the specified time range.") - break - - except KeyboardInterrupt: # pragma: no cover - toolkit.print_line() - break - except (ReadTimeout, HTTPError) as e: - # In follow mode, try to reconnect on connection issues - if follow and not isinstance(e, HTTPStatusError): - reconnect_attempts += 1 - if reconnect_attempts >= MAX_RECONNECT_ATTEMPTS: - toolkit.print( - "Lost connection to log stream. Please try again later.", - ) - raise typer.Exit(1) from None - - logger.debug( - "Connection lost, reconnecting (attempt %d/%d)...", - reconnect_attempts, - MAX_RECONNECT_ATTEMPTS, - ) - - # On reconnect, resume from last seen timestamp - # The API uses strict > comparison, so logs with the same timestamp - # as last_timestamp will be filtered out (no duplicates) - if last_timestamp: # pragma: no cover - current_since = last_timestamp.isoformat() - current_tail = 0 # Don't fetch historical logs again - - time.sleep(RECONNECT_DELAY_SECONDS) - continue - - if isinstance(e, HTTPStatusError) and e.response.status_code in (401, 403): - toolkit.print( - "The specified token is not valid. Use [blue]`fastapi login`[/] to generate a new token.", - ) - if isinstance(e, HTTPStatusError) and e.response.status_code == 404: - toolkit.print( - "App not found. Make sure to use the correct account.", - ) - elif isinstance(e, ReadTimeout): - toolkit.print( - "The request timed out. Please try again later.", - ) - else: - logger.exception("Failed to fetch logs") - - toolkit.print( - "Failed to fetch logs. Please try again later.", - ) - raise typer.Exit(1) from None - - -def logs( - path: Annotated[ - Optional[Path], - typer.Argument( - help="Path to the folder containing the app (defaults to current directory)" - ), - ] = None, - tail: int = typer.Option( - 100, - "--tail", - "-t", - help="Number of log lines to show before streaming.", - show_default=True, - ), - since: str = typer.Option( - "5m", - "--since", - "-s", - help="Show logs since a specific time (e.g., '5m', '1h', '2d').", - show_default=True, - ), - follow: bool = typer.Option( - True, - "--follow/--no-follow", - "-f", - help="Stream logs in real-time (use --no-follow to fetch and exit).", - ), -) -> None: - """Stream or fetch logs from your deployed app.""" - with get_rich_toolkit(minimal=True) as toolkit: - if not is_logged_in(): - toolkit.print( - "No credentials found. Use [blue]`fastapi login`[/] to login.", - tag="auth", - ) - raise typer.Exit(1) - - app_path = path or Path.cwd() - app_config = get_app_config(app_path) - - if not app_config: - toolkit.print( - "No app linked to this directory. Run [blue]`fastapi deploy`[/] first.", - ) - raise typer.Exit(1) - - logger.debug("Fetching logs for app ID: %s", app_config.app_id) - - if follow: - toolkit.print( - f"Streaming logs for [bold]{app_config.app_id}[/bold] (Ctrl+C to exit)...", - tag="logs", - ) - else: - toolkit.print( - f"Fetching logs for [bold]{app_config.app_id}[/bold]...", - tag="logs", - ) - toolkit.print_line() - - _process_log_stream( - toolkit=toolkit, - app_config=app_config, - tail=tail, - since=since, - follow=follow, - )