♻️ Use batch endpoint for environment variable updates and deletions

src/fastapi_cloud_cli/api/__init__.py

@@ -16,5 +16,9 @@
 from ._models import CustomDomainStatus as CustomDomainStatus
 from ._models import Deployment as Deployment
 from ._models import DeploymentStatus as DeploymentStatus
+from ._models import EnvironmentVariable as EnvironmentVariable
+from ._models import (
+    EnvironmentVariableCreatePayload as EnvironmentVariableCreatePayload,
+)
 from ._retry import STREAM_LOGS_MAX_RETRIES as STREAM_LOGS_MAX_RETRIES
 from .client import APIClient as APIClient

src/fastapi_cloud_cli/api/_models.py

@@ -4,6 +4,22 @@
 from pydantic import BaseModel, Field, TypeAdapter
 
 
+class EnvironmentVariable(BaseModel):
+    name: str
+    value: str | None = None
+    is_secret: bool = False
+    updated_at: str | None = None
+
+
+class EnvironmentVariableResponse(BaseModel):
+    data: list[EnvironmentVariable]
+
+
+class EnvironmentVariableCreatePayload(BaseModel):
+    value: str
+    is_secret: bool = False
+
+
 class AppLogEntry(BaseModel):
     timestamp: str
     message: str

src/fastapi_cloud_cli/api/client.py

@@ -32,6 +32,8 @@
     CustomDomainsAPIResponse,
     Deployment,
     DeploymentStatus,
+    EnvironmentVariableCreatePayload,
+    EnvironmentVariableResponse,
 )
 from ._retry import (
     STREAM_LOGS_MAX_RETRIES,
@@ -142,6 +144,34 @@ def get_deployment(self, deployment_id: str) -> Deployment:
         response.raise_for_status()
         return Deployment.model_validate(response.json())
 
+    def get_environment_variables(self, *, app_id: str) -> EnvironmentVariableResponse:
+        response = self.get(f"/apps/{app_id}/environment-variables/")
+        response.raise_for_status()
+
+        return EnvironmentVariableResponse.model_validate(response.json())
+
+    def batch_environment_variables(
+        self,
+        *,
+        app_id: str,
+        upsert: dict[str, EnvironmentVariableCreatePayload],
+        delete: list[str],
+        redeploy: bool = True,
+    ) -> EnvironmentVariableResponse:
+        response = self.put(
+            f"/apps/{app_id}/environment-variables/",
+            json={
+                "upsert": {
+                    name: payload.model_dump() for name, payload in upsert.items()
+                },
+                "delete": delete,
+                "redeploy": redeploy,
+            },
+        )
+        response.raise_for_status()
+
+        return EnvironmentVariableResponse.model_validate(response.json())
+
     def get_custom_domains(self, *, app_id: str) -> CustomDomainsAPIResponse:
         response = self.get(f"/apps/{app_id}/custom-domains")
         response.raise_for_status()

src/fastapi_cloud_cli/commands/env/_shared.py

@@ -1,31 +1,10 @@
-from pydantic import BaseModel
 from rich.text import Text
 
-from fastapi_cloud_cli.api import APIClient
+from fastapi_cloud_cli.api import EnvironmentVariable
 
 ENV_VAR_VALUE_MAX_LENGTH = 40
 
 
-class EnvironmentVariable(BaseModel):
-    name: str
-    value: str | None = None
-    is_secret: bool = False
-    updated_at: str | None = None
-
-
-class EnvironmentVariableResponse(BaseModel):
-    data: list[EnvironmentVariable]
-
-
-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())
-
-
 def _find_environment_variable(
     environment_variables: list[EnvironmentVariable], name: str
 ) -> EnvironmentVariable | None:

src/fastapi_cloud_cli/commands/env/delete.py

@@ -9,7 +9,6 @@
 from fastapi_cloud_cli.api import APIClient
 from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context
 from fastapi_cloud_cli.commands.env._app import env_app
-from fastapi_cloud_cli.commands.env._shared import _get_environment_variables
 from fastapi_cloud_cli.utils.apps import resolve_app_id_or_fail
 from fastapi_cloud_cli.utils.env import validate_environment_variable_name
 from fastapi_cloud_cli.utils.execution import JsonOutputOption
@@ -32,17 +31,6 @@ def _render_environment_variable_delete_output(
     toolkit.print(f"Environment variable [bold]{data.name}[/] deleted.", bullet=False)
 
 
-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
-
-    response.raise_for_status()
-
-    return True
-
-
 @env_app.command(cls=UserCommand)
 def delete(
     ctx: typer.Context,
@@ -84,10 +72,19 @@ def delete(
             help="Confirm deletion without prompting.",
         ),
     ] = False,
+    no_redeploy: Annotated[
+        bool,
+        typer.Option(
+            "--no-redeploy",
+            help="Delete the environment variable without redeploying the app.",
+        ),
+    ] = False,
     json_output: JsonOutputOption = False,
 ) -> Any:
     """
-    Delete an environment variable from the app.
+    Delete an environment variable and redeploy the app by default.
+
+    Succeeds even if the variable is already absent.
     """
 
     toolkit = get_user_command_context(ctx).toolkit
