♻️ Improve how we handle invalid tokens

Shortcake-Parent: main

Author: patrick91

src/fastapi_cloud_cli/commands/deploy.py

@@ -278,7 +278,7 @@ def _configure_app(toolkit: RichToolkit, path_to_deploy: Path) -> AppConfig:
 
     with toolkit.progress("Fetching teams...") as progress:
         with handle_http_errors(
-            progress, message="Error fetching teams. Please try again later."
+            progress, default_message="Error fetching teams. Please try again later."
         ):
             teams = _get_teams()
 
@@ -303,7 +303,7 @@ def _configure_app(toolkit: RichToolkit, path_to_deploy: Path) -> AppConfig:
     if not create_new_app:
         with toolkit.progress("Fetching apps...") as progress:
             with handle_http_errors(
-                progress, message="Error fetching apps. Please try again later."
+                progress, default_message="Error fetching apps. Please try again later."
             ):
                 apps = _get_apps(team.id)
 
@@ -586,10 +586,16 @@ def deploy(
             toolkit.print_title("Welcome to FastAPI Cloud!", tag="FastAPI")
             toolkit.print_line()
 
-            toolkit.print(
-                "You need to be logged in to deploy to FastAPI Cloud.",
-                tag="info",
-            )
+            if identity.token and identity.is_expired():
+                toolkit.print(
+                    "Your session has expired. Please log in again.",
+                    tag="info",
+                )
+            else:
+                toolkit.print(
+                    "You need to be logged in to deploy to FastAPI Cloud.",
+                    tag="info",
+                )
             toolkit.print_line()
 
             choice = toolkit.ask(
@@ -601,8 +607,6 @@ def deploy(
                 ],
             )
 
-            toolkit.print_line()
-
             if choice == "login":
                 login()
             else:

src/fastapi_cloud_cli/commands/link.py

@@ -49,7 +49,8 @@ def link() -> Any:
 
         with toolkit.progress("Fetching teams...") as progress:
             with handle_http_errors(
-                progress, message="Error fetching teams. Please try again later."
+                progress,
+                default_message="Error fetching teams. Please try again later.",
             ):
                 with APIClient() as client:
                     response = client.get("/teams/")
@@ -77,7 +78,7 @@ def link() -> Any:
 
         with toolkit.progress("Fetching apps...") as progress:
             with handle_http_errors(
-                progress, message="Error fetching apps. Please try again later."
+                progress, default_message="Error fetching apps. Please try again later."
             ):
                 with APIClient() as client:
                     response = client.get("/apps/", params={"team_id": team["id"]})

src/fastapi_cloud_cli/commands/login.py

@@ -78,17 +78,13 @@ def login() -> Any:
     """
     identity = Identity()
 
-    if identity.is_expired():
-        with get_rich_toolkit(minimal=True) as toolkit:
-            toolkit.print("Your session has expired. Logging in again...")
-            toolkit.print_line()
-
     if identity.is_logged_in():
         with get_rich_toolkit(minimal=True) as toolkit:
             toolkit.print("You are already logged in.")
             toolkit.print(
                 "Run [bold]fastapi cloud logout[/bold] first if you want to switch accounts."
             )
+
         return
 
     with get_rich_toolkit() as toolkit, APIClient() as client:

src/fastapi_cloud_cli/commands/logs.py

@@ -5,6 +5,7 @@
 from typing import Annotated, Optional
 
 import typer
+from httpx import HTTPError
 from rich.markup import escape
 from rich_toolkit import RichToolkit
 
@@ -16,7 +17,7 @@
 )
 from fastapi_cloud_cli.utils.apps import AppConfig, get_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.cli import get_rich_toolkit, handle_http_error
 
 logger = logging.getLogger(__name__)
 
@@ -85,21 +86,20 @@ def _process_log_stream(
                 return
     except KeyboardInterrupt:  # pragma: no cover
         toolkit.print_line()
+
         return
     except StreamLogError as e:
-        error_msg = str(e)
-        if "HTTP 401" in error_msg or "HTTP 403" in error_msg:
-            toolkit.print(
-                "The specified token is not valid. Use [blue]`fastapi login`[/] to generate a new token.",
-            )
-        elif "HTTP 404" in error_msg:
-            toolkit.print(
-                "App not found. Make sure to use the correct account.",
-            )
+        if e.status_code == 404:
+            message = "App not found. Make sure to use the correct account."
+
+        elif isinstance(e.__cause__, HTTPError):
+            message = handle_http_error(e.__cause__)
+
         else:
-            toolkit.print(
-                f"[red]Error:[/] {escape(error_msg)}",
-            )
+            message = f"[red]Error:[/] {escape(str(e))}"
+
+        toolkit.print(message)
+
         raise typer.Exit(1) from None
     except (TooManyRetriesError, TimeoutError):
         toolkit.print(

src/fastapi_cloud_cli/commands/whoami.py

@@ -24,7 +24,7 @@ def whoami() -> Any:
 
     with APIClient() as client:
         with Progress(title="⚡ Fetching profile", transient=True) as progress:
-            with handle_http_errors(progress, message=""):
+            with handle_http_errors(progress, default_message=""):
                 response = client.get("/users/me")
                 response.raise_for_status()
 

src/fastapi_cloud_cli/utils/api.py

@@ -32,7 +32,9 @@
 class StreamLogError(Exception):
     """Raised when there's an error streaming logs (build or app logs)."""
 
-    pass
+    def __init__(self, message: str, *, status_code: Optional[int] = None) -> None:
+        super().__init__(message)
+        self.status_code = status_code
 
 
 class TooManyRetriesError(Exception):
@@ -100,7 +102,8 @@ def _backoff() -> None:
             except Exception:
                 error_detail = "(response body unavailable)"
             raise StreamLogError(
-                f"HTTP {error.response.status_code}: {error_detail}"
+                f"HTTP {error.response.status_code}: {error_detail}",
+                status_code=error.response.status_code,
             ) from error
 
 

src/fastapi_cloud_cli/utils/cli.py

@@ -10,7 +10,7 @@
 from rich_toolkit.progress import Progress
 from rich_toolkit.styles import MinimalStyle, TaggedStyle
 
-from .auth import Identity
+from .auth import Identity, delete_auth_config
 
 logger = logging.getLogger(__name__)
 
@@ -68,10 +68,50 @@ def get_rich_toolkit(minimal: bool = False) -> RichToolkit:
     return RichToolkit(theme=theme)
 
 
+def handle_unauthorized() -> str:
+    message = "The specified token is not valid. "
+
+    identity = Identity()
+
+    if identity.auth_mode == "user":
+        delete_auth_config()
+
+        message += "Use `fastapi login` to generate a new token."
+    else:
+        message += "Make sure to use a valid token."
+
+    return message
+
+
+def handle_http_error(error: HTTPError, default_message: Optional[str] = None) -> str:
+    message: Optional[str] = None
+
+    if isinstance(error, HTTPStatusError):
+        status_code = error.response.status_code
+
+        # Handle validation errors from Pydantic models, this should make it easier to debug :)
+        if status_code == 422:
+            logger.debug(error.response.json())  # pragma: no cover
+
+        elif status_code == 401:
+            message = handle_unauthorized()
+
+        elif status_code == 403:
+            message = "You don't have permissions for this resource"
+
+    if not message:
+        message = (
+            default_message
+            or f"Something went wrong while contacting the FastAPI Cloud server. Please try again later. \n\n{error}"
+        )
+
+    return message
+
+
 @contextlib.contextmanager
 def handle_http_errors(
     progress: Progress,
-    message: Optional[str] = None,
+    default_message: Optional[str] = None,
 ) -> Generator[None, None, None]:
     try:
         yield
@@ -86,25 +126,7 @@ def handle_http_errors(
     except HTTPError as e:
         logger.debug(e)
 
-        # Handle validation errors from Pydantic models, this should make it easier to debug :)
-        if isinstance(e, HTTPStatusError) and e.response.status_code == 422:
-            logger.debug(e.response.json())  # pragma: no cover
-
-        if isinstance(e, HTTPStatusError) and e.response.status_code in (401, 403):
-            message = "The specified token is not valid. "
-
-            identity = Identity()
-
-            if identity.auth_mode == "user":
-                message += "Use `fastapi login` to generate a new token."
-            else:
-                message += "Make sure to use a valid token."
-
-        else:
-            message = (
-                message
-                or f"Something went wrong while contacting the FastAPI Cloud server. Please try again later. \n\n{e}"
-            )
+        message = handle_http_error(e, default_message)
 
         progress.set_error(message)
 

tests/test_auth.py

@@ -78,6 +78,11 @@ def test_is_jwt_expired_edge_case_one_second_before() -> None:
     assert not _is_jwt_expired(token)
 
 
+def test_is_expired_with_no_token(temp_auth_config: Path) -> None:
+    assert not temp_auth_config.exists()
+    assert Identity().is_expired()
+
+
 def test_is_logged_in_with_no_token(temp_auth_config: Path) -> None:
     assert not temp_auth_config.exists()
     assert not Identity().is_logged_in()

tests/test_cli_deploy.py

@@ -17,7 +17,7 @@
 from fastapi_cloud_cli.config import Settings
 from fastapi_cloud_cli.utils.api import StreamLogError, TooManyRetriesError
 from tests.conftest import ConfiguredApp
-from tests.utils import Keys, build_logs_response, changing_dir
+from tests.utils import Keys, build_logs_response, changing_dir, create_jwt_token
 
 runner = CliRunner()
 
@@ -216,6 +216,24 @@ def test_shows_waitlist_form_when_not_logged_in_longer_flow(
     assert "Let's go! Thanks for your interest in FastAPI Cloud! 🚀" in result.output
 
 
+def test_shows_login_prompt_when_token_is_expired(
+    temp_auth_config: Path, tmp_path: Path
+) -> None:
+    expired_token = create_jwt_token({"sub": "test_user", "exp": 0})
+    temp_auth_config.write_text(f'{{"access_token": "{expired_token}"}}')
+
+    with (
+        changing_dir(tmp_path),
+        patch("rich_toolkit.container.getchar") as mock_getchar,
+    ):
+        mock_getchar.side_effect = [Keys.CTRL_C]
+        result = runner.invoke(app, ["deploy"])
+
+    assert "Welcome to FastAPI Cloud!" in result.output
+    assert "Your session has expired. Please log in again." in result.output
+    assert "What would you like to do?" in result.output
+
+
 @pytest.mark.respx
 def test_shows_error_when_trying_to_get_teams(
     logged_in_cli: None, tmp_path: Path, respx_mock: respx.MockRouter

tests/test_cli_login.py

@@ -1,4 +1,3 @@
-import time
 from pathlib import Path
 from unittest.mock import patch
 
@@ -10,7 +9,6 @@
 
 from fastapi_cloud_cli.cli import app
 from fastapi_cloud_cli.config import Settings
-from tests.utils import create_jwt_token
 
 runner = CliRunner()
 
@@ -187,43 +185,3 @@ def test_notify_already_logged_in_user(
         "Run fastapi cloud logout first if you want to switch accounts."
         in result.output
     )
-
-
-@pytest.mark.respx
-def test_notify_expired_token_user(
-    respx_mock: respx.MockRouter, temp_auth_config: Path, settings: Settings
-) -> None:
-    past_exp = int(time.time()) - 3600
-    expired_token = create_jwt_token({"sub": "test_user_12345", "exp": past_exp})
-
-    temp_auth_config.write_text(f'{{"access_token": "{expired_token}"}}')
-
-    with patch("fastapi_cloud_cli.commands.login.typer.launch") as mock_open:
-        respx_mock.post(
-            "/login/device/authorization", data={"client_id": settings.client_id}
-        ).mock(
-            return_value=Response(
-                200,
-                json={
-                    "verification_uri_complete": "http://test.com",
-                    "verification_uri": "http://test.com",
-                    "user_code": "1234",
-                    "device_code": "5678",
-                },
-            )
-        )
-        respx_mock.post(
-            "/login/device/token",
-            data={
-                "device_code": "5678",
-                "client_id": settings.client_id,
-                "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
-            },
-        ).mock(return_value=Response(200, json={"access_token": "new_token_1234"}))
-
-        result = runner.invoke(app, ["login"])
-
-        assert result.exit_code == 0
-        assert "Your session has expired. Logging in again..." in result.output
-        assert "Now you are logged in!" in result.output
-        assert mock_open.called