Centralised error handling & convenience.

passkey/fastapi/api.py

@@ -30,20 +30,15 @@ def register_api_routes(app: FastAPI):
     @app.post("/auth/validate")
     async def validate_token(response: Response, auth=Cookie(None)):
         """Lightweight token validation endpoint."""
-        try:
         s = await get_session(auth)
         return {
             "valid": True,
             "user_uuid": str(s.user_uuid),
         }
-        except ValueError:
-            response.status_code = 401
-            return {"valid": False}
 
     @app.post("/auth/user-info")
     async def api_user_info(response: Response, auth=Cookie(None)):
         """Get full user information for the authenticated user."""
-        try:
         reset = passphrase.is_well_formed(auth)
         s = await (get_reset if reset else get_session)(auth)
         u = await db.instance.get_user_by_uuid(s.user_uuid)
@@ -96,12 +91,6 @@ def register_api_routes(app: FastAPI):
             "credentials": credentials,
             "aaguid_info": aaguid_info,
         }
-        except ValueError as e:
-            response.status_code = 400
-            return {"detail": f"Failed to get user info: {e}"}
-        except Exception:
-            response.status_code = 500
-            return {"detail": "Failed to get user info"}
 
     @app.post("/auth/logout")
     async def api_logout(response: Response, auth=Cookie(None)):
@@ -119,7 +108,6 @@ def register_api_routes(app: FastAPI):
     @app.post("/auth/set-session")
     async def api_set_session(response: Response, auth=Depends(bearer_auth)):
         """Set session cookie from Authorization header. Fetched after login by WebSocket."""
-        try:
         user = await get_session(auth.credentials)
         if not user:
             raise ValueError("Invalid Authorization header.")
@@ -130,25 +118,10 @@ def register_api_routes(app: FastAPI):
             "user_uuid": str(user.user_uuid),
         }
 
-        except ValueError as e:
-            response.status_code = 400
-            return {"detail": str(e)}
-        except Exception:
-            response.status_code = 500
-            return {"detail": "Failed to set session"}
-
     @app.delete("/auth/credential/{uuid}")
     async def api_delete_credential(
         response: Response, uuid: UUID, auth: str = Cookie(None)
     ):
         """Delete a specific credential for the current user."""
-        try:
         await delete_credential(uuid, auth)
         return {"message": "Credential deleted successfully"}
-
-        except ValueError as e:
-            response.status_code = 400
-            return {"detail": str(e)}
-        except Exception:
-            response.status_code = 500
-            return {"detail": "Failed to delete credential"}

passkey/fastapi/mainapp.py

@@ -1,9 +1,10 @@
 import contextlib
+import logging
 from contextlib import asynccontextmanager
 from pathlib import Path
 
 from fastapi import Cookie, FastAPI, Request, Response
-from fastapi.responses import FileResponse
+from fastapi.responses import FileResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
 
 from ..authsession import get_session
@@ -30,6 +31,21 @@ async def lifespan(app: FastAPI):
 
 app = FastAPI(lifespan=lifespan)
 
+
+# Global exception handlers
+@app.exception_handler(ValueError)
+async def value_error_handler(request: Request, exc: ValueError):
+    """Handle ValueError exceptions globally with 400 status code."""
+    return JSONResponse(status_code=400, content={"detail": str(exc)})
+
+
+@app.exception_handler(Exception)
+async def general_exception_handler(request: Request, exc: Exception):
+    """Handle all other exceptions globally with 500 status code."""
+    logging.exception("Internal Server Error")
+    return JSONResponse(status_code=500, content={"detail": "Internal server error"})
+
+
 # Mount the WebSocket subapp
 app.mount("/auth/ws", ws.app)
 

passkey/fastapi/reset.py

@@ -15,7 +15,6 @@ def register_reset_routes(app):
     @app.post("/auth/create-link")
     async def api_create_link(request: Request, response: Response, auth=Cookie(None)):
         """Create a device addition link for the authenticated user."""
-        try:
         # Require authentication
         s = await get_session(auth)
 
@@ -38,13 +37,6 @@ def register_reset_routes(app):
             "expires": expires().isoformat(),
         }
 
-        except ValueError:
-            response.status_code = 401
-            return {"detail": "Authentication required"}
-        except Exception as e:
-            response.status_code = 500
-            return {"detail": f"Failed to create registration link: {str(e)}"}
-
     @app.get("/auth/{reset_token}")
     async def reset_authentication(
         request: Request,