fastapi

对比 Flask,学习 FastAPI

本文假定你对 Flask 等 web framework 已经比较熟悉。不过多解释为什么要这么做,直接上干货。FastAPI 的一些优势大概有:类型检查、自动 swagger 文档、支持 asyncio、强大的依赖注入系统,我在 这个回答 里具体写过,就不展开了。

安装

注意一定要加上这个 [all], 免得后续缺什么依赖。

pip install fastapi[all]

Hello, World

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def hello():
    return "Hello, World!"

在命令行开启调试服务器:

uvicorn main:app --reload --port 8000

然后访问 http://localhost:8000 就可以看到 hello world 啦。更妙的是,访问 http://localhost:8000/docs 就可以看到自动生成,可以交互的 Swagger UI 文档了。

注意在上面的装饰器中,直接使用了 app.get 来定义 GET 访问的路径,其他的 post/put/delete 也都同理。Flask 在 1.x 中还必须使用 app.route,在 2.0 中才赶了上来。

同步和异步

# 同步函数
@app.get("/sync")
def hello():
    return "Hello, World!"

# 异步函数
@app.get("/async")
async def hello2():
    return "Hello, World!"

FastAPI 原生支持同时同步和异步函数,而且是完全一样的用法。使用异步函数肯定性能最好,但是有好多库,尤其是访问数据库的库用不了。使用同步函数的话,也不会把整个 Event Loop 锁死,而是会在线程池中执行。所以我个人的建议是,如果你已经习惯了写 Async Python,那么完全可以全部都写异步函数。如果你刚刚开始接触 asyncio 这一套,那么完全可以大部分都写同步函数,通过 profile 找到调用最多,或者对性能影响最大的地方,把这部分切换到异步函数。

请求参数

废话不多说,看请求参数都有哪些。

路径参数

@app.get("/{name})
def hello(name: str):
    return "Hello " + name

访问 http://127.0.0.1:8000/docs 就可以看到 OpenAPI 的文档了。

如果在文档中需要使用 Enum 的话,那么创建一个 Enum 的子类:

from enum import Enum

class ModelName(str, Enum):
   alexnet = "alexnet"
   resnet = "resnet"

@app.get("/models/{model_name}")
def get_model(model_name: ModelName):
    ...

值得注意的一点是,对于结尾的 “/”, FastAPI 不会强制要求匹配。如果你的请求和定义的不一样,那么 FastAPI 会 307 重定向到定义的那一个。比如定义了 app.get("/items/"), 而请求是 requests.get("/items"), 那么会 307 到 /items/. 但是如果部署在代理后面,重定向的地址可能不对,这个可能会引起问题,建议还是前后端统一一下。

另一个需要注意的地方,顺序是很重要的,FastAPI 按照定义的顺序来匹配路由。比如说你定义了两个路径 /users/me/users/{id}, 那么,一定要把 /users/me 放在前边,否则就会匹配到 users/{id}.

url query 参数

GET 参数是 FastAPI 和 Flask 不一样的地方。FastAPI 中统一用函数参数获取,这样也是为了更好地实现类型检查:

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

# GET /items/?skip=100&limit=10
@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
    return fake_items_db[skip : skip + limit]

对于 bool 类型,URL 中的 1/true/True/on/yes 都会被转化为 True, 其他都是 False.

http body 参数

POST 参数统一使用 json, 需要使用 PyDantic 的 BaseModel 来定义,只要参数是 BaseModel 类型的,FastAPI 就会从 body 中获取:

from pydantic import BaseModel

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

@app.post("/items/")
async def create_item(item: Item):
    return item

@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item):
    return {"item_id": item_id, **item.dict()}

如果不想定义 Model 验证,只想直接读取请求,那么可以使用 Request 对象,不过 Request 对象只有 async 版本,要想在同步函数中使用,需要一点点 work-around,这算是一个小小的坑吧:

from fastapi import Request, Depends

async def get_json(req: Request):
    return await req.json()

@app.get("/", response_model=None)
def get(req=Depends(get_json)):
    print(req)

