(轉)從零開始學FastAPI（1）-簡明的部分官方文檔抽取學習篇

本文轉載自查看原文 2020-06-27 15:09 801

前言之前線上業務一直是使用Bottle和Flask兩個框架來編寫API接口。bottle和Flask它們都是一些同步的框架，而支持異步框架的最新晉的一個非常不錯的角：之前在青南大大的文章里也了解到了它的奇異之處，處於對異步框架的學習興趣，我還是決定也開始使用Fastapi試一試，雖然…

前言

之前線上業務一直是使用Bottle和Flask兩個框架來編寫API接口。bottle和Flask它們都是一些同步的框架，而支持異步框架的最新晉的一個非常不錯的角：之前在青南大大的文章里也了解到了它的奇異之處，處於對異步框架的學習興趣，我還是決定也開始使用Fastapi試一試，雖然此前也用過Tornado,但是也僅限於使用同步的框架。

本來想上手試一試看看sanic,處於獵奇心，還是鼓搗以下這個Fastapi！

學一個新的框架最好的方式當然就是框架笨的提供的官方文檔了！

文獻資料: fastapi.tiangolo.com

源代碼: github.com/tiangolo/fa…

參考資料：www.jianshu.com/p/94710ed35…

開始擼碼

注意事項，因為FastAPI僅支持Python3.6+的API，所以需要再Python3.6+的環境進行學習實踐！

說明因為我是再winddos環境下的，僅用於開放的調試，真正到線上的，肯定都是在linux環境下運行才能到達AWGI的相關的極致的性能！

1：依賴庫的安裝

或者：

FastAPI - 是一個現代的，快速（高性能）python web框架

pip install fastapi

uvicorn - 主要用於加載和提供應用程序的服務器.

pip install uvicorn

2：第一個Hello World

 
                 import uvicorn 
                
                 from fastapi import FastAPI 
                
                 app = FastAPI() 
                
                 @app.get("/") 
                
                 async def root(): 
                
                 return {"message": "Hello World"} 
                
                 if __name__ == '__main__': 
                
                 uvicorn.run(app=app) 
                
                 復制代碼

深入到uvicorn.run(app=app)方法里面，看到一個：

 
                 def run(app, **kwargs): 
                
                 config = Config(app, **kwargs) 
                
                 server = Server(config=config) 
                
                 if (config.reload or config.workers > 1) and not isinstance(app, str): 
                
                 logger = logging.getLogger("uvicorn.error") 
                
                 logger.warn( 
                
                 "You must pass the application as an import string to enable 'reload' or 'workers'." 
                
                 ) 
                
                 sys.exit(1) 
                
                 if config.should_reload: 
                
                 sock = config.bind_socket() 
                
                 supervisor = StatReload(config, target=server.run, sockets=[sock]) 
                
                 supervisor.run() 
                
                 elif config.workers > 1: 
                
                 sock = config.bind_socket() 
                
                 supervisor = Multiprocess(config, target=server.run, sockets=[sock]) 
                
                 supervisor.run() 
                
                 else: 
                
                 server.run() 
                
                 復制代碼

再深入到 config = Config(app, **kwargs)里面，就看到一些很多的相關的配置信息項：

 
                 class Config: 
                
                 def __init__( 
                
                 self, 
                
                 app, 
                
                 host="127.0.0.1", 
                
                 port=8000, 
                
                 uds=None, 
                
                 fd=None, 
                
                 loop="auto", 
                
                 http="auto", 
                
                 ws="auto", 
                
                 lifespan="auto", 
                
                 env_file=None, 
                
                 log_config=LOGGING_CONFIG, 
                
                 log_level=None, 
                
                 access_log=True, 
                
                 use_colors=None, 
                
                 interface="auto", 
                
                 debug=False, 
                
                 reload=False, 
                
                 reload_dirs=None, 
                
                 workers=None, 
                
                 proxy_headers=True, 
                
                 forwarded_allow_ips=None, 
                
                 root_path="", 
                
                 limit_concurrency=None, 
                
                 limit_max_requests=None, 
                
                 backlog=2048, 
                
                 timeout_keep_alive=5, 
                
                 timeout_notify=30, 
                
                 callback_notify=None, 
                
                 ssl_keyfile=None, 
                
                 ssl_certfile=None, 
                
                 ssl_version=SSL_PROTOCOL_VERSION, 
                
                 ssl_cert_reqs=ssl.CERT_NONE, 
                
                 ssl_ca_certs=None, 
                
                 ssl_ciphers="TLSv1", 
                
                 headers=None, 
                
                 ): 
                
                 復制代碼

所以還可以添加的參數可以看上面的幾個配置的選項的信息來填：

於是乎還可以修改為：

 
                 import uvicorn 
                
                 from fastapi import FastAPI 
                
                 app = FastAPI() 
                
                 @app.get("/") 
                
                 async def root(): 
                
                 return {"message": "Hello 454533343433World"} 
                
                 if __name__ == '__main__': 
                
                 uvicorn.run(app=app, host="127.0.0.1", port=8000, reload=True, debug=True) 
                
                 復制代碼

