ThankNeko's Blog
Hoshinozora
2025-11-23
目录

FastAPI响应

# FastAPI响应

# 状态码

FastAPI可以在路径操作装饰器中使用 status_code 参数来指定默认响应状态码。

也可以在路径操作函数返回的 Response 对象中通过 status_code 参数指定响应状态码，会覆盖装饰器中指定的默认响应状态码。

from fastapi import status

# 指定状态码常量(推荐)
@app.post("/create", status_code=status.HTTP_201_CREATED)
async def create_item():
    return {"message": "Item created"}

# 直接指定状态码
@app.delete("/item/{id}", status_code=204)
async def delete_item(id: int):
    # 204 No Content，通常不返回body，直接返回None即可
    return
1
2
3
4
5
6
7
8
9
10
11
12

fastapi.status 中的常用状态码常量：

  • HTTP_200_OK
  • HTTP_201_CREATED
  • HTTP_204_NO_CONTENT
  • HTTP_400_BAD_REQUEST
  • HTTP_404_NOT_FOUND
  • HTTP_500_INTERNAL_SERVER_ERROR

# JSON响应

FastAPI会自动调用 .model_dump() 将路径操作函数返回的dict、list、Pydantic模型、基本类型转换为JSON响应。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    id: int
    name: str

@app.get("/user")
async def get_user():
    # 自动转为JSON，状态码200
    return {"id": 1, "name": "Alice"}

@app.get("/user2")
async def get_user2():
    # Pydantic模型也自动序列化
    return User(id=2, name="Bob")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

也可以通过 JSONResponse 对象手动返回JSON资源。

from fastapi.responses import JSONResponse

@app.get("/json")
async def json_resp():
    return JSONResponse(content={"hello": "world"}, status_code=200)
1
2
3
4
5

# HTML响应

可以通过 HTMLResponse 对象返回HTML页面资源。

from fastapi.responses import HTMLResponse

@app.get("/")
async def home():
    return HTMLResponse(content="<h1>Welcome</h1>", status_code=200)
1
2
3
4
5

# 文本响应

可以通过 PlainTextResponse 对象返回文本资源。

from fastapi.responses import PlainTextResponse

@app.get("/text")
async def text_resp():
    return PlainTextResponse("Hello, plain text!")
1
2
3
4
5

# 重定向响应

可以通过 RedirectResponse 对象返回重定向。

from fastapi.responses import RedirectResponse

@app.get("/old")
async def old_page():
    return RedirectResponse(url="/new", status_code=302)
1
2
3
4
5

# 文件响应

可以通过 RedirectResponse 对象返回文件资源。

from fastapi.responses import FileResponse

@app.get("/download")
async def download():
    return FileResponse(path="report.pdf", filename="report.pdf")
1
2
3
4
5

# 流式响应

可以通过 StreamingResponse 对象流式返回资源，可以传输视频、大文件等。

from fastapi.responses import StreamingResponse
import asyncio

async def fake_video_streamer():
    for i in range(10):
        yield f"chunk {i}\n".encode()
        await asyncio.sleep(0.5)

@app.get("/stream")
async def stream():
    return StreamingResponse(fake_video_streamer(), media_type="text/plain")
1
2
3
4
5
6
7
8
9
10
11

# 响应模型

可以在路径操作装饰器中使用 response_model 参数来指定响应模型，以约束或转换响应数据的结构。

# 定义响应模型
class DataIn(BaseModel):
    username: str
    password: str  # 敏感字段

class DataOut(BaseModel):
    username: str  # 不暴露 password
    
    # 可以重写model_dump方法来设置默认排除值为None的字段
    def model_dump(self, **kwargs):
        kwargs.setdefault("exclude_none", True)
        return super().model_dump(**kwargs)
    

# 指定响应模型
@app.post("/user", response_model=DataOut)
async def create_user(data: DataIn):
    # data内部包含username和password，但是因为response_model参数指向的DataOut中没有password字段，所以会自动过滤，只保留DataOut中有的字段
    return data
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