其中用得到的 Depends 是 FastAPI 中非常强大的依赖注入系统,后面有时间了再详细讲。

Request 的文档详见:https://www.starlette.io/requests/, 参考:https://github.com/tiangolo/fastapi/issues/2574

Cookie 参数

使用 Cookie 来定义一个从 Cookie 中读取的参数:

from fastapi import Cookie, FastAPI

@app.get("/items")
def read_items(ads_id: Optional[str] = Cookie(None)):
    return {"ads_id": ads_id}

这里有个坑爹的地方,限于浏览器的安全模型,在生成的 Swagger UI 中 Cookie 参数是不能使用的。这是上游 Swagger/OpenAPI 项目的 bug(https://github.com/tiangolo/fastapi/issues/880), 所以在 FastAPI 这里也无解。

一个简单的 work-around 是:如果你有生成 Cookie 的 API, 那么先调用一下生成 Cookie 的 API, 再调用需要 Cookie 认证的 API 的时候就会自动附上这个 Cookie.

Header 参数

和 Cookie 是类似的。FastAPI 会自动把 “User-Agent” 转化成 user_agent 的形式。

from fastapi import FastAPI, Header

@app.get("/items/")
async def read_items(user_agent: Optional[str] = Header(None)):
    return {"User-Agent": user_agent}

参数验证

如果需要对参数进行验证,需要使用 Query, Path 等对象来指定,分别对应 GET 参数和路径:

from fastapi import FastAPI, Query, Path

@app.get("/items")
def read_items(q: Optional[str] = Query(None, min_length=3, max_length=50, regex="^fixedquery$"))
    ...

# 如果不使用默认值的话,可以使用特殊对象 `...`
def read_items(q: Optional[str] = Query(..., max_length=50))
# 数字可以使用 lt/le 或者 gt/ge 定义
def read_items(item_id: int = Path(..., gt=0, lt=100))

响应参数

需要在 app.get 中的 response_model 参数指定响应的 BaseModel:

from pydantic import BaseModel

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

请求的 body 和响应的 body 经常公用一个 Model, 有些敏感信息可能不适合在响应的 body 中返回,这时候可以使用 response_model_includeresponse_model_exclude 来指定需要包含或者需要过滤出去的字段。

@app.get(
    "/items/{item_id}/name",
    response_model=Item,
    response_model_include=["name", "description"],
    response_model_exclued=["password"]
)
async def read_item_name(item_id: str):
    return items[item_id]

使用 status_code 参数可以更改返回的状态码:

@app.get("/", status_code=201)
def get():
    ...

更新 Cookie

使用 Reponse 对象的 set_cookiedelete_cookie 方法实现增删 Cookie. 注意的是,只需要对 response 对象进行操作,不需要 return response 对象。

from fastapi import FastAPI, Response

@app.post("/cookie-and-object/")
def create_cookie(response: Response):
    response.set_cookie(key="fakesession", value="fake-cookie-session-value")
    response.delete_cookie(key)
    return {"message": "Come to the dark side, we have cookies"}

产生错误

直接 raise HTTPException 就好了

from fastapi import HTTPException

@app.get("/")
def hello():
    raise HTTPException(status_code=404, detail="item not found")

使用路由模块化

FastAPI 中的 router 相当于 Flask 中的 Blueprint, 用来切分应用到不同的模块。

# views/users.py
from fastapi import APIRouter

router = APIRouter()

@router.get("/me")
def myinfo():
    ...

# main.py
from fastapi import FastAPI
from views import users

app = FastAPI()

app.include_router(users.router, prefix="/users")

就到这里吧,简单写了下基本的请求响应和 Router 的使用,这是这个系列的第一篇,敬请期待后续。最后说一句,FastAPI 的官方文档非常详尽,值得一看。不过对于已经有 web 开发知识的人来说稍显啰嗦。

访问日志

FastAPI 默认的日志似乎有一些问题,不会显示出访问日志来

参考

FastAPI 官方文档