52 lines
1.4 KiB
Python
52 lines
1.4 KiB
Python
"""Custom exception classes and FastAPI exception handlers."""
|
|
|
|
from fastapi import Request
|
|
from fastapi.responses import JSONResponse
|
|
|
|
|
|
class ChunkingError(Exception):
|
|
"""Base exception for chunking-related errors."""
|
|
|
|
|
|
class StrategyNotFoundError(ChunkingError):
|
|
"""Raised when an unknown strategy name is requested."""
|
|
|
|
|
|
class DocumentProcessingError(ChunkingError):
|
|
"""Raised when a document cannot be parsed or processed."""
|
|
|
|
|
|
class EnrichmentError(ChunkingError):
|
|
"""Raised when the contextual enrichment LLM call fails."""
|
|
|
|
|
|
class EmbeddingError(ChunkingError):
|
|
"""Raised when the OpenAI embedding API call fails."""
|
|
|
|
|
|
class QdrantError(ChunkingError):
|
|
"""Raised when a Qdrant operation fails."""
|
|
|
|
|
|
class BenchmarkError(Exception):
|
|
"""Base exception for benchmarking-related errors."""
|
|
|
|
|
|
class DryRunError(BenchmarkError):
|
|
"""Raised when a dry-run estimation fails."""
|
|
|
|
|
|
async def chunking_exception_handler(request: Request, exc: ChunkingError) -> JSONResponse:
|
|
"""Handle all chunking-related errors."""
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"detail": str(exc), "type": type(exc).__name__},
|
|
)
|
|
|
|
|
|
async def benchmark_exception_handler(request: Request, exc: BenchmarkError) -> JSONResponse:
|
|
"""Handle all benchmarking-related errors."""
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"detail": str(exc), "type": type(exc).__name__},
|
|
) |