🐛 Fix FASTAPI_CLOUD_TOKEN always overrides user token (solution with Context)

src/fastapi_cloud_cli/cli.py

@@ -9,18 +9,30 @@
 from .commands.setup_ci import setup_ci
 from .commands.unlink import unlink
 from .commands.whoami import whoami
+from .context import ctx
 from .logging import setup_logging
 from .utils.sentry import init_sentry
 
 setup_logging()
 
+COMMANDS_USE_TOKEN = {"deploy"}
+
 app = typer.Typer(rich_markup_mode="rich")
 
 cloud_app = typer.Typer(
     rich_markup_mode="rich",
     help="Manage [bold]FastAPI[/bold] Cloud deployments. 🚀",
 )
 
+
+@cloud_app.callback()
+def cloud_callback(typer_ctx: typer.Context) -> None:
+    if typer_ctx.invoked_subcommand in COMMANDS_USE_TOKEN:
+        ctx.initialize(prefer_auth_mode="token")
+    else:
+        ctx.initialize()
+
+
 # TODO: use the app structure
 
 # Additional commands

src/fastapi_cloud_cli/commands/deploy.py

@@ -19,6 +19,7 @@
 from rich_toolkit.menu import Option
 
 from fastapi_cloud_cli.commands.login import login
+from fastapi_cloud_cli.context import ctx
 from fastapi_cloud_cli.utils.api import (
     SUCCESSFUL_STATUSES,
     APIClient,
@@ -27,7 +28,6 @@
     TooManyRetriesError,
 )
 from fastapi_cloud_cli.utils.apps import AppConfig, get_app_config, write_app_config
-from fastapi_cloud_cli.utils.auth import Identity
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors
 
 logger = logging.getLogger(__name__)
@@ -652,7 +652,11 @@ def deploy(
         "Deploy path: %s, skip_wait: %s, app_id: %s", path, skip_wait, provided_app_id
     )
 
-    identity = Identity()
+    # Duplicate context initialization here to make `fastapi deploy` command work
+    # (callback doesn't take effect in this case)
+    ctx.initialize(prefer_auth_mode="token")
+
+    identity = ctx.get_identity()
 
     with get_rich_toolkit() as toolkit:
         if not identity.is_logged_in():
