admin
d703ab64e3
CI / docs-check (push) Canceled after 0s
CI / backend-test (push) Canceled after 0s
CI / service-test (push) Canceled after 0s
CI / frontend-test (push) Canceled after 0s
CI / rust-core (push) Canceled after 0s
CI / docs-check (pull_request) Canceled after 0s
CI / backend-test (pull_request) Canceled after 0s
CI / service-test (pull_request) Canceled after 0s
CI / frontend-test (pull_request) Canceled after 0s
CI / rust-core (pull_request) Canceled after 0s
68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
from typing import Any
|
|
|
|
from fastapi import Request
|
|
from fastapi.encoders import jsonable_encoder
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.exceptions import HTTPException as StarletteHttpException
|
|
|
|
from app.contracts import ErrorDetail, ErrorResponse
|
|
|
|
|
|
class ApiError(Exception):
|
|
def __init__(
|
|
self,
|
|
status_code: int,
|
|
code: str,
|
|
message: str,
|
|
details: dict[str, Any] | None = None,
|
|
) -> None:
|
|
super().__init__(message)
|
|
self.status_code = status_code
|
|
self.code = code
|
|
self.message = message
|
|
self.details = details or {}
|
|
|
|
|
|
async def api_error_handler(_: Request, exc: ApiError) -> JSONResponse:
|
|
from app.operation_logs import log_event
|
|
log_event('api', 'operation.failed', level='ERROR' if exc.status_code >= 500 else 'WARNING',
|
|
error=exc, status=exc.status_code,
|
|
**{key: value for key, value in exc.details.items() if key in {'run_id', 'task_id', 'note_id', 'job_id', 'provider_id'}})
|
|
body = ErrorResponse(
|
|
error=ErrorDetail(code=exc.code, message=exc.message, details=exc.details)
|
|
)
|
|
return JSONResponse(status_code=exc.status_code, content=jsonable_encoder(body))
|
|
|
|
|
|
async def validation_error_handler(_: Request, exc: RequestValidationError) -> JSONResponse:
|
|
body = ErrorResponse(
|
|
error=ErrorDetail(
|
|
code="VALIDATION_ERROR",
|
|
message="Request validation failed.",
|
|
# Pydantic ctx可以包含异常对象；输入可能包含 API 键。
|
|
details={"errors": [
|
|
{key: error[key] for key in ("type", "loc", "msg") if key in error}
|
|
for error in exc.errors()
|
|
]},
|
|
)
|
|
)
|
|
return JSONResponse(status_code=422, content=jsonable_encoder(body))
|
|
|
|
|
|
async def http_error_handler(_: Request, exc: StarletteHttpException) -> JSONResponse:
|
|
code = "RESOURCE_NOT_FOUND" if exc.status_code == 404 else "HTTP_ERROR"
|
|
body = ErrorResponse(
|
|
error=ErrorDetail(code=code, message=str(exc.detail), details={})
|
|
)
|
|
return JSONResponse(status_code=exc.status_code, content=jsonable_encoder(body))
|
|
|
|
|
|
def not_implemented(resource: str) -> None:
|
|
raise ApiError(
|
|
status_code=501,
|
|
code="NOT_IMPLEMENTED",
|
|
message=f"{resource} contract is available, but its business service is not implemented.",
|
|
details={"resource": resource},
|
|
)
|