统一异常处理

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="墨集 inkhub 内容平台")


class ArticleIn(BaseModel):
    title: str = Field(min_length=1)


@app.get("/articles/{article_id}")
def get_article(article_id: int) -> dict[str, int]:
    if article_id != 1:
        raise HTTPException(status_code=404, detail="文章不存在")
    return {"article_id": article_id}


@app.post("/articles")
def create_article(article: ArticleIn) -> dict[str, str]:
    return {"title": article.title}

curl http://127.0.0.1:8000/articles/2
# 404 → {"detail":"文章不存在"}

curl http://127.0.0.1:8000/articles/abc
# 422 → {"detail":[{"type":"int_parsing","loc":["path","article_id"],"msg":"Input should be a valid integer, ...", ...}]}

curl http://127.0.0.1:8000/no/such/path
# 404 → {"detail":"Not Found"}

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

# 写法一：装饰器
@app.exception_handler(异常类)
async def any_name(request: Request, exc: 异常类) -> JSONResponse:
    ...  # 返回任意 Response 对象

# 写法二：方法调用（适合集中注册，见 9.8 节）
app.add_exception_handler(异常类, any_name)

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI(title="墨集 inkhub 内容平台")


@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
    # exc.errors() 拿到错误列表，每条含 type / loc / msg
    parts = [
        f"{'.'.join(str(p) for p in err['loc'])}: {err['msg']}"
        for err in exc.errors()
    ]
    return JSONResponse(
        status_code=422,
        content={"code": "422000", "message": "；".join(parts)},
    )

class BusinessException(Exception):
    """业务异常基类：所有「能向用户解释」的错误都从它派生。"""

    http_status: int = 400                  # 映射的 HTTP 状态码
    code: str = "400000"                    # 业务错误码
    default_message: str = "业务处理失败"     # 默认人话消息

    def __init__(self, message: str | None = None, *, detail: str = "") -> None:
        # 不传消息 → 用类上的 default_message，raise 处一行搞定
        self.message = message or self.default_message
        # detail 是写给开发者的内部线索：只进日志，绝不进响应
        self.detail = detail
        super().__init__(self.message)

raise BusinessException()                       # 400 / 400000 / 业务处理失败
raise BusinessException("专栏订阅数已达上限")     # 临时覆盖消息

BusinessException               # 400 / 400000 / 业务处理失败
├── NotFoundError               # 404 / 404000 / 资源不存在
│   ├── ArticleNotFound         # 404 / 404001 / 文章不存在
│   ├── ChannelNotFound         # 404 / 404002 / 栏目不存在
│   └── VersionNotFound         # 404 / 404003 / 文章版本不存在
└── ConflictError               # 409 / 409000 / 数据冲突
    └── ArticleSlugExists       # 409 / 409001 / 文章 slug 已被占用

class NotFoundError(BusinessException):
    """「资源不存在」家族：404 楼层的通用房间。"""

    http_status = 404
    code = "404000"
    default_message = "资源不存在"


class ArticleNotFound(NotFoundError):
    code = "404001"                 # 只需要两行：换房间号、换默认消息
    default_message = "文章不存在"


class ChannelNotFound(NotFoundError):
    code = "404002"
    default_message = "栏目不存在"


class VersionNotFound(NotFoundError):
    code = "404003"
    default_message = "文章版本不存在"

raise ArticleNotFound()                            # 用默认消息
raise ArticleNotFound("文章不存在或已被作者删除")     # 临时覆盖：同一种错误，场景化提示
raise ArticleSlugExists(detail="dup key: fastapi-exception")  # 带内部线索，只进日志

@app.exception_handler(BusinessException)
async def business_handler(request: Request, exc: BusinessException) -> JSONResponse:
    return JSONResponse(
        status_code=exc.http_status,
        content={"code": exc.code, "message": exc.message},
    )

404001
└┬┘└┬┘
 │ └── 房间号：404 楼下的第 1 号场景（文章不存在）
 └─── 楼层号：HTTP 404

from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession


async def create_article(article: ArticleIn, session: AsyncSession) -> None:
    session.add(Article(**article.model_dump()))
    try:
        await session.commit()
    except IntegrityError as exc:
        # 数据库黑话 → 人话；原文挂到 detail，只进日志
        raise ArticleSlugExists(detail=str(exc.orig)) from exc

from sqlalchemy.exc import IntegrityError


@app.exception_handler(IntegrityError)
async def integrity_handler(request: Request, exc: IntegrityError) -> JSONResponse:
    # 分不清具体是哪种约束冲突，只能给家族通用的笼统话
    return JSONResponse(
        status_code=409,
        content={"code": "409000", "message": "数据冲突"},
    )

def register_exception_handlers(app: FastAPI) -> None:
    """给 app 装上全套异常处理器：四个出口，一种形状。"""

    @app.exception_handler(BusinessException)
    async def handle_business_exception(
        request: Request, exc: BusinessException
    ) -> JSONResponse:
        return error_response(exc.code, exc.message, exc.http_status)

    @app.exception_handler(RequestValidationError)
    async def handle_validation_exception(
        request: Request, exc: RequestValidationError
    ) -> JSONResponse:
        message = _format_validation_message(exc) or "请求数据校验失败"
        return error_response("422000", message, 422)

    @app.exception_handler(HTTPException)          # starlette 的那个！
    async def handle_http_exception(request: Request, exc: HTTPException) -> JSONResponse:
        message = PROTOCOL_MESSAGES.get(exc.status_code, str(exc.detail))
        return error_response(str(exc.status_code), message, exc.status_code)

    @app.exception_handler(Exception)
    async def handle_unknown_exception(request: Request, exc: Exception) -> JSONResponse:
        return error_response("500000", "服务器开小差了，请稍后重试", 500)

# main.py
app = FastAPI(title="墨集 inkhub 内容平台")
register_exception_handlers(app)     # 就这一行

from collections.abc import Iterator


def _iter_exception_tree(cls: type[BusinessException]) -> Iterator[type[BusinessException]]:
    """递归遍历所有子类：__subclasses__() 只给直接子类，孙类要自己往下挖。"""
    for sub in cls.__subclasses__():
        yield sub
        yield from _iter_exception_tree(sub)


def build_error_docs() -> str:
    lines = [
        "统一错误响应格式：`{\"code\": \"业务错误码\", \"message\": \"人话消息\"}`。",
        "",
        "| 错误码 | HTTP 状态码 | 默认消息 | 异常类 |",
        "| --- | --- | --- | --- |",
    ]
    for exc_type in _iter_exception_tree(BusinessException):
        lines.append(
            f"| {exc_type.code} | {exc_type.http_status} | "
            f"{exc_type.default_message} | {exc_type.__name__} |"
        )
    return "\n".join(lines)


app = FastAPI(
    title="墨集 inkhub 内容平台",
    description="统一异常处理演示。\n\n" + build_error_docs(),
)

cd demo
python demo_exception.py

# ① 合法请求 → 200
curl -i http://127.0.0.1:8000/articles/1
# {"id":1,"title":"FastAPI 异常处理指南","slug":"fastapi-exception","channel_id":1,"author_id":1}

# ② 文章不存在 → 404
curl -i http://127.0.0.1:8000/articles/999
# {"code":"404001","message":"文章不存在"}

# ③ 校验失败 → 422（路径参数 abc 过不了 int 校验）
curl -i http://127.0.0.1:8000/articles/abc
# {"code":"422000","message":"path.article_id: Input should be a valid integer, unable to parse string as an integer"}