🎨 Auto format

Author: github-actions[bot]

src/fastapi_cloud_cli/commands/login.py

@@ -77,48 +77,50 @@ def login() -> Any:
     Login to FastAPI Cloud. 🚀
     """
     identity = Identity()
+    is_logged_in = identity.is_logged_in()
 
-    if identity.is_logged_in():
-        with get_rich_toolkit(minimal=True) as toolkit:
+    with get_rich_toolkit(minimal=is_logged_in) as toolkit:
+        if is_logged_in:
             toolkit.print("You are already logged in.")
             toolkit.print(
                 "Run [bold]fastapi cloud logout[/bold] first if you want to switch accounts."
             )
 
-        return
+            return
 
-    if identity.has_deploy_token():
-        with get_rich_toolkit() as toolkit:
+        if identity.has_deploy_token():
             toolkit.print(
                 "You have [bold blue]FASTAPI_CLOUD_TOKEN[/] environment variable set.\n"
                 "This token will take precedence over the user token for "
                 "[blue]`fastapi deploy`[/] command.",
                 tag="Warning",
             )
 
-    with get_rich_toolkit() as toolkit, APIClient() as client:
-        toolkit.print_title("Login to FastAPI Cloud", tag="FastAPI")
+        with APIClient() as client:
+            toolkit.print_title("Login to FastAPI Cloud", tag="FastAPI")
 
-        toolkit.print_line()
+            toolkit.print_line()
 
-        with toolkit.progress("Starting authorization") as progress:
-            with client.handle_http_errors(progress):
-                authorization_data = _start_device_authorization(client)
+            with toolkit.progress("Starting authorization") as progress:
+                with client.handle_http_errors(progress):
+                    authorization_data = _start_device_authorization(client)
 
-            url = authorization_data.verification_uri_complete
+                url = authorization_data.verification_uri_complete
 
-            progress.log(f"Opening [link={url}]{url}[/link]")
+                progress.log(f"Opening [link={url}]{url}[/link]")
 
-        toolkit.print_line()
+            toolkit.print_line()
 
-        with toolkit.progress("Waiting for user to authorize...") as progress:
-            typer.launch(url)
+            with toolkit.progress("Waiting for user to authorize...") as progress:
+                typer.launch(url)
 
-            with client.handle_http_errors(progress):
-                access_token = _fetch_access_token(
-                    client, authorization_data.device_code, authorization_data.interval
-                )
+                with client.handle_http_errors(progress):
+                    access_token = _fetch_access_token(
+                        client,
+                        authorization_data.device_code,
+                        authorization_data.interval,
+                    )
 
-            write_auth_config(AuthConfig(access_token=access_token))
+                write_auth_config(AuthConfig(access_token=access_token))
 
-            progress.log("Now you are logged in! 🚀")
+                progress.log("Now you are logged in! 🚀")

src/fastapi_cloud_cli/commands/whoami.py

@@ -13,7 +13,9 @@ def whoami() -> Any:
 
     with get_rich_toolkit(minimal=True) as toolkit:
         if not identity.is_logged_in():
-            toolkit.print("No credentials found. Use [blue]`fastapi login`[/] to login.")
+            toolkit.print(
+                "No credentials found. Use [blue]`fastapi login`[/] to login."
+            )
         else:
             with APIClient() as client:
                 with toolkit.progress(

src/fastapi_cloud_cli/utils/cli.py

@@ -3,7 +3,6 @@
 from types import TracebackType
 from typing import Any, Literal
 
-import click
 from rich.segment import Segment
 from rich.style import Style
 from rich.text import Text
@@ -16,7 +15,6 @@
 )
 
 logger = logging.getLogger(__name__)
-VERSION_CHECK_CONTEXT_KEY = "fastapi_cloud_cli.version_check"
 
 
 class FastAPIStyle(TaggedStyle):
@@ -71,23 +69,15 @@ def __init__(
         self,
         style: BaseStyle | None = None,
         theme: RichToolkitTheme | None = None,
-        handle_keyboard_interrupts: bool = True,
-        print_spacing: bool = True,
     ) -> None:
-        super().__init__(
-            style=style,
-            theme=theme,
-            handle_keyboard_interrupts=handle_keyboard_interrupts,
-        )
-        self._print_spacing = print_spacing
+        super().__init__(style=style, theme=theme)
+
         self._version_check: BackgroundVersionCheck | None = None
-        self._print_update_on_exit = False
 
     def __enter__(self) -> "FastAPIRichToolkit":
         self._version_check = self._get_version_check()
 
-        if self._print_spacing:
-            self.console.print()
+        super().__enter__()  # type: ignore[no-untyped-call]
         return self
 
     def __exit__(
@@ -96,40 +86,21 @@ def __exit__(
         exc_value: BaseException | None,
         traceback: TracebackType | None,
     ) -> bool | None:
-        is_keyboard_interrupt = exc_type is KeyboardInterrupt
-
-        if is_keyboard_interrupt and self._version_check is not None:
-            self._version_check.suppress()
-        elif self._print_update_on_exit:
-            self._print_update_message()
-
-        if self._print_spacing and not is_keyboard_interrupt:
-            self.console.print()
-
-        if self.handle_keyboard_interrupts and is_keyboard_interrupt:
-            return True
+        self._print_update_message()
+        should_suppress_exception = super().__exit__(
+            exc_type,
+            exc_value,
+            traceback,
+        )
 
-        return None
+        return True if should_suppress_exception else None
 
     def _get_version_check(self) -> BackgroundVersionCheck | None:
         if os.environ.get(DISABLE_VERSION_CHECK_ENV) == "1":
             return None
 
-        context = click.get_current_context(silent=True)
-        if context is None:
-            version_check = BackgroundVersionCheck()
-            version_check.start()
-            self._print_update_on_exit = True
-            return version_check
-
-        stored_version_check = context.meta.get(VERSION_CHECK_CONTEXT_KEY)
-        if isinstance(stored_version_check, BackgroundVersionCheck):
-            return stored_version_check
-
         version_check = BackgroundVersionCheck()
         version_check.start()
-        context.meta[VERSION_CHECK_CONTEXT_KEY] = version_check
-        context.call_on_close(self._print_update_message)
 
         return version_check
 
@@ -142,12 +113,7 @@ def _print_update_message(self) -> None:
             self.print(Text.from_markup(message), tag="update", tag_style="tag.update")
 
 
-def get_rich_toolkit(
-    minimal: bool = False,
-    *,
-    print_spacing: bool = True,
-    handle_keyboard_interrupts: bool = True,
-) -> RichToolkit:
+def get_rich_toolkit(minimal: bool = False) -> RichToolkit:
     style = MinimalStyle() if minimal else FastAPIStyle(tag_width=11)
 
     theme = RichToolkitTheme(
@@ -166,8 +132,4 @@ def get_rich_toolkit(
         },
     )
 
-    return FastAPIRichToolkit(
-        theme=theme,
-        handle_keyboard_interrupts=handle_keyboard_interrupts,
-        print_spacing=print_spacing,
-    )
+    return FastAPIRichToolkit(theme=theme)

src/fastapi_cloud_cli/utils/version_check.py

@@ -176,11 +176,7 @@ def get_upgrade_command(
     *,
     detector: Callable[[str], InstallerInfo | None] = detect_installer,
 ) -> str:
-    try:
-        installer_info = detector(PACKAGE_NAME)
-    except Exception as error:
-        logger.debug("Could not detect CLI installer: %s", error)
-        return DEFAULT_UPGRADE_COMMAND
+    installer_info = detector(PACKAGE_NAME)
 
     if installer_info is None or installer_info.upgrade_cmd is None:
         return DEFAULT_UPGRADE_COMMAND
@@ -220,13 +216,7 @@ def start(self) -> None:
         self._thread.start()
 
     def _run(self) -> None:
-        try:
-            self._update = self._check_for_update()
-        except Exception as error:
-            logger.debug("Could not check latest CLI version: %s", error)
-
-    def suppress(self) -> None:
-        self._message_returned = True
+        self._update = self._check_for_update()
 
     def get_update_message(self) -> str | None:
         if self._message_returned:

tests/test_version_check.py

@@ -1,6 +1,6 @@
-import inspect
 import json
 import os
+import threading
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from types import SimpleNamespace
@@ -37,10 +37,6 @@ def test_is_newer_version() -> None:
     assert not is_newer_version("not-a-version", "0.17.1")
 
 
-def test_get_rich_toolkit_has_no_invocation_args_override() -> None:
-    assert "args" not in inspect.signature(get_rich_toolkit).parameters
-
-
 @respx.mock
 def test_check_for_update_returns_update_when_pypi_has_newer_version() -> None:
     route = respx.get(PYPI_JSON_URL).mock(
@@ -225,6 +221,14 @@ def test_get_upgrade_command_falls_back_when_installer_is_unknown() -> None:
     )
 
 
+def test_get_upgrade_command_propagates_detector_errors() -> None:
+    def broken_detector(package_name: str) -> None:
+        raise RuntimeError(f"Could not inspect {package_name}")
+
+    with pytest.raises(RuntimeError, match="Could not inspect fastapi-cloud-cli"):
+        get_upgrade_command(detector=broken_detector)
+
+
 def test_format_update_message() -> None:
     message = format_update_message(
         VersionUpdate(current="0.17.1", latest="0.18.0"),
@@ -245,44 +249,23 @@ def _seed_fresh_update_cache(latest_version: str = "999.0.0") -> None:
     )
 
 
-def test_get_rich_toolkit_prints_forced_update_message_without_wrapper(
-    monkeypatch: pytest.MonkeyPatch,
-    capsys: pytest.CaptureFixture[str],
-) -> None:
-    monkeypatch.setenv("FASTAPI_CLOUD_DISABLE_VERSION_CHECK", "")
-    _seed_fresh_update_cache()
-
-    with get_rich_toolkit() as toolkit:
-        toolkit.print("command output")
-
-    output = capsys.readouterr().out
-    assert "command output" in output
-    assert "update" in output
-    assert "A newer FastAPI Cloud CLI version is available" in output
-
-
-def test_get_rich_toolkit_prints_forced_update_once_per_click_command() -> None:
+def test_get_rich_toolkit_prints_update_after_toolkit_output() -> None:
     @click.command()
-    def command_with_multiple_toolkits() -> None:
-        with get_rich_toolkit() as toolkit:
-            toolkit.print("first output")
+    def command_with_toolkit() -> None:
         with get_rich_toolkit() as toolkit:
-            toolkit.print("second output")
-        click.echo("after command body")
+            toolkit.print("command output")
 
     _seed_fresh_update_cache()
     result = CliRunner().invoke(
-        command_with_multiple_toolkits,
+        command_with_toolkit,
         [],
         env={"FASTAPI_CLOUD_DISABLE_VERSION_CHECK": ""},
     )
 
     assert result.exit_code == 0, result.output
-    assert "first output" in result.output
-    assert "second output" in result.output
-    assert "after command body" in result.output
+    assert "command output" in result.output
     assert result.output.count("A newer FastAPI Cloud CLI version is available") == 1
-    assert result.output.rfind("after command body") < result.output.rfind("A newer")
+    assert result.output.rfind("command output") < result.output.rfind("A newer")
 
 
 def test_get_rich_toolkit_skips_update_when_disabled(
@@ -322,3 +305,27 @@ def test_background_check_returns_no_message_without_update() -> None:
     message = check.get_update_message()
 
     assert message is None
+
+
+def test_background_check_exposes_unexpected_errors() -> None:
+    captured_errors: list[threading.ExceptHookArgs] = []
+    original_hook = threading.excepthook
+
+    def capture_thread_error(args: threading.ExceptHookArgs) -> None:
+        captured_errors.append(args)
+
+    def broken_check() -> None:
+        raise RuntimeError("version check failed unexpectedly")
+
+    threading.excepthook = capture_thread_error
+    try:
+        check = BackgroundVersionCheck(check_for_update=broken_check, join_timeout=1)
+        check.start()
+        assert check._thread is not None
+        check._thread.join(timeout=1)
+    finally:
+        threading.excepthook = original_hook
+
+    assert captured_errors
+    assert captured_errors[0].exc_type is RuntimeError
+    assert str(captured_errors[0].exc_value) == "version check failed unexpectedly"