Refactor to use attempts

Author: savannahostrowski

src/fastapi_cloud_cli/commands/deploy.py

@@ -19,7 +19,7 @@
 from rich_toolkit.menu import Option
 
 from fastapi_cloud_cli.commands.login import login
-from fastapi_cloud_cli.utils.api import APIClient, BuildLogError, TooManyRetriesError
+from fastapi_cloud_cli.utils.api import APIClient, StreamLogError, TooManyRetriesError
 from fastapi_cloud_cli.utils.apps import AppConfig, get_app_config, write_app_config
 from fastapi_cloud_cli.utils.auth import is_logged_in
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors
@@ -399,7 +399,7 @@ def _wait_for_deployment(
 
                     last_message_changed_at = time.monotonic()
 
-        except (BuildLogError, TooManyRetriesError, TimeoutError) as e:
+        except (StreamLogError, TooManyRetriesError, TimeoutError) as e:
             progress.set_error(
                 dedent(f"""
                 [error]Build log streaming failed: {e}[/]
@@ -408,7 +408,7 @@ def _wait_for_deployment(
                 """).strip()
             )
 
-            raise typer.Exit(1) from e
+            raise typer.Exit(1) from None
 
 
 class SignupToWaitingList(BaseModel):

src/fastapi_cloud_cli/commands/logs.py

@@ -1,26 +1,25 @@
-import json
 import logging
-import time
-from collections.abc import Generator
 from datetime import datetime
 from pathlib import Path
 from typing import Annotated, Optional
 
 import typer
-from httpx import HTTPError, HTTPStatusError, ReadTimeout
-from pydantic import BaseModel, ValidationError
 from rich.markup import escape
 from rich_toolkit import RichToolkit
 
-from fastapi_cloud_cli.utils.api import APIClient
+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 is_logged_in
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit
 
 logger = logging.getLogger(__name__)
 
-MAX_RECONNECT_ATTEMPTS = 10
-RECONNECT_DELAY_SECONDS = 1
+
 LOG_LEVEL_COLORS = {
     "debug": "blue",
     "info": "cyan",
@@ -32,38 +31,11 @@
 }
 
 
-class LogEntry(BaseModel):
-    timestamp: datetime
-    message: str
-    level: str = "unknown"
-
-
-def _stream_logs(
-    app_id: str,
-    tail: int,
-    since: str,
-    follow: bool,
-) -> Generator[str, None, None]:
-    with APIClient() as client:
-        timeout = 120 if follow else 30
-        with client.stream(
-            "GET",
-            f"/apps/{app_id}/logs/stream",
-            params={
-                "tail": tail,
-                "since": since,
-                "follow": follow,
-            },
-            timeout=timeout,
-        ) as response:
-            response.raise_for_status()
-
-            yield from response.iter_lines()
-
-
-def _format_log_line(log: LogEntry) -> str:
+def _format_log_line(log: AppLogEntry) -> str:
     """Format a log entry for display with a colored indicator"""
-    timestamp_str = log.timestamp.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+    # Parse the timestamp string to format it consistently
+    timestamp = datetime.fromisoformat(log.timestamp.replace("Z", "+00:00"))
+    timestamp_str = timestamp.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
     color = LOG_LEVEL_COLORS.get(log.level.lower())
 
     message = escape(log.message)
@@ -81,104 +53,46 @@ def _process_log_stream(
     since: str,
     follow: bool,
 ) -> None:
+    """Stream app logs and print them to the console."""
     log_count = 0
-    last_timestamp: datetime | None = None
-    current_since = since
-    current_tail = tail
-    reconnect_attempts = 0
-
-    while True:
-        try:
-            for line in _stream_logs(
+
+    try:
+        with APIClient() as client:
+            for log in client.stream_app_logs(
                 app_id=app_config.app_id,
-                tail=current_tail,
-                since=current_since,
+                tail=tail,
+                since=since,
                 follow=follow,
             ):
-                if not line:  # pragma: no cover
-                    continue
-
-                try:
-                    data = json.loads(line)
-                except json.JSONDecodeError:
-                    logger.debug("Failed to parse log line: %s", line)
-                    continue
-
-                # Skip heartbeat messages
-                if data.get("type") == "heartbeat":  # pragma: no cover
-                    continue
-
-                if data.get("type") == "error":
-                    toolkit.print(
-                        f"Error: {data.get('message', 'Unknown error')}",
-                    )
-                    raise typer.Exit(1)
-
-                # Parse and display log entry
-                try:
-                    log_entry = LogEntry.model_validate(data)
-                    toolkit.print(_format_log_line(log_entry))
-                    log_count += 1
-                    last_timestamp = log_entry.timestamp
-                    # Reset reconnect attempts on successful log receipt
-                    reconnect_attempts = 0
-                except ValidationError as e:  # pragma: no cover
-                    logger.debug("Failed to parse log entry: %s - %s", data, e)
-                    continue
-
-            # Stream ended normally (only happens with --no-follow)
+                toolkit.print(_format_log_line(log))
+                log_count += 1
+
             if not follow and log_count == 0:
                 toolkit.print("No logs found for the specified time range.")
-            break
-
-        except KeyboardInterrupt:  # pragma: no cover
-            toolkit.print_line()
-            break
-        except (ReadTimeout, HTTPError) as e:
-            # In follow mode, try to reconnect on connection issues
-            if follow and not isinstance(e, HTTPStatusError):
-                reconnect_attempts += 1
-                if reconnect_attempts >= MAX_RECONNECT_ATTEMPTS:
-                    toolkit.print(
-                        "Lost connection to log stream. Please try again later.",
-                    )
-                    raise typer.Exit(1) from None
-
-                logger.debug(
-                    "Connection lost, reconnecting (attempt %d/%d)...",
-                    reconnect_attempts,
-                    MAX_RECONNECT_ATTEMPTS,
-                )
-
-                # On reconnect, resume from last seen timestamp
-                # The API uses strict > comparison, so logs with the same timestamp
-                # as last_timestamp will be filtered out (no duplicates)
-                if last_timestamp:  # pragma: no cover
-                    current_since = last_timestamp.isoformat()
-                    current_tail = 0  # Don't fetch historical logs again
-
-                time.sleep(RECONNECT_DELAY_SECONDS)
-                continue
-
-            if isinstance(e, HTTPStatusError) and e.response.status_code in (401, 403):
-                toolkit.print(
-                    "The specified token is not valid. Use [blue]`fastapi login`[/] to generate a new token.",
-                )
-            if isinstance(e, HTTPStatusError) and e.response.status_code == 404:
-                toolkit.print(
-                    "App not found. Make sure to use the correct account.",
-                )
-            elif isinstance(e, ReadTimeout):
-                toolkit.print(
-                    "The request timed out. Please try again later.",
-                )
-            else:
-                logger.exception("Failed to fetch logs")
-
-                toolkit.print(
-                    "Failed to fetch logs. Please try again later.",
-                )
-            raise typer.Exit(1) from None
+                return
+    except KeyboardInterrupt:  # pragma: no cover
+        toolkit.print_line()
+        return
+    except StreamLogError as e:
+        error_msg = str(e)
+        if "HTTP 401" in error_msg or "HTTP 403" in error_msg:
+            toolkit.print(
+                "The specified token is not valid. Use [blue]`fastapi login`[/] to generate a new token.",
+            )
+        elif "HTTP 404" in error_msg:
+            toolkit.print(
+                "App not found. Make sure to use the correct account.",
+            )
+        else:
+            toolkit.print(
+                f"[red]Error:[/] {escape(error_msg)}",
+            )
+        raise typer.Exit(1) from None
+    except (TooManyRetriesError, TimeoutError):
+        toolkit.print(
+            "Lost connection to log stream. Please try again later.",
+        )
+        raise typer.Exit(1) from None
 
 
 def logs(

src/fastapi_cloud_cli/utils/api.py

@@ -24,18 +24,26 @@
 
 logger = logging.getLogger(__name__)
 
-BUILD_LOG_MAX_RETRIES = 3
-BUILD_LOG_TIMEOUT = timedelta(minutes=5)
+STREAM_LOGS_MAX_RETRIES = 3
+STREAM_LOGS_TIMEOUT = timedelta(minutes=5)
 
 
-class BuildLogError(Exception):
+class StreamLogError(Exception):
+    """Raised when there's an error streaming logs (build or app logs)."""
+
     pass
 
 
 class TooManyRetriesError(Exception):
     pass
 
 
+class AppLogEntry(BaseModel):
+    timestamp: str
+    message: str
+    level: str
+
+
 class BuildLogLineGeneric(BaseModel):
     type: Literal["complete", "failed", "timeout", "heartbeat"]
     id: Optional[str] = None
@@ -90,7 +98,7 @@ def _backoff() -> None:
                 error_detail = error.response.text
             except Exception:
                 error_detail = "(response body unavailable)"
-            raise BuildLogError(
+            raise StreamLogError(
                 f"HTTP {error.response.status_code}: {error_detail}"
             ) from error
 
@@ -114,7 +122,7 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> Generator[T, None, None]:
             for attempt_number in range(total_attempts):
                 if time.monotonic() - start > timeout.total_seconds():
                     raise TimeoutError(
-                        f"Build log streaming timed out after {timeout.total_seconds():.0f}s"
+                        f"Log streaming timed out after {timeout.total_seconds():.0f}s"
                     )
 
                 with attempt(attempt_number):
@@ -144,7 +152,7 @@ def __init__(self) -> None:
             },
         )
 
-    @attempts(BUILD_LOG_MAX_RETRIES, BUILD_LOG_TIMEOUT)
+    @attempts(STREAM_LOGS_MAX_RETRIES, STREAM_LOGS_TIMEOUT)
     def stream_build_logs(
         self, deployment_id: str
     ) -> Generator[BuildLogLine, None, None]:
@@ -192,3 +200,44 @@ def _parse_log_line(self, line: str) -> Optional[BuildLogLine]:
         except (ValidationError, json.JSONDecodeError) as e:
             logger.debug("Skipping malformed log: %s (error: %s)", line[:100], e)
             return None
+
+    @attempts(STREAM_LOGS_MAX_RETRIES, STREAM_LOGS_TIMEOUT)
+    def stream_app_logs(
+        self,
+        app_id: str,
+        tail: int,
+        since: str,
+        follow: bool,
+    ) -> Generator[AppLogEntry, None, None]:
+        timeout = 120 if follow else 30
+        with self.stream(
+            "GET",
+            f"/apps/{app_id}/logs/stream",
+            params={
+                "tail": tail,
+                "since": since,
+                "follow": follow,
+            },
+            timeout=timeout,
+        ) as response:
+            response.raise_for_status()
+            for line in response.iter_lines():
+                if not line or not line.strip():  # pragma: no cover
+                    continue
+                try:
+                    data = json.loads(line)
+                except json.JSONDecodeError:
+                    logger.debug("Failed to parse log line: %s", line)
+                    continue
+
+                if data.get("type") == "heartbeat":
+                    continue
+
+                if data.get("type") == "error":
+                    raise StreamLogError(data.get("message", "Unknown error"))
+
+                try:
+                    yield AppLogEntry.model_validate(data)
+                except ValidationError as e:  # pragma: no cover
+                    logger.debug("Failed to parse log entry: %s - %s", data, e)
+                    continue

tests/test_api_client.py

@@ -9,10 +9,10 @@
 
 from fastapi_cloud_cli.config import Settings
 from fastapi_cloud_cli.utils.api import (
-    BUILD_LOG_MAX_RETRIES,
+    STREAM_LOGS_MAX_RETRIES,
     APIClient,
-    BuildLogError,
     BuildLogLineMessage,
+    StreamLogError,
     TooManyRetriesError,
 )
 from tests.utils import build_logs_response
@@ -243,7 +243,7 @@ def test_stream_build_logs_client_error_raises_immediately(
 ) -> None:
     logs_route.mock(return_value=Response(404, text="Not Found"))
 
-    with pytest.raises(BuildLogError, match="HTTP 404"):
+    with pytest.raises(StreamLogError, match="HTTP 404"):
         list(client.stream_build_logs(deployment_id))
 
 
@@ -255,7 +255,8 @@ def test_stream_build_logs_max_retries_exceeded(
 
     with patch("time.sleep"):
         with pytest.raises(
-            TooManyRetriesError, match=f"Failed after {BUILD_LOG_MAX_RETRIES} attempts"
+            TooManyRetriesError,
+            match=f"Failed after {STREAM_LOGS_MAX_RETRIES} attempts",
         ):
             list(client.stream_build_logs(deployment_id))
 
@@ -343,7 +344,7 @@ def test_stream_build_logs_connection_closed_without_complete_failed_or_timeout(
     logs = client.stream_build_logs(deployment_id)
 
     with patch("time.sleep"), pytest.raises(TooManyRetriesError, match="Failed after"):
-        for _ in range(BUILD_LOG_MAX_RETRIES + 1):
+        for _ in range(STREAM_LOGS_MAX_RETRIES + 1):
             next(logs)