發現本來想熱更新代碼，結果吶？有告警信息提示：

 
                 WARNING:  You must pass the application as an import string to enable 'reload' or 'workers'. 
                
                 復制代碼

翻譯過來就是說：警告：必須將應用程序作為導入字符串傳遞，才能啟用“重新加載” 然后呢：我修改為：

 
                 uvicorn.run(app='app', host="127.0.0.1", port=8000, reload=True, debug=True) 
                
                 復制代碼

又提示：

 
                 ERROR:    Error loading ASGI app. Import string "app" must be in format "< 
                 module 
                 >:< 
                 attribute 
                 >". 
                
                 復制代碼

好吧，我再看看官方文檔說是：

在命令行下是需要：模塊加app名稱：剛好上面的錯誤提示也是說需要:

好吧，了然：

 
                 uvicorn.run(app='main:app', host="127.0.0.1", port=8000, reload=True, debug=True) 
                
                 復制代碼

這樣之后就可以啟動熱更新重啟服務了！

然后訪問地址，正常的獲取到接口返回的消息體了：

然后按官網文檔查閱API文檔交互地址：

http://127.0.0.1:8000/docs

http://127.0.0.1:8000/redoc

路由方法有 GET, POST, PUT, PATCH, DELETE 和 OPTIONS。

 
                 @app.post("/") 
                
                 @app.put("/") 
                
                 @app.delete("/") 
                
                 @app.get("/") 
                
                 @app.options("/") 
                
                 @app.head("/") 
                
                 @app.patch("/") 
                
                 @app.trace("/") 
                
                 async def root(): 
                
                 return {"message": "Hello 454533333343433World"} 
                
                 復制代碼

3：路由Route上參數獲取和校驗

一般我們的路由分會靜態和動態，靜態路由就是參數是固定寫死，也就是訪問地址是寫死的，而動態地址，就是需要動態的生成，類似簡書的博文的地址94710ed35b92就是動態，其實和Bottle和Flask一樣。

www.jianshu.com/p/94710ed35…

 
                 from fastapi import FastAPI 
                
                 app = FastAPI() 
                
                 @app.get("/items/{item_id}") 
                
                 async def read_item(item_id): 
                
                 return {"item_id": item_id} 
                
                 復制代碼

上述的示例代碼中的item_id 就是一個動態的參數，你可以隨意傳一個進來。

然后就是和bottle一樣也可以對傳入的參數進行數據驗證的定義：如：

 
                 from fastapi import FastAPI 
                
                 app = FastAPI() 
                
                 @app.get("/items/{item_id}") 
                
                 async def read_item(item_id: int): 
                
                 return {"item_id": item_id} 
                
                 復制代碼

item_id: int 這種情況item_id必須是可以轉為int類似的數據，否則，肯定會報錯！

關於路由覆蓋問題：如下兩個路由地址：

 
                 from fastapi import FastAPI 
                
                 app = FastAPI() 
                
                 @app.get("/users/me") 
                
                 async def read_user_me(): 
                
                 return {"user_id": "the current user"} 
                
                 @app.get("/users/{user_id}") 
                
                 async def read_user(user_id: str): 
                
                 return {"被優先匹配到：": user_id} 
                
                 復制代碼

上面兩個路由同時存在的話，則/users/{user_id} 會覆蓋/users/me！

3.1 查詢路徑參數和參數校驗

關於查詢參數，其實就是在使用POSTMAN 提交的時候的參數信息: 如：

http://127.0.0.1:8000/items/?skip=0&limit=10

skip=0&limit就是所謂的查詢參數。

 
                 from fastapi import FastAPI 
                
                 app = FastAPI() 
                
                 fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}] 
                
                 @app.get("/items/") 
                
                 async def read_item(skip: int = 0, limit: int = 10): 
                
                 return fake_items_db[skip : skip + limit] 
                
                 復制代碼

第一種訪問的情況：

第二種訪問情況：

第三種訪問情況：

3.2 多路徑和查詢參數

所謂的多路徑和查詢參數就是URL上包含了有動態的參數，還有需要通過&分隔符提交的參數，這情況，通常再GET提交的中也很常見，那么如何處理吶？

 
                 from fastapi import FastAPI 
                
                 app = FastAPI() 
                
                 @app.get("/users/{user_id}/items/{item_id}") 
                
                 async def read_user_item( 
                
                 user_id: int, item_id: str, q: str = None, short: bool = False 
                
                 ): 
                
                 item = {"item_id": item_id, "owner_id": user_id} 
                
                 if q: 
                
                 item.update({"q": q}) 
                
                 if not short: 
                
                 item.update( 
                
                 {"description": "This is an amazing item that has a long description"} 
                
                 ) 
                
                 return item 
                
                 復制代碼

請求：

http://127.0.0.1:8000/users/123456/items/items_xinxiid/?q=assa&short=True

請求：

http://127.0.0.1:8000/users/123456/items/items_xinxiid/?q=assa&short=False

