Skip to content

统一错误响应 ​

文章不存在、创建时缺少标题、访问路径无效,都是客户端需要理解的失败结果。如果每个处理函数各自拼接错误响应,状态码和格式容易不一致。本专题先用 HTTPException 表达错误,再集中定义返回格式。

预期中的业务错误与程序故障要区分。前者应给出准确状态码和提示;后者需要保留服务器错误的语义。示例代码见下文。

报告文章不存在 ​

HTTPException 让处理函数声明状态码与错误详情;错误处理器再把它转换成响应。

先看最直接的对比。

直接返回错误响应 ​

示例文件:01_http_exception.py。运行:uvicorn 01_http_exception:app --reload。

python
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

ARTICLES = [
    {"id": 1, "title": "第一篇文章"},
    {"id": 2, "title": "第二篇文章"},
]

async def article_detail(request):
    article_id = request.path_params["article_id"]

    for article in ARTICLES:
        if article["id"] == article_id:
            return JSONResponse(article)

    return JSONResponse(
        {
            "message": "文章不存在",
        },
        status_code=404,
    )

routes = [
    Route("/articles/{article_id:int}", article_detail),
]

app = Starlette(routes=routes)

使用 HTTPException ​

更常见的方式是:

python
from starlette.applications import Starlette
from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse
from starlette.routing import Route

ARTICLES = [
    {"id": 1, "title": "第一篇文章"},
    {"id": 2, "title": "第二篇文章"},
]

async def article_detail(request):
    article_id = request.path_params["article_id"]

    for article in ARTICLES:
        if article["id"] == article_id:
            return JSONResponse(article)

    raise HTTPException(status_code=404, detail="文章不存在")

routes = [
    Route("/articles/{article_id:int}", article_detail),
]

app = Starlette(routes=routes)

两种写法都能实现 404。

但第二种写法更适合后面统一处理,因为它把“发现错误”和“组织错误响应”分开了:

text
endpoint 负责发现错误
异常处理器负责组织错误响应

重点记住:

  1. HTTPException 用来表示 HTTP 语义上的错误。
  2. raise 才是正确用法。
  3. 抛异常比到处手写错误 JSON 更利于统一处理。

错误返回解耦:手动拼接响应 vs 全局异常冒泡

统一错误格式 ​

注册异常类型处理器后,业务接口无需重复构造相同 JSON 结构。

现在给 HTTPException 注册一个统一处理器。

示例文件:02_exception_handler.py。运行:uvicorn 02_exception_handler:app --reload。

py
from starlette.applications import Starlette
from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Route

ARTICLES = [
    {"id": 1, "title": "认识 Starlette 应用对象"},
    {"id": 2, "title": "学习 Request 请求对象"},
]


async def http_exception_handler(request, exc):
    return JSONResponse(
        {
            "code": exc.status_code,
            "message": exc.detail,
        },
        status_code=exc.status_code,
    )


async def homepage(request):
    return PlainTextResponse("统一 HTTPException 处理")


async def article_detail(request):
    article_id = request.path_params["article_id"]

    for article in ARTICLES:
        if article["id"] == article_id:
            return JSONResponse(article)

    raise HTTPException(status_code=404, detail="文章不存在")


routes = [
    Route("/", homepage),
    Route("/articles/{article_id:int}", article_detail),
]

exception_handlers = {
    HTTPException: http_exception_handler,
}

app = Starlette(
    routes=routes,
    exception_handlers=exception_handlers,
)

if __name__ == '__main__':
    import uvicorn

    uvicorn.run("02_exception_handler:app", host='127.0.0.1', port=8000, reload=True)

现在流程变成:

text
endpoint 里 raise HTTPException
        ↓
Starlette 捕获异常
        ↓
交给 http_exception_handler
        ↓
统一返回 {"code": ..., "message": ...}

统一格式的作用 ​

这样做的好处很直接:

text
错误响应结构统一
接口代码更干净
后续修改格式更集中

区分 404 与服务器错误 ​

404 表示资源或路径不存在;500 表示服务端未处理的错误。不要用 200 掩盖错误,也不要把内部异常细节直接给客户端。

除了按异常类型处理,也可以按状态码处理。

示例文件:03_not_found_server_error.py。运行:uvicorn 03_not_found_server_error:app --reload。

py
from starlette.applications import Starlette
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Route


async def not_found_handler(request, exc):
    return JSONResponse(
        {
            "code": 404,
            "message": "你访问的路径不存在",
        },
        status_code=404,
    )


async def server_error_handler(request, exc):
    return JSONResponse(
        {
            "code": 500,
            "message": "服务器内部错误",
        },
        status_code=500,
    )


async def homepage(request):
    return PlainTextResponse("按状态码处理异常")


async def error_demo(request):
    result = 1 / 0
    return JSONResponse({"result": result})


routes = [
    Route("/", homepage),
    Route("/error-demo", error_demo),
]

exception_handlers = {
    404: not_found_handler,
    500: server_error_handler,
}

