🐛 Fix FASTAPI_CLOUD_TOKEN always overrides user token (#180)

Co-authored-by: Patrick Arminio <patrick.arminio@gmail.com>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: 3 people

src/fastapi_cloud_cli/commands/deploy.py

@@ -8,7 +8,7 @@
 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.cli import get_rich_toolkit
 from fastapi_cloud_cli.utils.env import validate_environment_variable_name
 
 logger = logging.getLogger(__name__)
@@ -23,17 +23,17 @@ class EnvironmentVariableResponse(BaseModel):
     data: list[EnvironmentVariable]
 
 
-def _get_environment_variables(app_id: str) -> EnvironmentVariableResponse:
-    with APIClient() as client:
-        response = client.get(f"/apps/{app_id}/environment-variables/")
-        response.raise_for_status()
+def _get_environment_variables(
+    client: APIClient, app_id: str
+) -> EnvironmentVariableResponse:
+    response = client.get(f"/apps/{app_id}/environment-variables/")
+    response.raise_for_status()
 
-        return EnvironmentVariableResponse.model_validate(response.json())
+    return EnvironmentVariableResponse.model_validate(response.json())
 
 
-def _delete_environment_variable(app_id: str, name: str) -> bool:
-    with APIClient() as client:
-        response = client.delete(f"/apps/{app_id}/environment-variables/{name}")
+def _delete_environment_variable(client: APIClient, app_id: str, name: str) -> bool:
+    response = client.delete(f"/apps/{app_id}/environment-variables/{name}")
 
     if response.status_code == 404:
         return False
@@ -44,14 +44,13 @@ def _delete_environment_variable(app_id: str, name: str) -> bool:
 
 
 def _set_environment_variable(
-    app_id: str, name: str, value: str, is_secret: bool = False
+    client: APIClient, app_id: str, name: str, value: str, is_secret: bool = False
 ) -> None:
-    with APIClient() as client:
-        response = client.post(
-            f"/apps/{app_id}/environment-variables/",
-            json={"name": name, "value": value, "is_secret": is_secret},
-        )
-        response.raise_for_status()
+    response = client.post(
+        f"/apps/{app_id}/environment-variables/",
+        json={"name": name, "value": value, "is_secret": is_secret},
+    )
+    response.raise_for_status()
 
 
 env_app = typer.Typer()
@@ -91,11 +90,14 @@ def list(
             )
             raise typer.Exit(1)
 
-        with toolkit.progress(
-            "Fetching environment variables...", transient=True
-        ) as progress:
-            with handle_http_errors(progress):
-                environment_variables = _get_environment_variables(app_config.app_id)
+        with APIClient() as client:
+            with toolkit.progress(
+                "Fetching environment variables...", transient=True
+            ) as progress:
+                with client.handle_http_errors(progress):
+                    environment_variables = _get_environment_variables(
+                        client=client, app_id=app_config.app_id
+                    )
 
         if not environment_variables.data:
             toolkit.print("No environment variables found.")
@@ -146,42 +148,45 @@ def delete(
             )
             raise typer.Exit(1)
 
-        if not name:
-            with toolkit.progress(
-                "Fetching environment variables...", transient=True
-            ) as progress:
-                with handle_http_errors(progress):
-                    environment_variables = _get_environment_variables(
-                        app_config.app_id
-                    )
-
-            if not environment_variables.data:
-                toolkit.print("No environment variables found.")
-                return
-
-            name = toolkit.ask(
-                "Select the environment variable to delete:",
-                options=[
-                    {"name": env_var.name, "value": env_var.name}
-                    for env_var in environment_variables.data
-                ],
-            )
-
-            assert name
-        else:
-            if not validate_environment_variable_name(name):
-                toolkit.print(
-                    f"The environment variable name [bold]{name}[/] is invalid."
+        with APIClient() as client:
+            if not name:
+                with toolkit.progress(
+                    "Fetching environment variables...", transient=True
+                ) as progress:
+                    with client.handle_http_errors(progress):
+                        environment_variables = _get_environment_variables(
+                            client=client, app_id=app_config.app_id
+                        )
+
+                if not environment_variables.data:
+                    toolkit.print("No environment variables found.")
+                    return
+
+                name = toolkit.ask(
+                    "Select the environment variable to delete:",
+                    options=[
+                        {"name": env_var.name, "value": env_var.name}
+                        for env_var in environment_variables.data
+                    ],
                 )
-                raise typer.Exit(1)
 
-            toolkit.print_line()
+                assert name
+            else:
+                if not validate_environment_variable_name(name):
+                    toolkit.print(
+                        f"The environment variable name [bold]{name}[/] is invalid."
+                    )
+                    raise typer.Exit(1)
+
+                toolkit.print_line()
 
-        with toolkit.progress(
-            "Deleting environment variable", transient=True
-        ) as progress:
-            with handle_http_errors(progress):
-                deleted = _delete_environment_variable(app_config.app_id, name)
+            with toolkit.progress(
+                "Deleting environment variable", transient=True
+            ) as progress:
+                with client.handle_http_errors(progress):
+                    deleted = _delete_environment_variable(
+                        client=client, app_id=app_config.app_id, name=name
+                    )
 
         if not deleted:
             toolkit.print("Environment variable not found.")
@@ -253,14 +258,21 @@ def set(
             else:
                 value = toolkit.input("Enter the value of the environment variable:")
 
-        with toolkit.progress(
-            "Setting environment variable", transient=True
-        ) as progress:
-            assert name is not None
-            assert value is not None
-
-            with handle_http_errors(progress):
-                _set_environment_variable(app_config.app_id, name, value, secret)
+        with APIClient() as client:
+            with toolkit.progress(
+                "Setting environment variable", transient=True
+            ) as progress:
+                assert name is not None
+                assert value is not None
+
+                with client.handle_http_errors(progress):
+                    _set_environment_variable(
+                        client=client,
+                        app_id=app_config.app_id,
+                        name=name,
+                        value=value,
+                        is_secret=secret,
+                    )
 
         if secret:
             toolkit.print(f"Secret environment variable [bold]{name}[/] set.")

src/fastapi_cloud_cli/commands/env.py

@@ -8,7 +8,7 @@
 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
+from fastapi_cloud_cli.utils.cli import get_rich_toolkit
 
 logger = logging.getLogger(__name__)
 
@@ -47,40 +47,42 @@ def link() -> Any:
         toolkit.print_title("Link to FastAPI Cloud", tag="FastAPI")
         toolkit.print_line()
 
-        with toolkit.progress("Fetching teams...") as progress:
-            with handle_http_errors(
-                progress,
-                default_message="Error fetching teams. Please try again later.",
-            ):
-                with APIClient() as client:
+        with APIClient() as client:
+            with toolkit.progress("Fetching teams...") as progress:
+                with client.handle_http_errors(
+                    progress,
+                    default_message="Error fetching teams. Please try again later.",
+                ):
                     response = client.get("/teams/")
                     response.raise_for_status()
                     teams_data = response.json()["data"]
 
-        if not teams_data:
-            toolkit.print(
-                "[error]No teams found. Please create a team first.[/]",
-            )
-            raise typer.Exit(1)
+            if not teams_data:
+                toolkit.print(
+                    "[error]No teams found. Please create a team first.[/]",
+                )
+                raise typer.Exit(1)
 
-        toolkit.print_line()
+            toolkit.print_line()
 
-        team = toolkit.ask(
-            "Select the team:",
-            tag="team",
-            options=[
-                Option({"name": t["name"], "value": {"id": t["id"], "name": t["name"]}})
-                for t in teams_data
-            ],
-        )
+            team = toolkit.ask(
+                "Select the team:",
+                tag="team",
+                options=[
+                    Option(
+                        {"name": t["name"], "value": {"id": t["id"], "name": t["name"]}}
+                    )
+                    for t in teams_data
+                ],
+            )
 
-        toolkit.print_line()
+            toolkit.print_line()
 
-        with toolkit.progress("Fetching apps...") as progress:
-            with handle_http_errors(
-                progress, default_message="Error fetching apps. Please try again later."
-            ):
-                with APIClient() as client:
+            with toolkit.progress("Fetching apps...") as progress:
+                with client.handle_http_errors(
+                    progress,
+                    default_message="Error fetching apps. Please try again later.",
+                ):
                     response = client.get("/apps/", params={"team_id": team["id"]})
                     response.raise_for_status()
                     apps_data = response.json()["data"]

src/fastapi_cloud_cli/commands/link.py

@@ -9,7 +9,7 @@
 from fastapi_cloud_cli.config import Settings
 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.cli import get_rich_toolkit, handle_http_errors
+from fastapi_cloud_cli.utils.cli import get_rich_toolkit
 
 logger = logging.getLogger(__name__)
 
@@ -87,13 +87,22 @@ def login() -> Any:
 
         return
 
+    if identity.has_deploy_token():
+        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")
 
         toolkit.print_line()
 
         with toolkit.progress("Starting authorization") as progress:
-            with handle_http_errors(progress):
+            with client.handle_http_errors(progress):
                 authorization_data = _start_device_authorization(client)
 
             url = authorization_data.verification_uri_complete
@@ -105,7 +114,7 @@ def login() -> Any:
         with toolkit.progress("Waiting for user to authorize...") as progress:
             typer.launch(url)
 
-            with handle_http_errors(progress):
+            with client.handle_http_errors(progress):
                 access_token = _fetch_access_token(
                     client, authorization_data.device_code, authorization_data.interval
                 )

src/fastapi_cloud_cli/commands/login.py

@@ -14,10 +14,11 @@
     AppLogEntry,
     StreamLogError,
     TooManyRetriesError,
+    handle_http_error,
 )
 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
+from fastapi_cloud_cli.utils.cli import get_rich_toolkit
 
 logger = logging.getLogger(__name__)
 

src/fastapi_cloud_cli/commands/logs.py

@@ -10,7 +10,7 @@
 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.cli import get_rich_toolkit
 
 logger = logging.getLogger(__name__)
 
@@ -96,19 +96,18 @@ def _set_github_secret(name: str, value: str) -> None:
         raise GitHubSecretError(f"Failed to set GitHub secret '{name}'") from e
 
 
-def _create_token(app_id: str, token_name: str) -> dict[str, str]:
+def _create_token(client: APIClient, app_id: str, token_name: str) -> dict[str, str]:
     """Create a new deploy token.
 
     Returns token_data dict with 'value' and 'expired_at' keys.
     """
-    with APIClient() as client:
-        response = client.post(
-            f"/apps/{app_id}/tokens",
-            json={"name": token_name, "expires_in_days": TOKEN_EXPIRES_DAYS},
-        )
-        response.raise_for_status()
-        data = response.json()
-        return {"value": data["value"], "expired_at": data["expired_at"]}
+    response = client.post(
+        f"/apps/{app_id}/tokens",
+        json={"name": token_name, "expires_in_days": TOKEN_EXPIRES_DAYS},
+    )
+    response.raise_for_status()
+    data = response.json()
+    return {"value": data["value"], "expired_at": data["expired_at"]}
 
 
 def _get_default_branch() -> str:
@@ -285,12 +284,15 @@ def setup_ci(
         token_name = f"GitHub Actions — {repo_slug} ({timestamp})"
 
         with (
+            APIClient() as client,
             toolkit.progress(title="Generating deploy token...") as progress,
-            handle_http_errors(
+            client.handle_http_errors(
                 progress, default_message="Error creating deploy token."
             ),
         ):
-            token_data = _create_token(app_config.app_id, token_name)
+            token_data = _create_token(
+                client=client, app_id=app_config.app_id, token_name=token_name
+            )
             progress.log(msg_token)
 
         toolkit.print_line()