✨ Improve error handling

Author: patrick91

src/fastapi_cli/commands/deploy.py

@@ -6,11 +6,11 @@
 from enum import Enum
 from itertools import cycle
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Union
 
 import rignore
 import typer
-from httpx import Client, HTTPError, ReadTimeout
+from httpx import Client
 from pydantic import BaseModel
 from rich_toolkit import RichToolkit
 from rich_toolkit.menu import Option
@@ -20,7 +20,7 @@
 from fastapi_cli.utils.api import APIClient
 from fastapi_cli.utils.apps import AppConfig, get_app_config, write_app_config
 from fastapi_cli.utils.auth import is_logged_in
-from fastapi_cli.utils.cli import get_rich_toolkit
+from fastapi_cli.utils.cli import get_rich_toolkit, handle_http_errors
 
 logger = logging.getLogger(__name__)
 
@@ -66,20 +66,14 @@ class Team(BaseModel):
     name: str
 
 
-def _get_teams() -> Optional[List[Team]]:
+def _get_teams() -> List[Team]:
     with APIClient() as client:
-        try:
-            response = client.get("/teams/")
-            response.raise_for_status()
-        except (HTTPError, ReadTimeout) as e:
-            logger.debug(e)
-            return None
+        response = client.get("/teams/")
+        response.raise_for_status()
 
         data = response.json()["data"]
 
-    teams = [Team.model_validate(team) for team in data]
-
-    return teams
+    return [Team.model_validate(team) for team in data]
 
 
 class CreateAppResponse(BaseModel):
@@ -89,19 +83,14 @@ class CreateAppResponse(BaseModel):
     slug: str
 
 
-def _create_app(team_slug: str, app_name: str) -> Optional[CreateAppResponse]:
+def _create_app(team_slug: str, app_name: str) -> CreateAppResponse:
     with APIClient() as client:
-        try:
-            response = client.post(
-                "/apps/",
-                json={"name": app_name, "team_slug": team_slug},
-            )
-        except (HTTPError, ReadTimeout) as e:
-            logger.debug(e)
-            return None
+        response = client.post(
+            "/apps/",
+            json={"name": app_name, "team_slug": team_slug},
+        )
 
-        if response.status_code != 201:
-            return None
+        response.raise_for_status()
 
         return CreateAppResponse.model_validate(response.json())
 
@@ -121,16 +110,10 @@ class CreateDeploymentResponse(BaseModel):
     status: DeploymentStatus
 
 
-def _create_deployment(app_id: str) -> Optional[CreateDeploymentResponse]:
+def _create_deployment(app_id: str) -> CreateDeploymentResponse:
     with APIClient() as client:
-        try:
-            response = client.post(f"/apps/{app_id}/deployments/")
-        except (HTTPError, ReadTimeout) as e:
-            logger.debug(e)
-            return None
-
-        if response.status_code != 201:
-            return None
+        response = client.post(f"/apps/{app_id}/deployments/")
+        response.raise_for_status()
 
         return CreateDeploymentResponse.model_validate(response.json())
 
@@ -140,32 +123,21 @@ class RequestUploadResponse(BaseModel):
     fields: Dict[str, str]
 
 
-def _upload_deployment(deployment_id: str, archive_path: Path) -> bool:
+def _upload_deployment(deployment_id: str, archive_path: Path) -> None:
     with APIClient() as client:
-        try:
-            response = client.post(f"/deployments/{deployment_id}/upload")
-
-            response.raise_for_status()
-        except (HTTPError, ReadTimeout) as e:
-            logger.debug(e)
-            return False
+        response = client.post(f"/deployments/{deployment_id}/upload")
+        response.raise_for_status()
 
         upload_data = RequestUploadResponse.model_validate(response.json())
 
     with Client() as client:
-        try:
-            response = client.post(
-                upload_data.url,
-                data=upload_data.fields,
-                files={"file": archive_path.open("rb")},
-            )
-
-            response.raise_for_status()
-        except (HTTPError, ReadTimeout) as e:
-            logger.debug(e)
-            return False
+        response = client.post(
+            upload_data.url,
+            data=upload_data.fields,
+            files={"file": archive_path.open("rb")},
+        )
 
-    return True
+        response.raise_for_status()
 
 
 class DeploymentResponse(BaseModel):
@@ -176,19 +148,13 @@ class DeploymentResponse(BaseModel):
     url: str
 
 
-def _get_deployment(app_id: str, deployment_id: str) -> Optional[DeploymentResponse]:
+def _get_deployment(app_id: str, deployment_id: str) -> DeploymentResponse:
     with APIClient() as client:
