global exception handling

Author: getmarkus

main.py

@@ -10,6 +10,8 @@
 
 from config import Settings
 from src.interface_adapters import api_router
+from src.interface_adapters.exceptions import AppException
+from src.interface_adapters.middleware.error_handler import app_exception_handler
 
 # https://brandur.org/logfmt
 # https://github.com/Delgan/loguru
@@ -44,6 +46,9 @@ async def isConfigured():
     openapi_url="/v1/openapi.json",
 )
 
+# Register global exception handler
+app.add_exception_handler(AppException, app_exception_handler)
+
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],  # Set this to lock down if applicable to your use case

rapidapi/api_tests.paw

@@ -4,10 +4,9 @@
 from loguru import logger
 
 from src.app.ports.repositories.issues import IssueRepository
-from src.app.repository import (
-    UnitOfWork,
-)
+from src.app.repository import UnitOfWork
 from src.domain.issue import Issue
+from src.interface_adapters.exceptions import NotFoundException
 
 
 class AnalyzeIssueProtocol(Protocol):
@@ -34,4 +33,9 @@ def __init__(
     async def analyze(self) -> Issue:
         logger.info(f"analyzing issue: {self.issue_number}")
         issue = await self.repo.get_by_id(self.issue_number)
+        if issue.issue_number == 0:
+            raise NotFoundException(
+                message="Issue not found",
+                detail=f"Issue with number {self.issue_number} does not exist"
+            )
         return issue

src/app/usecases/analyze_issue.py

@@ -4,28 +4,33 @@
 
 from src.app.usecases.analyze_issue import AnalyzeIssue
 from config import Settings
-
 from src.domain.issue import Issue
+from src.interface_adapters.exceptions import UnsupportedOperationException
 from src.resource_adapters.persistence.in_memory.issues import InMemoryIssueRepository
 from src.resource_adapters.persistence.in_memory.unit_of_work import UnitOfWork
-from typing import Optional
 
 router = APIRouter()
 
 
 # https://fastapi.tiangolo.com/tutorial/dependencies/
-async def configure_unit_of_work() -> Optional[UnitOfWork]:
+async def configure_unit_of_work() -> UnitOfWork:
     if Settings.get_settings().execution_mode == "in-memory":
         return UnitOfWork()
     else:
-        return None
+        raise UnsupportedOperationException(
+            message="Unsupported unit of work configuration",
+            detail="Only in-memory unit of work is currently supported"
+        )
 
 
-async def configure_repository() -> Optional[InMemoryIssueRepository]:
+async def configure_repository() -> InMemoryIssueRepository:
     if Settings.get_settings().execution_mode == "in-memory":
         return InMemoryIssueRepository()
     else:
-        return None
+        raise UnsupportedOperationException(
+            message="Unsupported repository configuration",
+            detail="Only in-memory repository is currently supported"
+        )
 
 
 @router.post("/issues/{issue_number}/analyze", response_model=Issue)

src/interface_adapters/endpoints/issues.py

@@ -0,0 +1,56 @@
+from typing import Optional
+
+class AppException(Exception):
+    """Base exception class for application exceptions"""
+    def __init__(
+        self,
+        message: str,
+        status_code: int = 500,
+        detail: Optional[str] = None
+    ) -> None:
+        self.message = message
+        self.status_code = status_code
+        self.detail = detail
+        super().__init__(self.message)
+
+
+class UnsupportedOperationException(AppException):
+    """Exception raised when an operation is not supported"""
+    def __init__(
+        self,
+        message: str = "Operation not supported",
+        detail: Optional[str] = None
+    ) -> None:
+        super().__init__(
+            message=message,
+            status_code=501,
+            detail=detail
+        )
+
+
+class ConfigurationException(AppException):
+    """Exception raised for configuration errors"""
+    def __init__(
+        self,
+        message: str = "Invalid configuration",
+        detail: Optional[str] = None
+    ) -> None:
+        super().__init__(
+            message=message,
+            status_code=500,
+            detail=detail
+        )
+
+
+class NotFoundException(AppException):
+    """Exception raised when a requested resource is not found"""
+    def __init__(
+        self,
+        message: str = "Resource not found",
+        detail: Optional[str] = None
+    ) -> None:
+        super().__init__(
+            message=message,
+            status_code=404,
+            detail=detail
+        )

src/interface_adapters/exceptions.py

@@ -0,0 +1,21 @@
+from fastapi import Request
+from fastapi.responses import JSONResponse
+from loguru import logger
+
+from src.interface_adapters.exceptions import AppException
+
+
+async def app_exception_handler(request: Request, exc: AppException) -> JSONResponse:
+    """Global exception handler for AppException and its subclasses"""
+    logger.error(f"Request to {request.url} failed: {exc.message}")
+    if exc.detail:
+        logger.error(f"Detail: {exc.detail}")
+    
+    return JSONResponse(
+        status_code=exc.status_code,
+        content={
+            "message": exc.message,
+            "detail": exc.detail,
+            "path": str(request.url)
+        }
+    )