响应模型 response_model 等
响应状态码、Response、JSONResponse等
错误处理HTTPException
后台任务

响应模型

response_model

在装饰器方法(get，post 等)中的参数response_model

  • 将输出数据转换为其声明的类型
  • 校验数据，将输出数据限制在该模型定义内
  • 在 OpenAPI 的路径操作中为响应添加一个 JSON Schema，并在自动生成文档系统中使用
from typing import List, Optional
import uvicorn

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: List[str] = []


@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    return item                 # 修改为 return {"item": item} 看看效果

if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

from typing import Optional

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Optional[str] = None


class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Optional[str] = None


@app.post("/user/", response_model=UserOut)   # response_model 声明为了不包含密码的 UserOut 模型
async def create_user(user: UserIn):
    return user                  # 即使返回包含密码的相同输入的用户user，FastAPI 也会过滤掉未在输出模型中声明的所有数据

if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

response_model_exclude_xx

  • response_model_exclude_unset：设置为True，不返回未被设置的值
  • response_model_exclude_defaults：设置为True，不返回包含默认值的值
  • response_model_exclude_none：设置为True，不返回值为null的值
from typing import List, Optional
import uvicorn

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float = 12.3
    tax: Optional[float] = None
    tags: List[str] = []


# 修改为esponse_model_exclude_unset、response_model_exclude_defaults看看效果
@app.post("/items/", response_model=Item, response_model_exclude_none=True)
async def create_item(item: Item):
    return item

if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

使用postman访问127.0.0.1:8000/items发送{"name":"jerry"}看看效果

response_model_include 和 response_model_exclude

  • 它们接收一个由属性名称 str 组成的 set(如果使用 list 或 tuple仍然有效(FastAPI 仍会将其转换为 set))，返回的数据将包含(include)或排除(exclude) set 中的这些属性
from typing import Optional

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Optional[str] = None

# 一定要同时声明响应模型response_model才生效
@app.post("/user/", response_model=UserIn, response_model_exclude={"password"})
async def create_user(user: UserIn):
    return user

if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

定义模型基类减少重复代码

from typing import Optional

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: Optional[str] = None


class UserIn(UserBase):
    password: str


class UserOut(UserBase):
    pass


class UserInDB(UserBase):
    hashed_password: str


# 模拟密码hash
def fake_password_hasher(raw_password: str):
    return "supersecret" + raw_password


# 密码hash后组装成UserInDB
def fake_save_user(user_in: UserIn):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)
    print("User saved! ..not really")
    return user_in_db


@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):
    user_saved = fake_save_user(user_in)
    return user_saved


if __name__ == '__main__':
    # user_in = UserIn(username="jerry", password="123458", email="jerry@monkey.com")
    # user_in_db = fake_save_user(user_in)
    # print(user_in_db, type(user_in_db))
    uvicorn.run("main:app", reload=True)

响应

我们知道 FastAPI 默认使用 JSONResponse 返回响应(使用jsonable_encoder将这些类型的返回值转换成 JSON 格式)

  • 可以使用 jsonable_encoder 将数据转换为兼容 JSON 的类型

响应状态码

  • 状态码：一个表示 HTTP 状态码的数字如200、301、404等，也可使用 fastapi.status 的便捷变量如HTTP_200_OK、HTTP_301_MOVED_PERMANENTLY、HTTP_404_NOT_FOUND等
  1. 在装饰器方法@app.get()、@app.post()等中的参数status_code
  2. 响应Response类中接收的参数status_code，如return Response(status_code=201)
  3. 在路径操作中声明 Response 类型的参数如 response，然后在代码逻辑中设置status_code，如response.status_code=status.HTTP_201_CREATED

路径操作中声明的response: Response 还可以用来添加/设置 response headers(如response.headers["X-Token"] = "screct token")和 response cookies(response.set_cookie)

