1. <dd id="erndk"></dd>
                1. 從零開始學FastAPI(1)-簡明的部分官方文檔抽取學習篇

                  小鐘同學 2020/3/19 17:01:54

                  前言之前線上業務一直是使用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(...)進行處理和提?。?/p>



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

                  {
                      "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 Depends, FastAPI
                  
                  app = FastAPI()
                  
                  
                  async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
                      return {"q": q, "skip": skip, "limit": limit}
                  
                  
                  @app.get("/items/")
                  async def read_items(commons: dict = Depends(common_parameters)):
                      commons.update({'小鐘':'同學'})
                      return commons
                  
                  
                  @app.get("/users/")
                  async def read_users(commons: dict = Depends(common_parameters)):
                      return commons
                  
                  if __name__ == '__main__':
                      import uvicorn
                      uvicorn.run(app='main:app', host="127.0.0.1", port=8100, reload=True, debug=True)
                  復制代碼

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

                  • 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用到一些知識點。下一步嘗試基于上述知識點弄一個簡單的腳手架看看。

                  END

                  小鐘同學 | 文  【原創】【轉載請聯系本人】| QQ:308711822

                  隨時隨地學軟件編程-關注百度小程序和微信小程序
                  關于找一找教程網

                  本站文章僅代表作者觀點,不代表本站立場,所有文章非營利性免費分享。
                  本站提供了軟件編程、網站開發技術、服務器運維、人工智能等等IT技術文章,希望廣大程序員努力學習,讓我們用科技改變世界。
                  [從零開始學FastAPI(1)-簡明的部分官方文檔抽取學習篇]http://www.yachtsalesaustralia.com/tech/detail-119883.html

                  贊(0)
                  關注微信小程序
                  程序員編程王-隨時隨地學編程

                  掃描二維碼或查找【程序員編程王】

                  可以隨時隨地學編程啦!

                  技術文章導航 更多>
                  掃一掃關注最新編程教程
                  国产在线拍揄自揄视频菠萝

                        1. <dd id="erndk"></dd>