FastAPI中Service Unavailable自定义错误消息不生效如何解决？

你写的FastAPI异常处理器没捕获到503错误，核心原因是：这个503是Uvicorn层面直接返回的，根本没把请求传到FastAPI应用里。

当你用--limit-concurrency 10启动Uvicorn时，这个并发限制由Uvicorn底层限流逻辑控制。一旦请求并发数超过阈值，Uvicorn会直接拒绝请求并返回503，整个过程不会经过FastAPI的路由和异常处理流程，所以你的自定义ServiceUnavailableException处理器完全碰不到这个错误。

推荐解决方案：改用FastAPI层面的限流 #

放弃Uvicorn的--limit-concurrency，用FastAPI生态的限流工具（比如slowapi）实现并发限制，这样触发的限流异常能被FastAPI捕获，轻松返回自定义响应。

步骤1：安装依赖 #

pip install slowapi limits

步骤2：编写代码 #

from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

# 初始化限流器，按客户端IP限制，默认每秒最多处理10个请求
limiter = Limiter(key_func=get_remote_address, default_limits=["10/second"])
app = FastAPI()

# 将限流器绑定到FastAPI应用
app.state.limiter = limiter
# 注册默认的限流异常处理器，再替换成自定义的
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# 自定义限流触发时的响应
@app.exception_handler(RateLimitExceeded)
async def custom_service_unavailable_handler(request: Request, exc: RateLimitExceeded):
    return JSONResponse(
        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
        content={"message": "服务繁忙，请稍后再试"}
    )

# 给路由添加限流规则（这里用默认规则，也可以单独指定）
@app.get("/")
@limiter.limit("10/second")
async def read_root(request: Request):
    return {"Hello": "World"}

优势 #

• 限流逻辑在FastAPI层面，异常能被完全捕获，自定义响应灵活；
• 可以给不同路由设置不同的限流规则；
• 不依赖Uvicorn的内部实现，版本兼容性更好。

备选方案：修改Uvicorn的503响应（不推荐） #

如果一定要用Uvicorn的--limit-concurrency，需要修改Uvicorn的底层错误处理逻辑，这个方法依赖Uvicorn内部实现，版本更新后可能失效。

示例代码（自定义Uvicorn Worker） #

from fastapi import FastAPI
from uvicorn.workers import UvicornWorker
from uvicorn.limits import ConcurrencyLimitExceeded
from starlette.responses import JSONResponse
import asyncio

app = FastAPI()

@app.get("/")
async def read_root():
    return {"Hello": "World"}

# 自定义Worker，重写并发限制异常的处理
class CustomUvicornWorker(UvicornWorker):
    async def handle_async(self, sock, addr):
        try:
            await super().handle_async(sock, addr)
        except ConcurrencyLimitExceeded:
            # 构造自定义503响应
            scope = {
                "type": "http",
                "method": "GET",
                "path": "/",
                "headers": [],
                "query_string": b"",
                "server": ("0.0.0.0", 8080),
                "client": addr
            }

            async def receive():
                return {"type": "http.request"}

            async def send(message):
                if message["type"] == "http.response.start":
                    # 发送状态码和响应头
                    await sock.send(b"HTTP/1.1 503 Service Unavailable\r\n")
                    await sock.send(b"Content-Type: application/json\r\n")
                    await sock.send(b"Content-Length: 44\r\n")
                    await sock.send(b"\r\n")
                elif message["type"] == "http.response.body":
                    # 发送响应内容
                    await sock.send(b'{"message": "服务繁忙，请稍后再试"}')
                    await sock.close()

            response = JSONResponse(
                status_code=503,
                content={"message": "服务繁忙，请稍后再试"}
            )
            await response(scope, receive, send)

启动命令 #

uvicorn main:app --host 0.0.0.0 --port 8080 --workers 2 --limit-concurrency 10 --worker-class main.CustomUvicornWorker

内容的提问来源于stack exchange，提问作者grayaii