Posted on:
Last modified:
本文假定你对 Flask 等 web framework 已经比较熟悉。不多解释为什么要这么做,直接上干货。 FastAPI 的一些优势大概有:类型检查、自动 swagger 文档、支持 asyncio、强大的依赖注入系统, 我在 这个回答 里具体写过, 就不展开了。
注意,最好加上 [all]
, 免得后续缺什么依赖。
pip install fastapi[all]
# 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 啦。更妙的是,访问 /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 这一套,那么完全可以大部分都写同步函数,通过 profiling 找到对性能影响最大的地方,把这部分切换到异步函数。
废话不多说,看请求参数都有哪些。
@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/")
, 而请求是
httpx.get("/items")
, 那么会 307 到 /items/
. 但是如果部署在代理后面,重定向的地址
可能不对,这个可能会引起问题,建议还是前后端统一一下。
另一个需要注意的地方,顺序是很重要的,FastAPI 按照定义的顺序来匹配路由。比如你定义了两个
路径 /users/me
和 /users/{id}
, 一定要把 /users/me
放在前边,否则就会匹配到 users/{id}
.
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.
和 GET 参数类似,但是需要使用 Form 表明是表单参数。
@app.post("/login")
async def login(username: str=Form(...), password: str=Form(...)):
auth(username, password)
return username
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 是类似的。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}
使用 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, 所以在 FastAPI 这里也无解。
一个简单的 work-around 是:如果你有生成 Cookie 的 API, 那么先调用一下生成 Cookie 的 API, 再调用需要 Cookie 认证的 API 的时候就会自动附上这个 Cookie.
使用 Response 对象的 set_cookie
和 delete_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"}
如果需要对参数进行验证,需要使用 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))
使用 Body(...)
from fastapi import Body, FastAPI
app = FastAPI()
@app.post('/test')
async def update_item(
payload: dict = Body(...)
):
return payload
需要在 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_include
和 response_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():
...
直接 raise HTTPException 就好了
from fastapi import HTTPException
@app.get("/")
def hello():
raise HTTPException(status_code=404, detail="item not found")
就到这里吧,简单写了下基本的请求响应和异常的使用,这是这个系列的第一篇,敬请期待后续。 最后,FastAPI 的官方文档非常详尽,值得一看。不过对于已经有 web 开发知识的人来说稍显啰嗦。
© 2016-2022 Yifei Kong. Powered by ynotes
All contents are under the CC-BY-NC-SA license, if not otherwise specified.
Opinions expressed here are solely my own and do not express the views or opinions of my employer.
友情链接: MySQL 教程站