3.3 路徑參數和查詢參數的必選和可選

參數的可選和必選主要是通過是否給默認值來決定的：

如：

 
                 @app.get("/items/{item_id}") 
                
                 async def read_user_item(item_id: str, needy: str): 
                
                 item = {"item_id": item_id, "needy": needy} 
                
                 return item 
                
                 復制代碼

上述的代碼中 needy 沒有給與默認的值，當個沒提交這個值的時候，會提示錯誤：

還可以定義可選參數和必選的參數的提交類型：其中還可以使用Optional來定義需要提交的數據類型：如：

 
                 from typing import Optional 
                
                 @app.get("/items/{item_id}") 
                
                 async def read_user_item(item_id: str, limit: Optional[int] = None): 
                
                 item = {"item_id": item_id, "limit": limit} 
                
                 return item 
                
                 復制代碼

我們把查詢參數limit規定為了int類型，但是它是可選的的參數，設置為了None:

3.4 路徑參數的枚舉

 
                 import uvicorn 
                
                 from fastapi import FastAPI 
                
                 from enum import Enum 
                
                 class ModelName(str, Enum): 
                
                 alexnet = "alexnet" 
                
                 resnet = "resnet" 
                
                 lenet = "lenet" 
                
                 app = FastAPI() 
                
                 @app.get("/model/{model_name}") 
                
                 async def get_model(model_name: ModelName): 
                
                 if model_name == ModelName.alexnet: 
                
                 return {"model_name": model_name, "message": "Deep Learning FTW!"} 
                
                 if model_name.value == "lenet": 
                
                 return {"model_name": model_name, "message": "LeCNN all the images"} 
                
                 return {"model_name": model_name, "message": "Have some residuals"} 
                
                 if __name__ == '__main__': 
                
                 uvicorn.run(app='main2:app', host="127.0.0.1", port=8000, reload=True, debug=True) 
                
                 復制代碼

通過訪問地址：

http://127.0.0.1:8000/model/alexnet

3.5 查詢參數Query參數的其他校驗

在以前通常是使用wtform來定義提交的字段信息的類似或可選或長度類型。在Fastapi里面，我們是通過： from fastapi import FastAPI, Query 中的Query來定義，如：

 
                 from fastapi import FastAPI, Query 
                
                 app = FastAPI() 
                
                 @app.get("/items/") 
                
                 async def read_items(q: str = Query(None, min_length=3,max_length=50),regex="^fixedquery$"): 
                
                 results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]} 
                
                 if q: 
                
                 results.update({"q": q}) 
                
                 return results 
                
                 復制代碼

q: q: str = Query(None, min_length=3,max_length=50),regex="^fixedquery$") 意思是：q參數是可選的參數，但是如果填寫的話，最大長度必須是小於50內,且最小的長度必須大於3：且需要符合regex的匹配

當然None可以修改為其他默認值，可以寫如：

q: q: str = Query('xiaozhong', min_length=3,max_length=50),regex="^fixedquery$")

不傳q的情況下:

http://127.0.0.1:8000/items/

傳q的情況下且長度大於50:

http://127.0.0.1:8000/items/

傳q的情況下且長度小於3:

http://127.0.0.1:8000/items/?q=4

查詢參數Query的參數正則校驗

3.6 查詢參數Query參數多值列表

一般在我們的接口中很少說同一個參數提交多個值如：

http://localhost:8000/items/?q=foo&q=bar

但也不排查這種情況的存在，所以也可以定義我們的參數類似必須是列表的形式：

 
                 from typing import List 
                
                 from fastapi import FastAPI, Query 
                
                 app = FastAPI() 
                
                 @app.get("/items/") 
                
                 async def read_items(q: List[str] = Query(["foo", "bar"])): 
                
                 #  
                 <!--也可以使用list直接代替List[str]:--> 
                
                 query_items = {"q": q} 
                
                 return query_items 
                
                 復制代碼

默認值：

非默認值：

3.7 路徑參數的其他校驗方式

對於查詢參數可以通過Query，同樣對於路徑參數也可以使用Fastapi自帶的Path來進行校驗。

 
                 from fastapi import FastAPI, Path 
                
                 app = FastAPI() 
                
                 @app.get("/items/{item_id}") 
                
                 async def read_items( 
                
                 q: str, item_id: int = Path(..., title="The ID of the item to get") 
                
                 ): 
                
                 results = {"item_id": item_id} 
                
                 if q: 
                
                 results.update({"q": q}) 
                
                 return results 
                
                 復制代碼

對於路徑參數校驗中，還可以對item_id進行大於或等於的校驗如：

 
                 from fastapi import FastAPI, Path 
                
                 app = FastAPI() 
                
                 @app.get("/items/{item_id}") 
                
                 async def read_items( 
                
                 *, item_id: int = Path(..., title="The ID of the item to get", ge=1), q: str ): 
                
                 results = {"item_id": item_id} 
                
                 if q: 
                
                 results.update({"q": q}) 
                
                 return results 
                
                 復制代碼

