统一错误响应
文章不存在、创建时缺少标题、访问路径无效,都是客户端需要理解的失败结果。如果每个处理函数各自拼接错误响应,状态码和格式容易不一致。本专题先用 HTTPException 表达错误,再集中定义返回格式。
预期中的业务错误与程序故障要区分。前者应给出准确状态码和提示;后者需要保留服务器错误的语义。示例代码见下文。
报告文章不存在
HTTPException 让处理函数声明状态码与错误详情;错误处理器再把它转换成响应。
先看最直接的对比。
直接返回错误响应
示例文件:01_http_exception.py。运行:uvicorn 01_http_exception:app --reload。
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
更常见的方式是:
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。
但第二种写法更适合后面统一处理,因为它把“发现错误”和“组织错误响应”分开了:
endpoint 负责发现错误
异常处理器负责组织错误响应重点记住:
HTTPException用来表示 HTTP 语义上的错误。raise才是正确用法。- 抛异常比到处手写错误 JSON 更利于统一处理。
统一错误格式
注册异常类型处理器后,业务接口无需重复构造相同 JSON 结构。
现在给 HTTPException 注册一个统一处理器。
示例文件:02_exception_handler.py。运行:uvicorn 02_exception_handler:app --reload。
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)现在流程变成:
endpoint 里 raise HTTPException
↓
Starlette 捕获异常
↓
交给 http_exception_handler
↓
统一返回 {"code": ..., "message": ...}统一格式的作用
这样做的好处很直接:
错误响应结构统一
接口代码更干净
后续修改格式更集中区分 404 与服务器错误
404 表示资源或路径不存在;500 表示服务端未处理的错误。不要用 200 掩盖错误,也不要把内部异常细节直接给客户端。
除了按异常类型处理,也可以按状态码处理。
示例文件:03_not_found_server_error.py。运行:uvicorn 03_not_found_server_error:app --reload。
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
这里可区分为:
404:资源不存在
500:服务端代码本身出错debug 与错误显示
如果 debug=True,开发环境下 500 错误通常会显示更详细的信息。 如果 debug=False,更接近统一错误响应的实际效果。
保留正确状态码
即使响应体里写了“出错了”,状态码也不能乱写成 200。 客户端判断成功或失败时,通常会同时看状态码和响应体。
自定义业务异常
若特定业务错误需要独立携带信息,可以定义异常类并为它注册处理器。
有些错误不只是通用的 404 或 400,而是更具体的业务问题。
示例文件:04_business_exception.py。运行:uvicorn 04_business_exception:app --reload。
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)这里的流程是:
业务中发现没有权限
↓
raise ArticlePermissionError
↓
异常处理器捕获它
↓
返回统一 403 JSON业务异常的适用场景
适合这种情况:
错误有明确业务含义
希望单独处理
不想把所有问题都混成 HTTPException文章服务的错误处理
完整案例运行 05_article_error_handling.py 后,可用 error_handling_demo.py 发送请求,查看正常与错误请求的状态码和响应体。
最后让文章服务的错误都经过登记的处理器。
目标包括:
文章不存在时返回统一 404
缺少标题时返回统一 400
访问不存在路径时返回统一 404
程序出错时返回统一 500
业务异常时返回统一 403示例文件:05_article_error_handling.py。运行:uvicorn 05_article_error_handling:app --reload。
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、访问文章列表
访问:
http://127.0.0.1:8000/articles2、创建时缺少标题
向下面地址发送 POST JSON,但不传 title:
http://127.0.0.1:8000/articles3、查询不存在的文章
访问:
http://127.0.0.1:8000/articles/9994、修改不存在的文章
向下面地址发送 PUT:
http://127.0.0.1:8000/articles/9995、删除不存在的文章
向下面地址发送 DELETE:
http://127.0.0.1:8000/articles/9996、访问不存在的路径
例如访问:
http://127.0.0.1:8000/not-exists7、访问错误演示接口
访问:
http://127.0.0.1:8000/error-demo8、错误处理流程
这个案例把本章的核心点串了起来:
业务里发现错误
↓
raise 对应异常
↓
Starlette 捕获异常
↓
交给统一处理器
↓
返回结构清晰的 JSON 错误响应案例源码
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)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()