Sergey Vanyushkin
ce2c052684
Add comprehensive integration and API tests: - Integration tests for SQLAlchemyPostRepository (34 tests) - API tests for posts endpoints and error handlers (22 tests) - Unit tests for PublishPostUseCase and ListPostsUseCase - Unit tests for SessionTransactionManager Also register generic exception handler in error_handler.py All 167 tests pass, coverage now meets CI threshold of 70%
133 lines
3.6 KiB
Python
133 lines
3.6 KiB
Python
"""Exception handling middleware.
|
|
|
|
This module provides exception handlers for FastAPI application.
|
|
Maps domain exceptions to appropriate HTTP status codes.
|
|
"""
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
|
|
|
from app.domain.exceptions import (
|
|
AlreadyExistsException,
|
|
DomainException,
|
|
ForbiddenException,
|
|
NotFoundException,
|
|
UnauthorizedException,
|
|
ValidationException,
|
|
)
|
|
|
|
|
|
def get_status_code(exc: DomainException) -> int:
|
|
"""Map domain exceptions to HTTP status codes.
|
|
|
|
Args:
|
|
exc: Domain exception instance.
|
|
|
|
Returns:
|
|
HTTP status code for the exception type.
|
|
"""
|
|
match exc:
|
|
case ValidationException():
|
|
return 400
|
|
case UnauthorizedException():
|
|
return 401
|
|
case ForbiddenException():
|
|
return 403
|
|
case NotFoundException():
|
|
return 404
|
|
case AlreadyExistsException():
|
|
return 409
|
|
case _:
|
|
return 500
|
|
|
|
|
|
async def domain_exception_handler(request: Request, exc: DomainException) -> JSONResponse:
|
|
"""Handle domain exceptions.
|
|
|
|
Converts domain exceptions to JSON error responses.
|
|
|
|
Args:
|
|
request: FastAPI request object.
|
|
exc: Domain exception instance.
|
|
|
|
Returns:
|
|
JSONResponse with error details.
|
|
"""
|
|
status_code = get_status_code(exc)
|
|
return JSONResponse(
|
|
status_code=status_code,
|
|
content={
|
|
"error": exc.__class__.__name__,
|
|
"message": exc.message,
|
|
"timestamp": datetime.now(UTC).isoformat(),
|
|
"path": str(request.url.path),
|
|
},
|
|
)
|
|
|
|
|
|
async def http_exception_handler(request: Request, exc: StarletteHTTPException) -> JSONResponse:
|
|
"""Handle HTTP exceptions.
|
|
|
|
Converts Starlette HTTP exceptions to JSON error responses.
|
|
|
|
Args:
|
|
request: FastAPI request object.
|
|
exc: Starlette HTTP exception instance.
|
|
|
|
Returns:
|
|
JSONResponse with error details.
|
|
"""
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content={
|
|
"error": "HTTPException",
|
|
"message": str(exc.detail),
|
|
"timestamp": datetime.now(UTC).isoformat(),
|
|
"path": str(request.url.path),
|
|
},
|
|
)
|
|
|
|
|
|
async def generic_exception_handler(request: Request, exc: Exception) -> JSONResponse:
|
|
"""Handle generic exceptions.
|
|
|
|
Converts unhandled exceptions to generic error responses.
|
|
Hides internal details for security.
|
|
|
|
Args:
|
|
request: FastAPI request object.
|
|
exc: Generic exception instance.
|
|
|
|
Returns:
|
|
JSONResponse with generic error message.
|
|
"""
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={
|
|
"error": "InternalServerError",
|
|
"message": "An unexpected error occurred",
|
|
"timestamp": datetime.now(UTC).isoformat(),
|
|
"path": str(request.url.path),
|
|
},
|
|
)
|
|
|
|
|
|
def register_exception_handlers(app: FastAPI) -> None:
|
|
"""Register all exception handlers with FastAPI app.
|
|
|
|
Args:
|
|
app: FastAPI application instance.
|
|
|
|
Raises:
|
|
TypeError: If app is not a FastAPI instance.
|
|
"""
|
|
if not isinstance(app, FastAPI):
|
|
raise TypeError("app must be a FastAPI instance")
|
|
|
|
app.add_exception_handler(DomainException, domain_exception_handler) # type: ignore[arg-type]
|
|
app.add_exception_handler(StarletteHTTPException, http_exception_handler) # type: ignore[arg-type]
|
|
app.add_exception_handler(Exception, generic_exception_handler)
|