在上面代碼意思是，當ge = 1時，item_id必須是整數“ g大於或等於e等於1”。

3.8 參數提交的Request Body

一般對於Request Body不會通過get提交，對於get提交的參數一般稱為是查詢參數。所以，如果是通過POTS,PUT等方式提交的參數信息，我們一般是放到Request Body來提交到我們的后端。

對於如何接收和校驗請求體，FastApi提供的形式是使用：from pydantic import BaseModel

示例如下：

 
                 from fastapi import FastAPI 
                
                 from pydantic import BaseModel 
                
                 class Item(BaseModel): 
                
                 name: str 
                
                 description: str = None 
                
                 price: float 
                
                 tax: float = None 
                
                 app = FastAPI() 
                
                 @app.post("/items/") 
                
                 async def create_item(item: Item): 
                
                 return item 
                
                 復制代碼

在上面的模型中我，定義如果提交的Item它必須是怎么樣的一個格式，比如name是必選字段，description是可選且默認為None, price是必選，且需要是float類型的，tax是可須且默認為None。

那客戶端如何提交上面那些參數吶？

嘗試提交參數什么都不寫的情況下：

使用JSON格式提交參數的情況下：

故意提交錯誤參數格式請求：

3.8 Request Body 和 Query 和 Path的混合

在設計一些API過程中難免的可能也會需要綜合遇到上述的一些混搭的組合，需要同時多個參數的提交和獲取

那么我們通常接收這次參數的話一般怎么接收吶？

示例代碼如：

 
                 from fastapi import FastAPI, Path 
                
                 from pydantic import BaseModel 
                
                 app = FastAPI() 
                
                 class Item(BaseModel): 
                
                 name: str 
                
                 description: str = None 
                
                 price: float 
                
                 tax: float = None 
                
                 @app.put("/items/{item_id}") 
                
                 async def update_item( 
                
                 *, 
                
                 item_id: int = Path(..., title="The ID of the item to get", ge=0, le=1000), 
                
                 q: str = None, 
                
                 item: Item = None, 
                
                 ): 
                
                 results = {"item_id": item_id} 
                
                 if q: 
                
                 results.update({"q": q}) 
                
                 if item: 
                
                 results.update({"item": item}) 
                
                 return results 
                
                 復制代碼

通過之前的學習，其實也很簡單道理也還是一樣，如上的示例請求的話：

3.9 多個Request Body的提交

更復雜的業務其實會存在多體的Boay的提交，之前做的商城下單里面，客戶端有可能就會同時提交多個實體的對象信息到后端，如訂單實體，地址實體，商品信息實體等。

那么在Fastapi如何接受多個Body實體吶？通常以前的話，在bottle,通常直接的request.body 或 request.json就可以獲取客戶端部提交的信息了。

在Fastapi假設客戶端提交的參數是這樣的形式：

 
                 { 
                
                 "item": { 
                
                 "name": "Foo", 
                
                 "description": "The pretender", 
                
                 "price": 42.0, 
                
                 "tax": 3.2 
                
                 }, 
                
                 "user": { 
                
                 "username": "dave", 
                
                 "full_name": "Dave Grohl" 
                
                 } 
                
                 } 
                
                 復制代碼

那如何的接收處理吶？

 
                 from fastapi import FastAPI 
                
                 from pydantic import BaseModel 
                
                 app = FastAPI() 
                
                 class Item(BaseModel): 
                
                 name: str 
                
                 description: str = None 
                
                 price: float 
                
                 tax: float = None 
                
                 class User(BaseModel): 
                
                 username: str 
                
                 full_name: str = None 
                
                 @app.put("/items/{item_id}") 
                
                 async def update_item(*, item_id: int, item: Item, user: User): 
                
                 results = {"item_id": item_id, "item": item, "user": user} 
                
                 return results 
                
                 復制代碼

這種情況，其實就是客戶端提交多個實體對象。那可以定義多個模型對象即可。fastapi它會自動幫你處理提取信息。

如果另外再假設：

在Fastapi假設客戶端提交的參數是這樣的形式：

 
                 { 
                
                 "item": { 
                
                 "name": "Foo", 
                
                 "description": "The pretender", 
                
                 "price": 42.0, 
                
                 "tax": 3.2 
                
                 }, 
                
                 "user": { 
                
                 "username": "dave", 
                
                 "full_name": "Dave Grohl" 
                
                 }, 
                
                 "importance": 5 
                
                 } 
                
                 復制代碼

其實這種可能也不是不存在滴，那如何的讀取解析importance參數吶？既然參數有Query 和 Path，當然也會有 Body 。

 
                 from fastapi import Body, FastAPI 
                
                 from pydantic import BaseModel 
                
                 app = FastAPI() 
                
                 class Item(BaseModel): 
                
                 name: str 
                
                 description: str = None 
                
                 price: float 
                
                 tax: float = None 
                
                 class User(BaseModel): 
                
                 username: str 
                
                 full_name: str = None 
                
                 @app.put("/items/{item_id}") 
                
                 async def update_item( 
                
                 *, item_id: int, item: Item, user: User, importance: int = Body(...,gt=0) 
                
                 ): 
                
                 results = {"item_id": item_id, "item": item, "user": user, "importance": importance} 
                
                 return results 
                
                 復制代碼