@@ -688,6 +692,13 @@ def deploy(
                 _waitlist_form(toolkit)
                 raise typer.Exit(1)
 
+        if identity.auth_mode == "token":
+            toolkit.print(
+                "Using token from [bold blue]FASTAPI_CLOUD_TOKEN[/] environment variable",
+                tag="info",
+            )
+            toolkit.print_line()
+
         toolkit.print_title("Starting deployment", tag="FastAPI")
         toolkit.print_line()
 

src/fastapi_cloud_cli/commands/env.py

@@ -5,9 +5,9 @@
 import typer
 from pydantic import BaseModel
 
+from fastapi_cloud_cli.context import ctx
 from fastapi_cloud_cli.utils.api import APIClient
 from fastapi_cloud_cli.utils.apps import get_app_config
-from fastapi_cloud_cli.utils.auth import Identity
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors
 from fastapi_cloud_cli.utils.env import validate_environment_variable_name
 
@@ -70,7 +70,7 @@ def list(
     List the environment variables for the app.
     """
 
-    identity = Identity()
+    identity = ctx.get_identity()
 
     with get_rich_toolkit(minimal=True) as toolkit:
         if not identity.is_logged_in():
@@ -125,7 +125,7 @@ def delete(
     Delete an environment variable from the app.
     """
 
-    identity = Identity()
+    identity = ctx.get_identity()
 
     with get_rich_toolkit(minimal=True) as toolkit:
         if not identity.is_logged_in():
@@ -218,7 +218,7 @@ def set(
     Set an environment variable for the app.
     """
 
-    identity = Identity()
+    identity = ctx.get_identity()
 
     with get_rich_toolkit(minimal=True) as toolkit:
         if not identity.is_logged_in():

src/fastapi_cloud_cli/commands/link.py

@@ -5,9 +5,9 @@
 import typer
 from rich_toolkit.menu import Option
 
+from fastapi_cloud_cli.context import ctx
 from fastapi_cloud_cli.utils.api import APIClient
 from fastapi_cloud_cli.utils.apps import AppConfig, get_app_config, write_app_config
-from fastapi_cloud_cli.utils.auth import Identity
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors
 
 logger = logging.getLogger(__name__)
@@ -17,7 +17,7 @@ def link() -> Any:
     """
     Link a local directory to an existing FastAPI Cloud app.
     """
-    identity = Identity()
+    identity = ctx.get_identity()
 
     with get_rich_toolkit() as toolkit:
         if not identity.is_logged_in():

src/fastapi_cloud_cli/commands/login.py

@@ -7,8 +7,9 @@
 from pydantic import BaseModel
 
 from fastapi_cloud_cli.config import Settings
+from fastapi_cloud_cli.context import ctx
 from fastapi_cloud_cli.utils.api import APIClient
-from fastapi_cloud_cli.utils.auth import AuthConfig, Identity, write_auth_config
+from fastapi_cloud_cli.utils.auth import AuthConfig, write_auth_config
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors
 
 logger = logging.getLogger(__name__)
@@ -76,7 +77,12 @@ def login() -> Any:
     """
     Login to FastAPI Cloud. 🚀
     """
-    identity = Identity()
+
+    # Duplicate context initialization here to make `fastapi login` command work
+    # (callback doesn't take effect in this case)
+    ctx.initialize()
+
+    identity = ctx.get_identity()
 
     if identity.is_logged_in():
         with get_rich_toolkit(minimal=True) as toolkit:
@@ -87,6 +93,17 @@ def login() -> Any:
 
         return
 
+    if identity.deploy_token is not None:
+        with get_rich_toolkit() as toolkit:
+            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")
 

src/fastapi_cloud_cli/commands/logs.py

@@ -9,14 +9,14 @@
 from rich.markup import escape
 from rich_toolkit import RichToolkit
 
+from fastapi_cloud_cli.context import ctx
 from fastapi_cloud_cli.utils.api import (
     APIClient,
     AppLogEntry,
     StreamLogError,
     TooManyRetriesError,
 )
 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, handle_http_error
 
 logger = logging.getLogger(__name__)
@@ -144,7 +144,7 @@ def logs(
         fastapi cloud logs --no-follow          # Fetch recent logs and exit
         fastapi cloud logs --tail 50 --since 1h # Last 50 logs from the past hour
     """
-    identity = Identity()
+    identity = ctx.get_identity()
     with get_rich_toolkit(minimal=True) as toolkit:
         if not identity.is_logged_in():
             toolkit.print(

src/fastapi_cloud_cli/commands/setup_ci.py

@@ -7,9 +7,9 @@
 
 import typer
 
+from fastapi_cloud_cli.context import ctx
 from fastapi_cloud_cli.utils.api import APIClient
 from fastapi_cloud_cli.utils.apps import get_app_config
-from fastapi_cloud_cli.utils.auth import Identity
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors
 
 logger = logging.getLogger(__name__)
@@ -198,7 +198,7 @@ def setup_ci(
         fastapi cloud setup-ci --file ci.yml        # Writes workflow to .github/workflows/ci.yml
     """
 
-    identity = Identity()
+    identity = ctx.get_identity()
 
     with get_rich_toolkit() as toolkit:
         if not identity.is_logged_in():

src/fastapi_cloud_cli/commands/whoami.py

@@ -4,30 +4,32 @@
 from rich import print
 from rich_toolkit.progress import Progress
 
+from fastapi_cloud_cli.context import ctx
 from fastapi_cloud_cli.utils.api import APIClient
-from fastapi_cloud_cli.utils.auth import Identity
 from fastapi_cloud_cli.utils.cli import handle_http_errors
 
 logger = logging.getLogger(__name__)
 
 
 def whoami() -> Any:
-    identity = Identity()
-
-    if identity.auth_mode == "token":
-        print("⚡ [bold]Using API token from environment variable[/bold]")
-        return
+    identity = ctx.get_identity()
 
     if not identity.is_logged_in():
         print("No credentials found. Use [blue]`fastapi login`[/] to login.")
-        return
-
-    with APIClient() as client:
-        with Progress(title="⚡ Fetching profile", transient=True) as progress:
-            with handle_http_errors(progress, default_message=""):
-                response = client.get("/users/me")
-                response.raise_for_status()
-
-        data = response.json()
-
-        print(f"⚡ [bold]{data['email']}[/bold]")
+    else:
+        with APIClient() as client:
+            with Progress(title="⚡ Fetching profile", transient=True) as progress:
+                with handle_http_errors(progress, default_message=""):
+                    response = client.get("/users/me")
+                    response.raise_for_status()
+
+            data = response.json()
+
+            print(f"⚡ [bold]{data['email']}[/bold]")
+
+    # Deployment token status
+    if identity.deploy_token is not None:
+        print(
+            "⚡ [bold]Using API token from environment variable for "
+            "[blue]`fastapi deploy`[/blue] command.[/bold]"
+        )

src/fastapi_cloud_cli/context.py

@@ -0,0 +1,24 @@
+import logging
+from typing import Literal
+
+from fastapi_cloud_cli.utils.auth import Identity
+
+logger = logging.getLogger(__name__)
+
+
+class Context:
+    def __init__(self) -> None:
+        self._is_initialized = False
+
+    def initialize(self, prefer_auth_mode: Literal["token", "user"] = "user") -> None:
+        logger.debug("Initializing context with prefer_auth_mode: %s", prefer_auth_mode)
+        self.prefer_auth_mode = prefer_auth_mode
+        self._is_initialized = True
+
+    def get_identity(self) -> Identity:
+        if not self._is_initialized:  # pragma: no cover
+            raise RuntimeError("Context must be initialized before use")
+        return Identity(prefer_auth_mode=self.prefer_auth_mode)
+
+
+ctx = Context()

src/fastapi_cloud_cli/utils/api.py

@@ -18,8 +18,7 @@
 
 from fastapi_cloud_cli import __version__
 from fastapi_cloud_cli.config import Settings
-
-from .auth import Identity
+from fastapi_cloud_cli.context import ctx
 
 logger = logging.getLogger(__name__)
 
@@ -193,7 +192,7 @@ def to_human_readable(cls, status: "DeploymentStatus") -> str:
 class APIClient(httpx.Client):
     def __init__(self) -> None:
         settings = Settings.get()
-        identity = Identity()
+        identity = ctx.get_identity()
 
         super().__init__(
             base_url=settings.base_api_url,

src/fastapi_cloud_cli/utils/auth.py

@@ -109,34 +109,57 @@ def _is_jwt_expired(token: str) -> bool:
 
 
 class Identity:
-    auth_mode: Literal["token", "user"]
-
-    def __init__(self) -> None:
-        self.token = _get_auth_token()
-        self.auth_mode = "user"
+    def __init__(self, prefer_auth_mode: Literal["token", "user"] = "user") -> None:
+        self._user_token = _get_auth_token()
+        self._auth_mode: Literal["token", "user"] = "user"
+        self._deploy_token: str | None = None
 
         # users using `FASTAPI_CLOUD_TOKEN`
         if env_token := self._get_token_from_env():
-            self.token = env_token
-            self.auth_mode = "token"
+            logger.debug("Reading token from FASTAPI_CLOUD_TOKEN environment variable")
+            self._deploy_token = env_token
+            if prefer_auth_mode == "token":
+                self._auth_mode = "token"
+                logger.debug("Using `token` auth mode")
 
     def _get_token_from_env(self) -> str | None:
         return os.environ.get("FASTAPI_CLOUD_TOKEN")
 
     def is_expired(self) -> bool:
-        if not self.token:
+        if self._auth_mode != "user":  # pragma: no cover  # Should never happen
+            raise RuntimeError("Expiration check is only applicable for user tokens")
+
+        if not self.user_token:
             return True
 
-        return _is_jwt_expired(self.token)
+        return _is_jwt_expired(self.user_token)
 
     def is_logged_in(self) -> bool:
         if self.token is None:
             logger.debug("Login status: False (no token)")
             return False
 
-        if self.auth_mode == "user" and self.is_expired():
+        if self._auth_mode == "user" and self.is_expired():
             logger.debug("Login status: False (token expired)")
             return False
 
         logger.debug("Login status: True")
         return True
+
+    @property
+    def auth_mode(self) -> Literal["token", "user"]:
+        return self._auth_mode
+
+    @property
+    def token(self) -> str | None:
+        if self._auth_mode == "token":
+            return self.deploy_token or self.user_token
+        return self.user_token
+
+    @property
+    def user_token(self) -> str | None:
+        return self._user_token
+
+    @property
+    def deploy_token(self) -> str | None:
+        return self._deploy_token