Spaces:
Sleeping
Sleeping
File size: 1,820 Bytes
8b2e901 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | """自定义异常 + FastAPI 异常处理."""
from __future__ import annotations
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
class AppError(Exception):
"""应用层业务异常的基类."""
code: str = "app_error"
status_code: int = 500
retryable: bool = False
def __init__(
self,
message: str,
*,
code: str | None = None,
status_code: int | None = None,
retryable: bool | None = None,
detail: dict | None = None,
) -> None:
super().__init__(message)
self.message = message
if code is not None:
self.code = code
if status_code is not None:
self.status_code = status_code
if retryable is not None:
self.retryable = retryable
self.detail = detail or {}
class DocumentNotFoundError(AppError):
code = "document_not_found"
status_code = 404
class IngestionFailedError(AppError):
code = "ingestion_failed"
status_code = 500
retryable = True
class LLMUnavailableError(AppError):
code = "llm_unavailable"
status_code = 503
retryable = True
class VectorStoreUnavailableError(AppError):
code = "vector_store_unavailable"
status_code = 503
retryable = True
def install_exception_handlers(app: FastAPI) -> None:
"""挂载全局异常处理. 由 main.py 在创建 app 后调用."""
@app.exception_handler(AppError)
async def _app_error_handler(_: Request, exc: AppError) -> JSONResponse:
return JSONResponse(
status_code=exc.status_code,
content={
"code": exc.code,
"message": exc.message,
"retryable": exc.retryable,
"detail": exc.detail,
},
)
|