上面的代碼中我們引入了Body 並且在importance: int = Body(...)進行處理和提取：

如果另外再假設,客戶端提交的是一個單體對象內嵌的話，我們需要怎么處理？：

 
                 { 
                
                 "item": { 
                
                 "name": "Foo", 
                
                 "description": "The pretender", 
                
                 "price": 42.0, 
                
                 "tax": 3.2 
                
                 } 
                
                 } 
                
                 復制代碼

FastAPI提供了一個：

item: Item = Body(..., embed=True) 具體如下：

 
                 from fastapi import Body, FastAPI 
                
                 from pydantic import BaseModel 
                
                 app = FastAPI() 
                
                 class Item(BaseModel): 
                
                 name: str 
                
                 description: str = None 
                
                 price: float 
                
                 tax: float = None 
                
                 @app.put("/items/{item_id}") 
                
                 async def update_item(*, item_id: int, item: Item = Body(..., embed=True)): 
                
                 results = {"item_id": item_id, "item": item} 
                
                 return results 
                
                 復制代碼

請求示例如：

如果另外再假設,客戶端提交一個更復雜的嵌套模型的話，怎么辦？麻蛋的肯定也是會有這樣的情況滴！嵌套里面有列表有實體。

如：

 
                 { 
                
                 "name": "Foo", 
                
                 "description": "The pretender", 
                
                 "price": 42.0, 
                
                 "tax": 3.2, 
                
                 "tags": ["rock", "metal", "bar"], 
                
                 "image": { 
                
                 "url": "http://example.com/baz.jpg", 
                
                 "name": "The Foo live" 
                
                 } 
                
                 } 
                
                 復制代碼

這時候，我們就需要所謂的子內嵌啦：

 
                 from typing import Set 
                
                 from fastapi import FastAPI 
                
                 from pydantic import BaseModel 
                
                 app = FastAPI() 
                
                 class Image(BaseModel): 
                
                 url: str 
                
                 name: str 
                
                 class Item(BaseModel): 
                
                 name: str 
                
                 description: str = None 
                
                 price: float 
                
                 tax: float = None 
                
                 tags: Set[str] = [] 
                
                 image: Image = None 
                
                 @app.put("/items/{item_id}") 
                
                 async def update_item(*, item_id: int, item: Item): 
                
                 results = {"item_id": item_id, "item": item} 
                
                 return results 
                
                 復制代碼

如上代碼，Item里面包含了Image，也包含了，tags類型的列表定義。

MMP更深層的嵌套也是可以定義的如：

{
    "name":"Foo",
    "description":"The pretender",
    "price":42,
    "items":[
        {
            "name":"Foo",
            "description":"The pretender",
            "price":42,
            "tax":3.2,
            "tags":[
                "rock",
                "metal",
                "bar"
            ],
            "image":{
                "url":"http://example.com/baz.jpg",
                "name":"The Foo live"
            }
        },
        {
            "name":"Foo2",
            "description":"The 2",
            "price":422,
            "tax":3.2,
            "tags":[
                "rock",
                "metal",
                "bar"
            ],
            "image":{
                "url":"http://example.com/baz.jpg",
                "name":"The Foo live"
            }
        }
    ]
}

對應的解析為：

from typing import Set
from typing import List, Set
 
class Image(BaseModel):
    url: str
    name: str
class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None
    tags: Set[str] = []
    # images: List[Image] = None
    image: Image = None
 
 
class Offer(BaseModel):
    name: str
    description: str = None
    price: float
    items: List[Item]
 
 
@app.post("/offers/")
async def create_offer(*, offer: Offer):
    return offer

3.10 Request Body的Field

Field字段的意思其實就是類似上面Query, Path，也同樣給Body內的字段的信息添加相關的校驗。

也就是說。通過Field來規范提交的Body參數信息。

如：

from fastapi import Body, FastAPI
from pydantic import BaseModel, Field
 
app = FastAPI()
 
 
class Item(BaseModel):
    name: str
    description: str = Field(None, title="標題啊",description="錯誤提示文字啊", max_length=300)
    price: float = Field(..., gt=0, description="錯誤提示文字啊")
    tax: float = None
 
 
@app.put("/items/{item_id}")
async def update_item(*, item_id: int, item: Item = Body(..., embed=True)):
    results = {"item_id": item_id, "item": item}
    return results

上面的意思就是和之前定義參數校驗其實一樣

正常情況：

異常情況：

3.11 其他數據類型的校驗

對於數據格式的校驗，通常，我們不止於

int
float
str
bool

但是提交參數不止於上述的幾種格式，有時候比如是對手機號碼的校驗，有些時候是時間類型的校驗等

其他類型：

其他數據類型¶ 以下是您可以使用的一些其他數據類型（來自官方文檔）：

