🎨 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/config.py

@@ -25,3 +25,7 @@ def get_cli_config_path() -> Path:
     cli_config_path.parent.mkdir(parents=True, exist_ok=True)
 
     return cli_config_path
+
+
+def get_version_check_cache_path() -> Path:
+    return get_config_folder() / "version-check.json"

src/fastapi_cloud_cli/utils/version_check.py

@@ -2,27 +2,26 @@
 import logging
 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
 
 import httpx
-from detect_installer import InstallerInfo, detect_installer
+from detect_installer import detect_installer
 
 from fastapi_cloud_cli import __version__
-from fastapi_cloud_cli.utils.config import get_config_folder
+from fastapi_cloud_cli.utils.config import get_version_check_cache_path
 
 logger = logging.getLogger(__name__)
 
 PACKAGE_NAME = "fastapi-cloud-cli"
 DEFAULT_UPGRADE_COMMAND = f"pip install --upgrade {PACKAGE_NAME}"
-PYPI_JSON_URL = "https://pypi.org/pypi/fastapi-cloud-cli/json"
+PYPI_JSON_URL = f"https://pypi.org/pypi/{PACKAGE_NAME}/json"
 VERSION_CHECK_TIMEOUT_SECONDS = 2.0
 VERSION_CHECK_JOIN_TIMEOUT_SECONDS = 0.2
 VERSION_CHECK_CACHE_TTL = timedelta(hours=24)
 DISABLE_VERSION_CHECK_ENV = "FASTAPI_CLOUD_DISABLE_VERSION_CHECK"
-VERSION_CHECK_CACHE_FILENAME = "version-check.json"
 SIMPLE_RELEASE_VERSION_RE = re.compile(r"\d+(?:\.\d+)*")
 
 
@@ -32,10 +31,6 @@ class VersionUpdate:
     latest: str
 
 
-def get_version_check_cache_path() -> Path:
-    return get_config_folder() / VERSION_CHECK_CACHE_FILENAME
-
-
 def _parse_simple_release_version(version: str) -> tuple[int, ...] | None:
     if not SIMPLE_RELEASE_VERSION_RE.fullmatch(version):
         logger.debug("Skipping non-simple version string: %r", version)
@@ -51,10 +46,6 @@ def is_newer_version(latest: str, current: str) -> bool:
     if latest_parts is None or current_parts is None:
         return False
 
-    version_length = max(len(latest_parts), len(current_parts))
-    latest_parts += (0,) * (version_length - len(latest_parts))
-    current_parts += (0,) * (version_length - len(current_parts))
-
     return latest_parts > current_parts
 
 
@@ -117,11 +108,8 @@ def write_latest_version_cache(
         logger.debug("Could not write CLI version cache: %s", error)
 
 
-def fetch_latest_version(
-    *,
-    current_version: str = __version__,
-) -> str | None:
-    headers = {"User-Agent": f"fastapi-cloud-cli/{current_version}"}
+def fetch_latest_version() -> str | None:
+    headers = {"User-Agent": f"fastapi-cloud-cli/{__version__}"}
 
     try:
         with httpx.Client(
@@ -147,16 +135,14 @@ def fetch_latest_version(
 
 def check_for_update(
     *,
-    current_version: str = __version__,
     cache_path: Path | None = None,
-    now: datetime | None = None,
 ) -> VersionUpdate | None:
-    now = now or datetime.now(timezone.utc)
+    now = datetime.now(timezone.utc)
     cache_path = cache_path or get_version_check_cache_path()
 
     latest_version = read_cached_latest_version(cache_path, now=now)
     if latest_version is None:
-        latest_version = fetch_latest_version(current_version=current_version)
+        latest_version = fetch_latest_version()
         if latest_version is None:
             return None
 
@@ -166,21 +152,14 @@ def check_for_update(
             now=now,
         )
 
-    if not is_newer_version(latest_version, current_version):
+    if not is_newer_version(latest_version, __version__):
         return None
 
-    return VersionUpdate(current=current_version, latest=latest_version)
+    return VersionUpdate(current=__version__, latest=latest_version)
 
 
-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
+def get_upgrade_command() -> str:
+    installer_info = detect_installer(PACKAGE_NAME)
 
     if installer_info is None or installer_info.upgrade_cmd is None:
         return DEFAULT_UPGRADE_COMMAND
@@ -190,50 +169,32 @@ def get_upgrade_command(
 
 def format_update_message(
     update: VersionUpdate,
-    *,
-    upgrade_command: str | None = None,
 ) -> str:
-    upgrade_command = upgrade_command or get_upgrade_command()
-
     return (
         "A newer FastAPI Cloud CLI version is available: "
         f"{update.current} → [bold]{update.latest}[/]\n\n"
-        f'Run "[blue]{upgrade_command}[/]" to upgrade.'
+        f'Run "[blue]{get_upgrade_command()}[/]" to upgrade.'
     )
 
 
 class BackgroundVersionCheck:
-    def __init__(
-        self,
-        *,
-        check_for_update: Callable[[], VersionUpdate | None] = check_for_update,
-        join_timeout: float = VERSION_CHECK_JOIN_TIMEOUT_SECONDS,
-    ) -> None:
-        self._check_for_update = check_for_update
-        self._join_timeout = join_timeout
-        self._thread: threading.Thread | None = None
+    def __init__(self) -> 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:
-            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
+        with suppress(Exception):
+            self._update = check_for_update()
 
     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=VERSION_CHECK_JOIN_TIMEOUT_SECONDS)
 
         if self._update:
             self._message_returned = True