🎨 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):
@@ -34,7 +32,9 @@ def _get_tag_segments(
             tag_segments, left_padding = super()._get_tag_segments(
                 metadata, is_animated, done, animation_status=animation_status
             )
+
             tag_style = metadata.get("tag_style")
+
             if isinstance(tag_style, (str, Style)):
                 style = self.console.get_style(tag_style)
                 tag_segments = [
@@ -71,83 +71,42 @@ 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
-        self._version_check: BackgroundVersionCheck | None = None
-        self._print_update_on_exit = False
-
-    def __enter__(self) -> "FastAPIRichToolkit":
+        super().__init__(style=style, theme=theme)
         self._version_check = self._get_version_check()
 
-        if self._print_spacing:
-            self.console.print()
-        return self
-
     def __exit__(
         self,
         exc_type: type[BaseException] | None,
         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()
+        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
-
-        return None
+        return super().__exit__(
+            exc_type,
+            exc_value,
+            traceback,
+        )
 
     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
 
     def _print_update_message(self) -> None:
         if self._version_check is None:
             return
 
-        message = self._version_check.get_update_message()
-        if message:
+        if message := self._version_check.get_update_message():
             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 +125,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

@@ -3,6 +3,7 @@
 import re
 import threading
 from collections.abc import Callable
+from contextlib import suppress
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
@@ -176,11 +177,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
@@ -211,29 +208,22 @@ def __init__(
     ) -> None:
         self._check_for_update = check_for_update
         self._join_timeout = join_timeout
-        self._thread: threading.Thread | None = None
+        self._thread = threading.Thread(target=self._run, daemon=True)
         self._update: VersionUpdate | None = None
         self._message_returned = False
 
     def start(self) -> None:
-        self._thread = threading.Thread(target=self._run, daemon=True)
         self._thread.start()
 
     def _run(self) -> None:
-        try:
+        with suppress(Exception):
             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
 
     def get_update_message(self) -> str | None:
         if self._message_returned:
             return None
 
-        if self._thread:
-            self._thread.join(timeout=self._join_timeout)
+        self._thread.join(timeout=self._join_timeout)
 
         if self._update:
             self._message_returned = True

tests/test_version_check.py

@@ -1,4 +1,3 @@
-import inspect
 import json
 import os
 from datetime import datetime, timedelta, timezone
@@ -37,10 +36,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 +220,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 +248,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 +304,20 @@ def test_background_check_returns_no_message_without_update() -> None:
     message = check.get_update_message()
 
     assert message is None
+
+
+def test_background_check_hides_unexpected_errors(
+    recwarn: pytest.WarningsRecorder,
+) -> None:
+    def broken_check() -> None:
+        raise RuntimeError("version check failed unexpectedly")
+
+    check = BackgroundVersionCheck(check_for_update=broken_check, join_timeout=1)
+
+    check.start()
+
+    assert check.get_update_message() is None
+    assert not any(
+        warning.category is pytest.PytestUnhandledThreadExceptionWarning
+        for warning in recwarn
+    )