UUID:

一個標准的“通用唯一標識符”，在許多數據庫和系統中常見於ID。
在請求和答復中，將表示為str.

datetime.datetime:

一只Pythondatetime.datetime.
在請求和答復中，將表示為str采用ISO 8601格式，如：2008-09-15T15:53:00+05:00.

datetime.date:

Pythondatetime.date.
在請求和答復中，將表示為str采用ISO 8601格式，如：2008-09-15.

datetime.time:

一只Pythondatetime.time.
在請求和答復中，將表示為str采用ISO 8601格式，如：14:23:55.003.

datetime.timedelta:

一只Pythondatetime.timedelta.
在請求和答復中，將表示為float總秒數。
Pydantic還允許將其表示為“ISO 8601時間差異編碼”，有關更多信息，請參閱文檔。.

frozenset:

在請求和答復中，將其視為set:
在請求中，將讀取列表，消除重復，並將其轉換為set.
在答復中，set將轉換為list.
生成的架構將指定set值是唯一的(使用JSONSchema的uniqueItems).

bytes:

標准Pythonbytes.
在請求和答復中將被視為str.
生成的架構將指定它是str帶着binary“格式”。

Decimal:

標准PythonDecimal.
在請求和響應中，處理方式與float.

所以我還可以使用其他類型來校驗：

from datetime import datetime, time, timedelta
from uuid import UUID
 
from fastapi import Body, FastAPI
 
app = FastAPI()
 
 
@app.put("/items/{item_id}")
async def read_items(
    item_id: UUID,
    start_datetime: datetime = Body(None),
    end_datetime: datetime = Body(None),
    repeat_at: time = Body(None),
    process_after: timedelta = Body(None),
):
    start_process = start_datetime + process_after
    duration = end_datetime - start_process
    return {
        "item_id": item_id,
        "start_datetime": start_datetime,
        "end_datetime": end_datetime,
        "repeat_at": repeat_at,
        "process_after": process_after,
        "start_process": start_process,
        "duration": duration,
    }

4：響應報文

4.1 使用response_model定義

請求一個接口返回來我們客戶端可見的東西都是所謂的響應報文，如響應頭，響應碼，響應內容等。

通常不會那么傻的用戶輸入什么就返回什么。以下的官網示例純粹的演示看：

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
 
app = FastAPI()
 
 
 
class UserIn(BaseModel):
    username: str
    password: str
    email: str
    full_name: str = None
 
 
class UserOut(BaseModel):
    username: str
    email: str
    full_name: str = None
 
 
@app.post("/user/", response_model=UserOut)
async def create_user(*, user: UserIn):
    return user

請求之后，獲取到是UserOut的內容信息：

通常再定義我們的API返回響應的時候，一般是返回固定JSON格式的，所以可以直接使用定義response_model為一個字典：

from typing import Dict
 
from fastapi import FastAPI
 
app = FastAPI()
 
 
@app.get("/keyword-weights/", response_model=Dict[str, float])
async def read_keyword_weights():
    return {"foo": 2.3, "bar": 3.4}

4.2 關於響應狀態碼status_code

通常的一個接口請求完成，如果沒有什么異常通常會返回200：如日志打印出來一樣：

 
                 INFO:     127.0.0.1:58141 - "POST /user/ HTTP/1.1" 400 
                
                 INFO:     127.0.0.1:58315 - "POST /user/ HTTP/1.1" 200 
                
                 復制代碼

FastAPI運行我們的指定返回的status_code

如下示例：

@app.post("/user/", response_model=UserOut,status_code=500)
async def create_user(*, user: UserIn):
    return user

導致請求的接口返回：

甚至還可以通過導入status來指定：

from fastapi import FastAPI, status
 
app = FastAPI()
 
 
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    return {"name": name}

5：錯誤處理

5.1 HTTPException異常拋出

再之前Bottle 中其實有一個就是HttpError異常類，在FastAPI也存在這么一個HTTPException。

如示例：

from fastapi import FastAPI, HTTPException
 
app = FastAPI()
 
items = {"foo": "The Foo Wrestlers"}
 
 
@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

在上面的代碼中，通過判斷item_id是不是存在於items來主動的拋出了一個404的錯誤

我們查看HTTPException和StarletteHTTPException的源碼發現他們也是繼承與Exception：

class HTTPException(StarletteHTTPException):
    def __init__(
        self, status_code: int, detail: Any = None, headers: dict = None
    ) -> None:
        super().__init__(status_code=status_code, detail=detail)
        self.headers = headers

所以我們對於異常通常可以直接的使用 raise來拋出異常。

5.2 HTTPException且返回新增自定義請求頭

from fastapi import FastAPI, HTTPException
 
app = FastAPI()
 
items = {"foo": "The Foo Wrestlers"}
 
 
@app.get("/items-header/{item_id}")
async def read_item_header(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]}

5.3 自定義返回HTTPException

類似之前Bottle我們通過添加一個自定義的全局的錯誤，來統一的處理返回。FastAPI其實也提供一個自定義錯誤的機制：