-        try:
-            response = client.get(f"/apps/{app_id}/deployments/{deployment_id}")
-        except (HTTPError, ReadTimeout) as e:
-            logger.debug(e)
-            return None
+        response = client.get(f"/apps/{app_id}/deployments/{deployment_id}")
+        response.raise_for_status()
 
         data = response.json()
 
-        if response.status_code != 200:
-            return None
-
     return DeploymentResponse.model_validate(data)
 
 
@@ -220,12 +186,10 @@ def _configure_app(toolkit: RichToolkit, path_to_deploy: Path) -> AppConfig:
     toolkit.print_line()
 
     with toolkit.progress("Fetching teams...") as progress:
-        teams = _get_teams()
-
-        if teams is None:
-            progress.set_error("Error fetching teams. Please try again later.")
-
-            raise typer.Exit(1)
+        with handle_http_errors(
+            progress, message="Error fetching teams. Please try again later."
+        ):
+            teams = _get_teams()
 
     toolkit.print_line()
 
@@ -246,12 +210,8 @@ def _configure_app(toolkit: RichToolkit, path_to_deploy: Path) -> AppConfig:
     toolkit.print_line()
 
     with toolkit.progress(title="Creating app...") as progress:
-        app_data = _create_app(team_slug, app_name)
-
-        if app_data is None:
-            progress.set_error("Error creating app. Please try again later.")
-
-            raise typer.Exit(1)
+        with handle_http_errors(progress):
+            app_data = _create_app(team_slug, app_name)
 
         progress.log(f"App created successfully! App slug: {app_data.slug}")
 
@@ -278,14 +238,8 @@ def _wait_for_deployment(
 
     with toolkit.progress("Deploying...") as progress:
         while True:
-            deployment = _get_deployment(app_id, deployment_id)
-
-            if deployment is None:
-                progress.set_error(
-                    "[error]Error fetching deployment status.[/]\n\nPlease try again later."
-                )
-
-                raise typer.Exit(1) from None
+            with handle_http_errors(progress):
+                deployment = _get_deployment(app_id, deployment_id)
 
             if deployment.status == DeploymentStatus.success:
                 progress.log(
@@ -354,25 +308,16 @@ def deploy(
         archive_path = archive(path or Path.cwd())  # noqa: F841
 
         with toolkit.progress(title="Creating deployment") as progress:
-            deployment = _create_deployment(app_config.id)
-
-            if deployment is None:
-                progress.set_error("Error creating deployment. Please try again later.")
-
-                raise typer.Exit(1)
-
-            progress.log(
-                f"Deployment created successfully! Deployment slug: {deployment.slug}"
-            )
-
-            progress.log("Uploading deployment...")
+            with handle_http_errors(progress):
+                deployment = _create_deployment(app_config.id)
 
-            if not _upload_deployment(deployment.id, archive_path):
-                progress.set_error(
-                    "[error]Something went while upload the archive[/]\n\nPlease try again later."
+                progress.log(
+                    f"Deployment created successfully! Deployment slug: {deployment.slug}"
                 )
 
-                raise typer.Exit(1) from None
+                progress.log("Uploading deployment...")
+
+                _upload_deployment(deployment.id, archive_path)
 
             progress.log("Deployment uploaded successfully!")
 

src/fastapi_cli/utils/cli.py

@@ -1,6 +1,15 @@
+import contextlib
+import logging
+from typing import Generator, Optional
+
+import typer
+from httpx import HTTPError, ReadTimeout
 from rich_toolkit import RichToolkit, RichToolkitTheme
+from rich_toolkit.progress import Progress
 from rich_toolkit.styles import TaggedStyle
 
+logger = logging.getLogger(__name__)
+
 
 def get_rich_toolkit() -> RichToolkit:
     theme = RichToolkitTheme(
@@ -18,3 +27,31 @@ def get_rich_toolkit() -> RichToolkit:
     )
 
     return RichToolkit(theme=theme)
+
+
+@contextlib.contextmanager
+def handle_http_errors(
+    progress: Progress,
+    message: Optional[str] = None,
+) -> Generator[None, None, None]:
+    try:
+        yield
+    except ReadTimeout as e:
+        logger.debug(e)
+
+        progress.set_error(
+            "The request to the FastAPI Cloud server timed out. Please try again later."
+        )
+
+        raise typer.Exit(1) from None
+    except HTTPError as e:
+        logger.debug(e)
+
+        message = (
+            message
+            or f"Something went wrong while contacting the FastAPI Cloud server. Please try again later. \n\n{e}"
+        )
+
+        progress.set_error(message)
+
+        raise typer.Exit(1) from None