app = Starlette(
    routes=routes,
    exception_handlers=exception_handlers,
)

if __name__ == '__main__':
    import uvicorn

    uvicorn.run("03_not_found_server_error:app", host='127.0.0.1', port=8000, reload=True)

404 与 500 ​

这里可区分为:

text
404:资源不存在
500:服务端代码本身出错

debug 与错误显示 ​

如果 debug=True,开发环境下 500 错误通常会显示更详细的信息。 如果 debug=False,更接近统一错误响应的实际效果。

保留正确状态码 ​

即使响应体里写了“出错了”,状态码也不能乱写成 200。 客户端判断成功或失败时,通常会同时看状态码和响应体。

ExceptionMiddleware 异常查找优先级机制

自定义业务异常 ​

若特定业务错误需要独立携带信息,可以定义异常类并为它注册处理器。

有些错误不只是通用的 404 或 400,而是更具体的业务问题。

示例文件:04_business_exception.py。运行:uvicorn 04_business_exception:app --reload。

py
from starlette.applications import Starlette
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Route

ARTICLES = [
    {"id": 1, "title": "认识 Starlette 应用对象"},
    {"id": 2, "title": "学习 Request 请求对象"},
]


class ArticlePermissionError(Exception):
    pass


async def article_permission_handler(request, exc):
    return JSONResponse(
        {
            "code": 403,
            "message": "你没有权限访问后台文章管理",
        },
        status_code=403,
    )


async def homepage(request):
    return PlainTextResponse("自定义业务异常")


async def admin_articles(request):
    raise ArticlePermissionError()


routes = [
    Route("/", homepage),
    Route("/admin/articles", admin_articles),
]

exception_handlers = {
    ArticlePermissionError: article_permission_handler,
}

app = Starlette(
    routes=routes,
    exception_handlers=exception_handlers,
)
if __name__ == '__main__':
    import uvicorn

    uvicorn.run("04_business_exception:app", host='127.0.0.1', port=8000, reload=True)

这里的流程是:

text
业务中发现没有权限
        ↓
raise ArticlePermissionError
        ↓
异常处理器捕获它
        ↓
返回统一 403 JSON

业务异常的适用场景 ​

适合这种情况:

text
错误有明确业务含义
希望单独处理
不想把所有问题都混成 HTTPException

领域业务异常与 HTTP 响应适配管道

文章服务的错误处理 ​

完整案例运行 05_article_error_handling.py 后,可用 error_handling_demo.py 发送请求,查看正常与错误请求的状态码和响应体。

最后让文章服务的错误都经过登记的处理器。

目标包括:

text
文章不存在时返回统一 404
缺少标题时返回统一 400
访问不存在路径时返回统一 404
程序出错时返回统一 500
业务异常时返回统一 403

示例文件:05_article_error_handling.py。运行:uvicorn 05_article_error_handling:app --reload。

py
from starlette.applications import Starlette
from starlette.endpoints import HTTPEndpoint
from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse
from starlette.routing import Route

ARTICLES = [
    {"id": 1, "title": "认识 Starlette 应用对象"},
    {"id": 2, "title": "学习 Request 请求对象"},
    {"id": 3, "title": "学习 Response 响应对象"},
    {"id": 4, "title": "学习 Starlette 路由系统"},
]


class ArticlePermissionError(Exception):
    pass


async def http_exception_handler(request, exc):
    return JSONResponse(
        {
            "code": exc.status_code,
            "message": exc.detail,
        },
        status_code=exc.status_code,
    )


async def not_found_handler(request, exc):
    return JSONResponse(
        {
            "code": 404,
            "message": "你访问的路径不存在",
        },
        status_code=404,
    )


async def server_error_handler(request, exc):
    return JSONResponse(
        {
            "code": 500,
            "message": "服务器内部错误",
        },
        status_code=500,
    )


async def article_permission_handler(request, exc):
    return JSONResponse(
        {
            "code": 403,
            "message": "你没有权限访问后台文章管理",
        },
        status_code=403,
    )


async def error_demo(request):
    result = 1 / 0
    return JSONResponse({"result": result})


class ArticleListEndpoint(HTTPEndpoint):
    async def get(self, request):
        return JSONResponse({"items": ARTICLES})

    async def post(self, request):
        request_data = await request.json()
        title = request_data.get("title")

        if not title:
            raise HTTPException(status_code=400, detail="文章标题不能为空")

        new_article = {
            "id": len(ARTICLES) + 1,
            "title": title,
        }
        ARTICLES.append(new_article)

        return JSONResponse(
            {
                "message": "文章创建成功",
                "article": new_article,
            },
            status_code=201,
        )