官方示例如下：

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
 
 
class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name
 
 
app = FastAPI()
 
 
@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )
 
 
@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

觀察請求結果：

當請求name == yolo的時候，我們主動拋出了UnicornException，而且我們，@app.exception_handler(UnicornException)也捕獲到相關的異常信息，且返回了相關的信息。

5.4 覆蓋FastAPI默認的異常處理

按官方文檔說明就是，當請求包含無效的數據的時候，或參數提交異常錯誤的時候，會拋出RequestValidationError，

那其實我也可以通過上面的自定義異常的方式來覆蓋重寫我們的RequestValidationError所返回信息：

如：默認代碼沒有添加覆蓋處理的話：發生異常的時候是提示是：

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
 
app = FastAPI()
 
 
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)
 
 
# @app.exception_handler(RequestValidationError)
# async def validation_exception_handler(request, exc):
#     return JSONResponse({'mes':'觸發了RequestValidationError錯誤，，錯誤信息:%s 你妹的錯了！'%(str(exc))})
 
 
 
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}
 
 
 
if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='main4:app', host="127.0.0.1", port=8000, reload=True, debug=True)

發生異常的請求下返回：

恢復覆蓋的時候：

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
 
app = FastAPI()
 
 
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)
 
 
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse({'mes':'觸發了RequestValidationError錯誤，，錯誤信息:%s 你妹的錯了！'%(str(exc))})
 
 
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}
 
 
 
if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='main4:app', host="127.0.0.1", port=8000, reload=True, debug=True)

請求結果：

上面的返回其實我們還可以修改一下返回如下，指定響應碼：

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
    )

說明：

# 注意fastapi包中的HTTPException才可以定義請求頭
from fastapi import Depends, status, HTTPException
# from starlette.exceptions import HTTPException

6：FastAPI 中間件

所謂的中間件，其實和我們bottle中的中間件作用是一致。有些方法或操作需要在所有路由之前執行，比如要加一個http訪問的攔截器，可以對部分接口API需要授權才能訪問的接口進行驗證之類的。

FastAPI提供了一個@app.middleware("http")可以做到類似上面的攔截功能。其實和bottle或flask 鈎子函數很相似

示例如下：

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
 
import time
from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
 
app = FastAPI()
 
 
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)
 
 
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse({'mes':'觸發了RequestValidationError錯誤，，錯誤信息:%s 你妹的錯了！'%(str(exc))})
 
 
@app.get("/items/{item_id}")
async def read_item(item_id: int):
 
    return {"item_id": item_id}
 
 
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response
 
 
if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='main4:app', host="127.0.0.1", port=8000, reload=True, debug=True)

然后我們請求完成后發現，我們的響應頭里多了一個新增的請求頭：

7：FastAPI 跨域處理

為啥需要跨域處理，通常我們的API一般是給到前端去調用，但是前端可能使用域名和沒提供的API域名是不一樣，這就引發了瀏覽器同源策略問題，所以我們需要做跨域請求支持。

FastAPI支持跨域的話，可以通過添加中間的形式，和bottle也有相似之處。不僅如此他還支持僅限於支持哪些域名進行跨域請求：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
 
app = FastAPI()
 
origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]
 
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
 
 
@app.get("/")
async def main():
    return {"message": "Hello World"}

懶得起一個js訪問了，所以這個暫時不測試了，后期有機會再測試驗證一下，感覺應該就是這樣的。

8：FastAPI 依賴注入之Depends（補充）

看官網的描述Depends的使用，似乎比較懵懵懂懂的，於是乎還是需要花些時間再次學習一下關於依賴注入。

首先依賴注入它可以是函數也可以是類，如下面的函數形式的依賴注入：

8.1 簡單的依賴說明

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
 
app = FastAPI()
 
origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]
 
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
 
 
@app.get("/")
async def main():
    return {"message": "Hello World"}

梳理一下接口請求的流程：

1：上面的commons: dict = Depends(common_parameters)它聲明了一個依賴關系： Depends(common_parameters)：這對接口的依賴進行了一個聲明，表示的是接口參數請求依賴於common_parameters的函數。

當接口被調用的時候，回調用common_parameters函數進行請求處理。
2： common_parameters函數主要是負責接收函數，處理后返回一個字典，
3：然后把Depends(common_parameters)返回的結果傳遞給commons: dict，這個就是一個依賴注入的過程。

所以在上面的示例中common_parameters是我們被依賴對象

這個被依賴的對象，對接口請求的要求就是:

可選查詢參數q那是一個str.
可選查詢參數skip那是int，默認情況下是0.
可選查詢參數limit那是int，默認情況下是100.
返回一個字典

請求示例：

這依賴注入的方式其實也挺方便，類似於接口裝飾器的方式，比如common_parameters中我們可以先對相關的參數進行校驗攔截，還可以再傳遞。場景可以和我們之前的bottle的裝飾器差不多類似：

相同的邏輯判斷處理
用戶身份鑒權

8.2 把類當作被依賴對象