Response

  • 全部的响应都继承自主类 Response 如 JSONResponse、HTMLResponse、PlainTextResponse、RedirectResponse、StreamingResponse、FileResponse、UJSONResponse及性能更好的ORJSONResponse等
  • 可以返回任意 Response 、任意 Response 的子类或一个自定义的 Response 子类
  • Response 类接受如下参数：
    • content：str 或者 bytes
    • status_code ：int 类型的 HTTP 状态码
    • headers：字符串组成的 dict，可添加自定义response headers如 X-Token
    • media_type：媒体类型的 str，比如 "text/html"
import uvicorn
from fastapi import FastAPI, Response

app = FastAPI()


@app.get("/legacy/")
def get_legacy_data():
    data = """<?xml version="1.0"?>
    <shampoo>
    <Header>
        Apply shampoo here.
    </Header>
    <Body>
        You'll have to use soap here.
    </Body>
    </shampoo>
    """
    return Response(content=data, media_type="application/xml", status_code=201, headers={"x-token": "jerry"})


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

HTMLResponse 和 PlainTextResponse

  • HTMLResponse：接受文本或字节并返回 HTML 响应
  • PlainTextResponse：接受文本或字节并返回纯文本响应
    • 方式1：直接返回Response类/子类的实例
    • 方式2：将Response类/子类作为路径操作的response_class参数传入（参数response_class来定义响应的媒体类型）
import uvicorn
from fastapi import FastAPI
from fastapi.responses import HTMLResponse, PlainTextResponse

app = FastAPI()


# 方式1
@app.get("/items/")
async def read_items():
    html_content = """
    <html>
        <head>
            <title>Some HTML in here</title>
        </head>
        <body>
            <h1>Look ma! HTML!</h1>
        </body>
    </html>
    """
    return HTMLResponse(content=html_content, status_code=200)


# 方式2
@app.get("/", response_class=PlainTextResponse)
async def main():
    return "Hello World"


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

RedirectResponse

  • 返回 HTTP 重定向。默认情况下使用 307 状态代码（临时重定向）
import uvicorn
from fastapi import FastAPI
from fastapi.responses import RedirectResponse

app = FastAPI()


@app.get("/typer")
async def redirect_typer():
    return RedirectResponse("https://baidu.com")   # 接收重定向的url参数


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

StreamingResponse 和 FileResponse

  • StreamingResponse：采用异步生成器或普通生成器/迭代器，然后流式传输响应主体，例子及更多细节见官方文档
  • FileResponse：异步传输文件作为响应
import uvicorn

from fastapi import FastAPI
from fastapi.responses import FileResponse

some_file_path = "/Users/xmly/Desktop/jerry.mp4"  # large-video-file path
app = FastAPI()

@app.get("/")
async def main():
    return FileResponse(some_file_path)

if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

错误处理

  • 向客户端返回 HTTP 错误响应，可以raise HTTPException
  • 添加自定义响应头：有些场景下要为 HTTP 错误添加自定义响应头，在HTTPException 中的 参数headers中设置
import uvicorn

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}


@app.get('/items/{item_id}')
async def read_items(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="item not found", headers={"X-Error": "There goes my error"})
    return {"item": items[item_id]}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

detail参数类型是tying.Any，也就是说detail不仅限于 str，还支持传递 dict、list 等数据结构

自定义异常处理器

  • 使用@app.exception_handler()添加自定义异常控制器
import uvicorn

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

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}


class MyException(Exception):
    def __init__(self, name: str):
        self.name = name


@app.exception_handler(MyException)
async def my_exception_handler(request: Request, exc: MyException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} not found"},
    )


@app.get('/items/{item_id}')
async def read_items(item_id: str):
    if item_id not in items:
        raise MyException(item_id)
    return {"item": items[item_id]}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

覆盖默认异常处理器

  • 覆盖默认异常处理器时需要导入 RequestValidationError，并用 @app.excption_handler(RequestValidationError)装饰异常处理器
  • 相关例子见官方文档

后台任务

  • 不需要等待任务完成才收到响应，使用BackgroundTasks
  • BackgroundTasks.add_task
import time
import uvicorn
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def bg_task(t):
    time.sleep(t)
    print("bg task done")

@app.get('/items/{item_id}')
async def read_items(item_id: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(bg_task, t=10)
    return {"item": item_id}


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)
  • 在依赖注入系统使用后台任务可参看依赖注入章节