@@ -109,9 +106,9 @@ def delete(
             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=target_app_id
+                with client.handle_http_errors(progress, toolkit=toolkit):
+                    environment_variables = client.get_environment_variables(
+                        app_id=target_app_id
                     )
 
             toolkit.print_title("environment variables")
@@ -163,23 +160,14 @@ def delete(
         with toolkit.progress(
             "Deleting environment variable", transient=True
         ) as progress:
-            with client.handle_http_errors(progress):
-                deleted = _delete_environment_variable(
-                    client=client, app_id=target_app_id, name=name
+            with client.handle_http_errors(progress, toolkit=toolkit):
+                client.batch_environment_variables(
+                    app_id=target_app_id,
+                    upsert={},
+                    delete=[name],
+                    redeploy=not no_redeploy,
                 )
 
-    if not deleted:
-        message = (
-            f"Environment variable {name} not found."
-            if toolkit.mode == "json"
-            else "Environment variable not found."
-        )
-        toolkit.fail(
-            "not_found",
-            message,
-            hint="Run `fastapi cloud env list` to see available variables.",
-        )
-
     toolkit.success(
         EnvironmentVariableDeleteOutput(
             app_id=target_app_id, name=name, show_tag=name_provided

src/fastapi_cloud_cli/commands/env/get.py

@@ -7,14 +7,12 @@
 from rich_toolkit import RichToolkit
 from rich_toolkit.menu import Option
 
-from fastapi_cloud_cli.api import APIClient
+from fastapi_cloud_cli.api import APIClient, EnvironmentVariable
 from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context
 from fastapi_cloud_cli.commands.env._app import env_app
 from fastapi_cloud_cli.commands.env._shared import (
-    EnvironmentVariable,
     _find_environment_variable,
     _format_env_var_value,
-    _get_environment_variables,
 )
 from fastapi_cloud_cli.utils.apps import resolve_app_id_or_fail
 from fastapi_cloud_cli.utils.execution import JsonOutputOption
@@ -92,8 +90,8 @@ def get_variable(
             "Fetching environment variables...", transient=True
         ) as progress:
             with client.handle_http_errors(progress):
-                environment_variables = _get_environment_variables(
-                    client=client, app_id=target_app_id
+                environment_variables = client.get_environment_variables(
+                    app_id=target_app_id
                 )
 
     if name is None:

src/fastapi_cloud_cli/commands/env/list.py

@@ -7,14 +7,12 @@
 from rich.text import Text
 from rich_toolkit import RichToolkit
 
-from fastapi_cloud_cli.api import APIClient
+from fastapi_cloud_cli.api import APIClient, EnvironmentVariable
 from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context
 from fastapi_cloud_cli.commands.env._app import env_app
 from fastapi_cloud_cli.commands.env._shared import (
     ENV_VAR_VALUE_MAX_LENGTH,
-    EnvironmentVariable,
     _format_env_var_value,
-    _get_environment_variables,
 )
 from fastapi_cloud_cli.utils.apps import resolve_app_id_or_fail
 from fastapi_cloud_cli.utils.dates import format_last_updated
@@ -94,8 +92,8 @@ def list_variables(
             "Fetching environment variables...", transient=True
         ) as progress:
             with client.handle_http_errors(progress):
-                environment_variables = _get_environment_variables(
-                    client=client, app_id=target_app_id
+                environment_variables = client.get_environment_variables(
+                    app_id=target_app_id
                 )
 
     toolkit.success(

src/fastapi_cloud_cli/commands/env/set.py

@@ -6,7 +6,7 @@
 from pydantic import BaseModel, Field
 from rich_toolkit import RichToolkit
 
-from fastapi_cloud_cli.api import APIClient
+from fastapi_cloud_cli.api import APIClient, EnvironmentVariableCreatePayload
 from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context
 from fastapi_cloud_cli.commands.env._app import env_app
 from fastapi_cloud_cli.utils.apps import resolve_app_id_or_fail
@@ -95,16 +95,6 @@ def _resolve_environment_variable_value(
     return _input(toolkit, "Enter the value of the environment variable:")
 
 
-def _set_environment_variable(
-    client: APIClient, app_id: str, name: str, value: str, is_secret: bool = False
-) -> None:
-    response = client.post(
-        f"/apps/{app_id}/environment-variables/",
-        json={"name": name, "value": value, "is_secret": is_secret},
-    )
-    response.raise_for_status()
-
-
 @env_app.command(cls=UserCommand)
 def set(
     ctx: typer.Context,
@@ -153,13 +143,20 @@ def set(
         bool,
         typer.Option(
             "--secret",
-            help="Mark the environment variable as secret",
+            help="Mark a new environment variable as secret. Existing variables keep their secret status.",
+        ),
+    ] = False,
+    no_redeploy: Annotated[
+        bool,
+        typer.Option(
+            "--no-redeploy",
+            help="Save the environment variable without redeploying the app.",
         ),
     ] = False,
     json_output: JsonOutputOption = False,
 ) -> Any:
     """
-    Set an environment variable for the app.
+    Create or update an environment variable and redeploy the app by default.
     """
 
     toolkit = get_user_command_context(ctx).toolkit
@@ -190,20 +187,27 @@ def set(
         with toolkit.progress(
             "Setting environment variable", transient=True
         ) as progress:
-            with client.handle_http_errors(progress):
-                _set_environment_variable(
-                    client=client,
+            with client.handle_http_errors(progress, toolkit=toolkit):
+                environment_variables = client.batch_environment_variables(
                     app_id=target_app_id,
-                    name=name,
-                    value=value,
-                    is_secret=secret,
+                    upsert={
+                        name: EnvironmentVariableCreatePayload(
+                            value=value, is_secret=secret
+                        )
+                    },
+                    delete=[],
+                    redeploy=not no_redeploy,
                 )
 
+    variable = next(
+        variable for variable in environment_variables.data if variable.name == name
+    )
+
     toolkit.success(
         EnvironmentVariableSetOutput(
             app_id=target_app_id,
             name=name,
-            is_secret=secret,
+            is_secret=variable.is_secret,
             show_tag=not prompts_user,
         ),
         render_output=_render_environment_variable_set_output,