f1c2b73ade
- Set up FastAPI project structure with SQLite and SQLAlchemy - Create models for users, books, authors, categories, and orders - Implement JWT authentication and authorization - Add CRUD endpoints for all resources - Set up Alembic for database migrations - Add health check endpoint - Add proper error handling and validation - Create comprehensive documentation
68 lines
1.9 KiB
Python
68 lines
1.9 KiB
Python
from fastapi import Request, status
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.exceptions import RequestValidationError
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from typing import Union, Dict, Any
|
|
|
|
from app.main import app
|
|
|
|
|
|
class BookstoreException(Exception):
|
|
"""Base exception for the bookstore application"""
|
|
|
|
def __init__(
|
|
self,
|
|
status_code: int,
|
|
detail: Union[str, Dict[str, Any]],
|
|
headers: Dict[str, str] = None,
|
|
):
|
|
self.status_code = status_code
|
|
self.detail = detail
|
|
self.headers = headers
|
|
|
|
|
|
@app.exception_handler(BookstoreException)
|
|
async def bookstore_exception_handler(request: Request, exc: BookstoreException):
|
|
"""Handler for custom bookstore exceptions"""
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content={"detail": exc.detail},
|
|
headers=exc.headers,
|
|
)
|
|
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def validation_exception_handler(request: Request, exc: RequestValidationError):
|
|
"""Handler for request validation errors"""
|
|
return JSONResponse(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
content={
|
|
"detail": "Validation error",
|
|
"errors": exc.errors(),
|
|
},
|
|
)
|
|
|
|
|
|
@app.exception_handler(SQLAlchemyError)
|
|
async def sqlalchemy_exception_handler(request: Request, exc: SQLAlchemyError):
|
|
"""Handler for database errors"""
|
|
return JSONResponse(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
content={
|
|
"detail": "Database error",
|
|
"error": str(exc),
|
|
},
|
|
)
|
|
|
|
|
|
@app.exception_handler(Exception)
|
|
async def general_exception_handler(request: Request, exc: Exception):
|
|
"""Handler for unhandled exceptions"""
|
|
return JSONResponse(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
content={
|
|
"detail": "Internal server error",
|
|
"error": str(exc),
|
|
},
|
|
)
|