可以自动隐藏敏感字段，同时Swagger文档只显示 UserOut 结构。

# 抛出异常响应

我们可以在路径操作函数中直接抛出HTTPException异常，FastAPI会自动将其转换为错误响应并返回。

from fastapi import HTTPException

@app.get("/item/{id}")
async def read_item(id: int):
    if id <= 0:
        # 返回400响应
        raise HTTPException(
            status_code=400,  # 错误响应状态码
            detail="ID must be positive",  # 错误描述
            headers={"X-Error": "Invalid ID"}  # 可选: 添加响应头
        )
    return {"id": id}
1
2
3
4
5
6
7
8
9
10
11
12

# Response对象

可以通过 Response 对象来添加Headers和Cookies。

from fastapi import Response

# 设置Headers
@app.get("/with-header")
async def with_header(response: Response):
    response.headers["X-Custom"] = "Hello"
    response.headers["Cache-Control"] = "no-cache"
    return {"message": "Check headers!"}

# 设置Cookies
@app.post("/login")
async def login(response: Response):
    response.set_cookie(key="session_id", value="abc123", httponly=True, secure=True)
    return {"status": "logged in"}

# 删除Cookie
@app.post("/logout")
async def logout(response: Response):
    response.delete_cookie("session_id")
    return {"status": "logged out"}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 自定义异常处理器

为了保证接口对外输出结构一致，我们可以通过 @app.exception_handler(异常类) 装饰器来声明自定义异常处理器，以便在发生异常错误时，统一返回的JSON格式响应。

# 处理自定义异常

exceptions.py 将异常处理器都放到一个文件中。

# 自定义异常
class UserNotFoundException(Exception):
    def __init__(self, user_id: int):
        self.user_id = user_id

# # 通过装饰器在app中注册异常处理器函数
@app.exception_handler(UserNotFoundException)
async def user_not_found_handler(request: Request, exc: UserNotFoundException):
    # 还可以添加日志
    logger.error(f"未处理异常: {exc}")
    return JSONResponse(
        # 返回的响应状态码
        status_code=404,
        # 返回的JSON格式响应
        content={
            "code":-1,
            "message":f"用户 {exc.user_id} 不存在",
            "data": null
        }
    )
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

比默认的 {"detail": [{"loc": ..., "msg": ..., "type": ...}]} 更清晰。

# 统一错误响应格式

我们可以将响应格式定义为一个Pydantic模型，然后在使用时传入参数实例化，再转为字典格式即可。

from fastapi import FastAPI, Request, HTTPException
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

# 统一JSON响应格式
class JsonResponseModel(BaseModel):
    code: int
    message: str
    data: dict | None = None

# HTTP异常(如404、400等)
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content=JsonResponseModel(
            code=exc.status_code,
            message=exc.detail
        ).model_dump()
    )

# 请求验证错误，请求参数不符合模型时出现
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content=JsonResponseModel(
            code=422,
            message="数据校验失败",
            data={"errors": jsonable_encoder(
                # 列表推导式
                [{
                    "field": "->".join(str(loc) for loc in error["loc"][1:]),
                    "type": error["type"],
                    "message": error["msg"]
                } for error in exc.errors()]
            )}
        ).model_dump()
    )

# 全局异常(500)
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content=JsonResponseModel(code=500, message="服务器内部错误").model_dump()
    )
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

# 手动注册异常处理函数

手动注册异常处理器可以通过 app.add_exception_handler(异常类, 处理函数) 来显式注册，这种方式更灵活，适合动态注册、模块化组织。

app.add_exception_handler(
    exc_class_or_status_code,  # 异常类或HTTP状态码(如404)
    exception_handler          # 处理函数
)
1
2
3
4
#Python#模块#FastAPI#WEB#后端
FastAPI请求
FastAPI路由

← FastAPI路由→

最近更新
01
Vue路由
12-09
02
FastAPI实现用户管理
11-23
03
Tortoise ORM
11-23
更多文章>
Theme by Vdoing | Copyright © 2022-2026 Hoshinozora | MIT License
湘ICP备2022022820号-1