class ArticleDetailEndpoint(HTTPEndpoint):
    async def get(self, request):
        article_id = request.path_params["article_id"]
        article = self.find_article(article_id)

        if article is None:
            raise HTTPException(status_code=404, detail="文章不存在")

        return JSONResponse(article)

    async def put(self, request):
        article_id = request.path_params["article_id"]
        article = self.find_article(article_id)

        if article is None:
            raise HTTPException(status_code=404, detail="文章不存在")

        request_data = await request.json()
        title = request_data.get("title")

        if not title:
            raise HTTPException(status_code=400, detail="文章标题不能为空")

        article["title"] = title

        return JSONResponse(
            {
                "message": "文章修改成功",
                "article": article,
            }
        )

    async def delete(self, request):
        article_id = request.path_params["article_id"]
        article = self.find_article(article_id)

        if article is None:
            raise HTTPException(status_code=404, detail="文章不存在")

        ARTICLES.remove(article)

        return JSONResponse(
            {
                "message": "文章删除成功",
                "article": article,
            }
        )

    def find_article(self, article_id):
        for article in ARTICLES:
            if article["id"] == article_id:
                return article
        return None


class AdminArticleEndpoint(HTTPEndpoint):
    async def get(self, request):
        raise ArticlePermissionError()


exception_handlers = {
    HTTPException: http_exception_handler,
    ArticlePermissionError: article_permission_handler,
    404: not_found_handler,
    500: server_error_handler,
}

routes = [
    Route("/articles", ArticleListEndpoint),
    Route("/articles/{article_id:int}", ArticleDetailEndpoint),
    Route("/admin/articles", AdminArticleEndpoint),
    Route("/error-demo", error_demo),
]

app = Starlette(
    routes=routes,
    exception_handlers=exception_handlers,
)

if __name__ == '__main__':
    import uvicorn

    uvicorn.run("05_article_error_handling:app", host='127.0.0.1', port=8000, reload=True)

本例同时注册了 HTTPException 处理器和 404 状态码处理器。遇到 404 时,状态码处理器优先生效,因此“文章不存在”和“路径不存在”都会得到案例中统一的 404 文案。如果需要保留两者区别,可在 404 处理器中使用 exc.detail。

1、访问文章列表

访问:

text
http://127.0.0.1:8000/articles

2、创建时缺少标题

向下面地址发送 POST JSON,但不传 title:

text
http://127.0.0.1:8000/articles

3、查询不存在的文章

访问:

text
http://127.0.0.1:8000/articles/999

4、修改不存在的文章

向下面地址发送 PUT:

text
http://127.0.0.1:8000/articles/999

5、删除不存在的文章

向下面地址发送 DELETE:

text
http://127.0.0.1:8000/articles/999

6、访问不存在的路径

例如访问:

text
http://127.0.0.1:8000/not-exists

7、访问错误演示接口

访问:

text
http://127.0.0.1:8000/error-demo

8、错误处理流程

这个案例把本章的核心点串了起来:

text
业务里发现错误
        ↓
raise 对应异常
        ↓
Starlette 捕获异常
        ↓
交给统一处理器
        ↓
返回结构清晰的 JSON 错误响应

文章服务统一错误响应拦截与降级全景图

案例源码 ​

py
from starlette.applications import Starlette
from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Route

ARTICLES = [
    {"id": 1, "title": "认识 Starlette 应用对象"},
    {"id": 2, "title": "学习 Request 请求对象"},
]


async def homepage(request):
    return PlainTextResponse("异常处理演示")


async def article_detail(request):
    article_id = request.path_params["article_id"]

    for article in ARTICLES:
        if article["id"] == article_id:
            return JSONResponse(article)

    raise HTTPException(status_code=404, detail="文章不存在")


routes = [
    Route("/", homepage),
    Route("/articles/{article_id:int}", article_detail),
]

app = Starlette(routes=routes)

if __name__ == '__main__':
    import uvicorn

    uvicorn.run("01_http_exception:app", host='127.0.0.1', port=8000, reload=True)
py
import requests


BASE_URL = "http://127.0.0.1:8000"



def test_create_article_success():
    response = requests.post(
        f"{BASE_URL}/articles",
        json={"title": "错误处理专题的新文章"},
    )
    print("POST /articles 创建成功", response.status_code, response.json())


def test_create_article_bad_request():
    response = requests.post(
        f"{BASE_URL}/articles",
        json={},
    )
    print("POST /articles 缺少标题", response.status_code, response.json())


def test_update_article_success():
    response = requests.put(
        f"{BASE_URL}/articles/1",
        json={"title": "更新后的文章标题"},
    )
    print("PUT /articles/1 修改成功", response.status_code, response.json())


def test_update_article_not_found():
    response = requests.put(
        f"{BASE_URL}/articles/999",
        json={"title": "新的标题"},
    )
    print("PUT /articles/999 文章不存在", response.status_code, response.json())


def test_delete_article_success():
    response = requests.delete(f"{BASE_URL}/articles/2")
    print("DELETE /articles/2 删除成功", response.status_code, response.json())


def test_delete_article_not_found():
    response = requests.delete(f"{BASE_URL}/articles/999")
    print("DELETE /articles/999 文章不存在", response.status_code, response.json())


if __name__ == "__main__":
    test_create_article_success()
    test_create_article_bad_request()
    test_update_article_success()
    test_update_article_not_found()
    test_delete_article_success()
    test_delete_article_not_found()