上面我們的被依賴的對象是以函數的形式出現，那FastAPI它也支持以類的形式來表達。按官網說法被依賴對象它應該是必須一個可以調用的對象比如：類，函數之類的···

這里看一下以類的形式：

from fastapi import Depends, FastAPI
 
app = FastAPI()
 
 
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
 
 
class CommonQueryParams:
    def __init__(self, q: str = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit
 
 
@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    response = {}
    # 如果q存在
    if commons.q:
        # 我們就把q加到一個新字典
        response.update({"q": commons.q})
        response.update({"小鍾": '同學'})
    #然后在我們的fake_items_db進行截取
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

上面我們的CommonQueryParams是一個類，它和我的函數其實差不多，當我們的接口被調用的時候，類對象就回被初始化，按官網的說法： commons: CommonQueryParams = Depends(CommonQueryParams) 和 commons = Depends(CommonQueryParams) 是等價的。還有一種是： commons: CommonQueryParams = Depends()

示例運行演示

有Q參數：

沒有Q參數：

8.3 多層嵌套依賴

多層嵌套的意思就是可以類可以類的意思。函數可以依賴函數。其實和我們的之前的參數校驗一樣。

比如下面的代碼：

from fastapi import Cookie, Depends, FastAPI
 
app = FastAPI()
 
 
def query_extractor(q: str = None):
   return q
 
 
def query_or_cookie_extractor(
   q: str = Depends(query_extractor), last_query: str = Cookie(None)
):
   if not q:
       return last_query
   return q
 
 
@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
   return {"q_or_cookie": query_or_default}

其實意思就是query_or_cookie_extractor 依賴於query_extractor，然后query_or_cookie_extractor被注入到接口上也被依賴的對象。

官網的截圖上面接口執行流程如下：

對於同一個依賴，如果處理的結果是一樣的，就是返回值是一樣的話，我們可以進行多次調用依賴，這時候可以對被依賴的對象設置是否使用緩存機制：

async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):
  return {"fresh_value": fresh_value}

8.4 list列表依賴

我們先看官方提供的示例代碼：

from fastapi import Depends, FastAPI, Header, HTTPException
 
app = FastAPI()
 
 
async def verify_token(x_token: str = Header(...)):
 if x_token != "fake-super-secret-token":
     raise HTTPException(status_code=400, detail="X-Token header invalid")
 
 
async def verify_key(x_key: str = Header(...)):
 if x_key != "fake-super-secret-key":
     raise HTTPException(status_code=400, detail="X-Key header invalid")
 return x_key
 
 
@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
 return [{"item": "Foo"}, {"item": "Bar"}]

上述的代碼意思是對我們的請求頭部信息Header進行驗證，因為示例是...三個點，說明是必選的字段：

分析上述的代碼之后，運行一下試一試看看結果：

1：什么頭部參數都不傳遞的情況提示，我們的頭部參數異常

2：頭部參數填寫的情況：

注意點：參數提交的格式，因為是頭部的參數，所以我們的代碼上的x_token 會應該要寫：x-token才對

錯誤的示例：

所以上面列表的依賴的意思就是必須兩天條件都成立才通過。這個感覺后期還是用到的比較多的喲！

8.5 多依賴對象注入和列表其實是一樣的：

from fastapi import Depends, FastAPI
 
 
from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi import Depends, FastAPI
 
app = FastAPI()
 
 
 
 
async def verify_token(x_token: str = Header(...)):
  if x_token != "fake-super-secret-token":
      raise HTTPException(status_code=400, detail="X-Token header invalid")
  return x_token
 
async def verify_key(x_key: str = Header(...)):
  if x_key != "fake-super-secret-key":
      raise HTTPException(status_code=400, detail="X-Key header invalid")
  return x_key
 
 
@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
  return [{"item": "Foo"}, {"item": "Bar"}]
 
 
@app.get("/items2/")
async def items2(xt: str = Depends(verify_token),xk: str = Depends(verify_key)):
 return {"xt": xt,'xk':xk}
 
 
 
if __name__ == '__main__':
  import uvicorn
  uvicorn.run(app='main:app', host="127.0.0.1", port=8100, reload=True, debug=True)

如上面的xt: str = Depends(verify_token),xk: str = Depends(verify_key)，也是需要必須兩個條件成立才可以。

正常情況：

非正常情況：

總結

文章總體是跟着官方文檔的思路走，簡單梳理了一下，后續我們的API用到一些知識點。下一步嘗試基於上述知識點弄一個簡單的腳手架看看。

原文鏈接http://www.zyiz.net/tech/detail-119883.html

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 從零開始學AB測試：躲坑篇從零開始學Sketch——進階篇 [轉]postman 官方文檔解說 Hive Beeline 官方文檔學習 Spring JMS 官方文檔學習從零開始一起學習SLAM | 為什么要學SLAM？ Android File類根據官方文檔理解(轉) 跟着官方文檔學Maven構建生命周期 Vue2.0 官方文檔學習筆記 sqlserver 微軟官方學習文檔