fix: better error handling

Author: mairasalazar

app/activities/extract_pdf_content.py

@@ -1,9 +1,10 @@
 import httpx
+from fastapi import HTTPException
 from pydantic import BaseModel
 from temporalio import activity
 from temporalio.exceptions import ApplicationError
 
-from app.config import Environment, get_settings
+from app.config import get_settings
 from app.extractors import get_extractor
 
 settings = get_settings()
@@ -28,7 +29,7 @@ class ExtractPdfContentResponse(BaseModel):
 @activity.defn
 async def create(request: ExtractPdfContentRequest) -> ExtractPdfContentResponse:
     """Read a file and extract its text content using the specified extractor."""
-    if settings.orcha_env in {Environment.LOCAL, Environment.DEV}:
+    if settings.orcha_env in {"local", "dev"}:
         try:
             with open(request.url, "rb") as f:
                 pdf_bytes = f.read()
@@ -39,9 +40,20 @@ async def create(request: ExtractPdfContentRequest) -> ExtractPdfContentResponse
             ) from e
     else:
         async with httpx.AsyncClient() as client:
-            response = await client.get(request.url)
-            response.raise_for_status()
-            pdf_bytes = response.content
+            try:
+                response = await client.get(request.url)
+                response.raise_for_status()
+                pdf_bytes = response.content
+            except httpx.TimeoutException:
+                raise HTTPException(
+                    status_code=504,
+                    detail="Request to fetch file timed out.",
+                )
+            except httpx.HTTPError as e:
+                raise HTTPException(
+                    status_code=502,
+                    detail=f"Failed to fetch file from URL: {e.response.text}.",
+                )
 
     # Extract content using the extraction module
     try:

app/config.py

@@ -1,20 +1,10 @@
 """Centralized application settings using Pydantic Settings."""
 
-from enum import Enum
 from functools import lru_cache
 
 from pydantic_settings import BaseSettings
 
 
-class Environment(str, Enum):
-    """Application environment."""
-
-    LOCAL = "local"
-    DEV = "dev"
-    QA = "qa"
-    PROD = "prod"
-
-
 class Settings(BaseSettings):
     """Application configuration loaded from environment variables."""
 
@@ -37,7 +27,7 @@ class Settings(BaseSettings):
     allowed_origins: list[str] = ["http://localhost:3000", "http://127.0.0.1:3000"]
 
     # Environment
-    orcha_env: Environment = Environment.LOCAL
+    orcha_env: str = "local"
 
     @property
     def database_url(self) -> str:

app/routers/workflows.py

@@ -6,6 +6,7 @@
 from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel
 from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm.exc import NoResultFound
 from sqlmodel import Session, select
 from sse_starlette import EventSourceResponse, ServerSentEvent
 from temporalio.client import Client
@@ -14,6 +15,7 @@
 from app.database.models import Workflow, WorkflowStatus
 from app.database.session import get_session
 from app.dependencies import get_current_user
+from app.errors import WorkflowEventError
 from app.workflows.extract_metadata_workflow import ExtractMetadata
 
 router = APIRouter(
@@ -150,48 +152,43 @@ async def workflow_event(request: Request, workflow_id: str):
         if await request.is_disconnected():
             break
 
-        with Session(request.app.state.db_engine) as session:
-            try:
-                workflow = session.exec(
-                    select(Workflow).where(Workflow.public_id == workflow_id)
-                ).one()
-
-                status = workflow.status
-
-                if status == WorkflowStatus.SUCCESS:
-                    yield ServerSentEvent(
-                        data=json.dumps(workflow.result), event="metadata"
-                    )
-                    yield ServerSentEvent(data="done", event="end")
-                    break
-
-                if status == WorkflowStatus.ERROR:
-                    yield ServerSentEvent(
-                        # TODO: improve it with a better error message for end users
-                        data="The Temporal Workflow failed",
-                        event="error",
-                    )
-                    yield ServerSentEvent(data="done", event="end")
-                    break
-
-            except SQLAlchemyError as e:
-                print("Error in fetching from database (stream_workflow)", e)
+        try:
+            with Session(request.app.state.db_engine) as session:
+                try:
+                    workflow = session.exec(
+                        select(Workflow).where(Workflow.public_id == workflow_id)
+                    ).one()
+                except NoResultFound:
+                    raise WorkflowEventError(error_code="WORKFLOW_NOT_FOUND")
+                except SQLAlchemyError as e:
+                    print("Error in fetching from database (stream_workflow)", e)
+                    raise WorkflowEventError(error_code="DB_FETCH_FAILED")
+
+            status = workflow.status
+
+            if status == WorkflowStatus.SUCCESS:
                 yield ServerSentEvent(
-                    data="Failed to read workflow status.",
-                    event="error",
+                    data=json.dumps(workflow.result), event="metadata"
                 )
                 yield ServerSentEvent(data="done", event="end")
                 break
 
-            except Exception as e:
-                print("Error(stream_workflow)", e)
+            if status == WorkflowStatus.ERROR:
                 yield ServerSentEvent(
-                    data="An unexpected error occurred while streaming workflow results.",  # noqa: E501
+                    data=json.dumps({"error_code": "WORKFLOW_FAILED"}),
                     event="error",
                 )
                 yield ServerSentEvent(data="done", event="end")
                 break
 
+        except WorkflowEventError as e:
+            yield ServerSentEvent(
+                data=json.dumps({"error_code": e.error_code}),
+                event="error",
+            )
+            yield ServerSentEvent(data="done", event="end")
+            break
+
         await asyncio.sleep(STREAM_DELAY)