FastAPI快速查閱

• 2022 年 1 月 6 日
• 筆記
• FastAPI, Python

官方文檔主要側重點是循序漸進地學習FastAPI, 不利於有其他框架使用經驗的人快速查閱
故本文與官方文檔不一樣, 並補充了一些官方文檔沒有的內容

安裝

包括安裝uvicorn

$pip install fastapi[all]

分開安裝

$pip install fastapi
$pip install uvicorn[standard]

uvicorn使用

uvicorn是一個非常快速的 ASGI 伺服器。
官方文檔在這裡: uvicorn

命令行啟動

# mian.py
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def index():
	return {"index": "root"}

$uvicorn --reload main:app

程式碼中啟動

# main.py
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def index():
	return {"index": "root1"}


if __name__ == '__main__':
	import uvicorn

	uvicorn.run("main:app", host="127.0.0.1", port=8888, reload=True)

配置

注: 使用uvicorn --help可以查看完整配置

$uvicorn --help
Usage: uvicorn [OPTIONS] APP
...

路由

單個文件

和其他輕型web框架一樣: 使用@xx.請求方式, 指定路徑

一般的使用: app = FastAPI()

• @app.get()
• @app.post()
• @app.put()
• @app.delete()
• @app.options()
• @app.head()
• @app.patch()
• @app.trace()

# 1 導入fast api
from fastapi import FastAPI

# 2 創建實例
app = FastAPI()

# 3 綁定路由
"""

常見的REST url通常:
POST：創建數據。
GET：讀取數據。
PUT：更新數據。
DELETE：刪除數據。
"""


@app.get("/")
async def root():
	return {"message": "hello world"}

參數見下文的app.get等的參數

多個文件

假如, 文件結構這樣:

+--- app
|   +--- main.py
|   +--- routers
|   |   +--- movie.py
|   |   +--- music.py
|   |   +--- __init__.py

• main.py: 網站主頁, 負責啟動fast
• movie.py: 處理/movie/xxx的URL
• music.py: 處理/music/xxx的URL

具體程式碼

使用兩種方式定義

# movie.py

from fastapi import APIRouter

router = APIRouter()


@router.get("/")
async def movie():
	return {"message": "movie"}

# music.py

from fastapi import APIRouter

# 前綴不能以 / 作為結尾
router = APIRouter(prefix="/music")


@router.get("/")
async def music():
	return {"message": "music"}

# main.py

from fastapi import FastAPI
from routers import music, movie

app = FastAPI()
# 方式一,直接導入
app.include_router(music.router)
# 方式二, 添加額外參數, 為已存在router修飾
app.include_router(prefix="/movie", router=movie.router)


@app.get("/")
async def root():
	return {"message": "hello world"}


if __name__ == "__main__":
	import uvicorn

	config = {
		"app": "main:app",
		"host": "127.0.0.1",
		"port": 8000,
		"reload": True

	}
	uvicorn.run(**config)

訪問//127.0.0.1:8000/music/和//127.0.0.1:8000/movie/可以找到對應的頁面

include_router的參數見下文的app.include_router的參數
APIRouter的參數見: APIRouter的參數

設置子應用

將一個app掛載到另一個app上

from fastapi import Depends, FastAPI

app = FastAPI()
sub_app = FastAPI()


# /home/
@app.get("/home/")
async def home():
    return {"index": "home"}


#  /api/users/
@sub_app.get("/users/")
async def users():
    return {"index": "users"}


# 將 /api 掛在到 / 
app.mount("/api", sub_app)

if __name__ == '__main__':
    import uvicorn

    uvicorn.run("fast_test:app", host="127.0.0.1", port=8888, reload=True)

見: sub-applications

一些參數

這部分內容包括FastAPI APIRouter app app.include_router的參數

FastAPI的參數

FastAPI繼承Starlette, 一些參數與Starlette的參數相同

APIRouter的參數

app.get等的參數

說實話app.get等的參數著實有點多, 而且很多都有生產doc有關, 具體如何使用可以點擊表格中的鏈接.

app.include_router的參數

Reqeust

解析請求參數的順序: 路徑參數 > 查詢參數 > 請求體參數

路徑參數

即, 一般的路由
不會把參數轉換為對應的數據類型

from fastapi import FastAPI

app = FastAPI()


# 路徑參數
@app.get("/test/{item_id}")
async def retrieve(item_id):
    return {"item_id": item_id}

有類型的路徑參數

為參數指定參數類型即可
一些常用的類型見: typing

@app.get("/test/{item_id}")
async def retrieve(item_id: int):
    # item_id 會自動轉換為int
    return {"item_id": item_id}

參數對應的類型不對應的話, 報錯

給路徑參數設置預設值

使用枚舉類型, 定義預設值

from fastapi import FastAPI
from typing import Optional
from enum import Enum

# ...

class ItemId(str, Enum):
    a = "aa"
    b = "bb"
    c = "cc"


@app.get("/test2/{item_id}")
async def test2(item_id: ItemId):
    # item_id只能是aa/bb/cc
    
    # 裡面可以if判斷,處理不同的邏輯
    return {"item_id": item_id}

參數對應的值, 不為預設值的話, 報錯

為路徑參數作描述或限制

使用fastapi.Path接收, 可以為路徑參數聲明相同類型的校驗和元數據

from typing import Optional

from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
    item_id: int = Path(..., title="The ID of the item to get"),
    q: Optional[str] = Query(None, alias="item-query"),
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

注: Path是Param的子類, 具有通用的方法, 具體參數見: Param

路徑轉換器

# 以下為路徑轉換器
@app.get("/test3/{file_path:path}")
async def file_retrieve(file_path):
    return {"file_path": file_path}

這個例子, 會將形如: /test3//root/, 那麼, file_path:path為 /root/, 注意是兩個//.

查詢參數

聲明不屬於路徑參數的其他函數參數時，它們將被自動解釋為”查詢字元串”參數

默認參數

和路徑參數, 不一樣
查詢參數是可以有默認值的

# 沒有默認值:必選參數
# 有默認值: Optional, 非必選參數
# 可以是布爾類型, 可將1/True/true/on/yes轉換為python的bool值

@app.get("/test")
async def test_list(page: int, limit: Optional[int] = None):
    return {"page": page, "limit": limit}

設置參數預設值

from fastapi import FastAPI
from typing import Optional
from enum import Enum


# ...

# 參數預設值
class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"


@app.get("/models")
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"}

為查詢參數作描述或限制

fastapi.Query可以為查詢參數進行校驗

@app.get("/items")
async def test3(item_id: List[int] = Query(..., title="id錯誤", description="id 必須大於10", alias="item-id", ge=10)):
    # 路徑形如: //127.0.0.1:8000/items?item-id=11&item-id=12
    return {"item_id": item_id}

注: Query是Param的子類, 具有通用的方法, 更多參數見: Param

請求體參數

請求體是客戶端發送給 API 的數據

pydantic庫是python中用於數據介面定義檢查與設置管理的庫。
FastAPI會將pydantic的類型在請求體中匹配

關於Pydantic的詳細操作, 見: Pydantic使用

BaseModel 一般使用

定義pydantic.BaseModel的子類, 作為接收請求體的類型

和typing使用一樣, 使用=指定默認值, 為可選參數, 不知道默認值則為必須參數

from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


# 1. 定義pydantic.BaseModel 子類
class Item(BaseModel):
    # 2. 定義數據類型
    name: str
    age: int
    description: Optional[str] = None


# 3. 混合使用
# ** 請使用 postman等工具調試
# ** Item

@app.post("/test/item/{item_id}")
async def item_retrieve(item_id, item: Item, page: int = 1, limit: Optional[int] = None):
    print(item_id)
    print(page)
    print(limit)
    return item.dict()

使用:

curl -X 'POST' \
  '//127.0.0.1:8000/test/item/1?page=1' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "name": "string",
  "age": 10,
  "description": "string"
}'

fastAPI會將請求體中的數據賦值給Item (我們定義的baseModel子類)
關於BaseModel的方法, 可以看這裡Model屬性
一般的使用方法有item.name或item.dict()

Field 額外約束

即pydantic.BaseModel與pydantic.Field相結合
pydantic.Field可以為BaseModel的欄位添加額外的約束條件

Field參數:

1. default 默認值, 注意: ...為必須值
2. alias 別名, 即請求體的key
3. const 是否只能是默認值
4. title 標題名稱, 默認為欄位名稱的title()方法
5. description 詳細, 用於文檔使用
6. gt/ge/lt/le/regex 大於/大於等於/小於/小於等於/正則表達式驗證

class Item(BaseModel):
    # 2. 定義數據類型
    name: str
    age: int = Field(..., ge=10, description="age must ge 10", title="age title")    # !!! 使用Field
    description: Optional[str] = None

單個請求體參數

pydantic.BaseModel可以匹配多條數據, 而fastapi.Body只能匹配一條數據
當pydantic.BaseModel與fastapi.Body結合時, 傳入的數據需要裹上一個{}

@app.post("/test2/{item_id}")
async def test2_retrieve(item_id, item: Item, username: str = Body(..., regex=r"^lcz"), page: int = 1):
    return {"username": username}

"""
發送: //127.0.0.1:8000/test2/1
{
  "item": {
    "name": "string",
    "age":11,
    "description": "string"
  },
  "username": "lczmx"
}
"""

注: Body是FieldInfo的子類, 具有通用的方法, 更多參數見: Body

多個請求體模型-並列

多個pydantic.BaseModel參數, 請求體數據同樣在外面裹上一個{}

from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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


class User(BaseModel):
    username: str
    full_name: Optional[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

數據:

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

多個請求體模型-嵌套

一個BaseModel的欄位為另一個BaseModel時, 傳入的數據同樣是嵌套的.

from typing import Optional, Set

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Image(BaseModel):
    url: str
    name: str


class Item(BaseModel):
    name: str
	# 嵌套另一個模型
    image: Optional[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

數據:

{
    "name": "Foo",
    "image": {
        "url": "//example.com/baz.jpg",
        "name": "The Foo live"
    }
}

列表請求體數據

只需要將參數指定為List[BaseModel]即可:

from typing import List

from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Image(BaseModel):
    url: HttpUrl
    name: str


@app.post("/images/multiple/")
async def create_multiple_images(images: List[Image]):
    return images

數據:

[
    {
        "url": "//xxx.com/1.jpg",
        "name": "1.jpg"
    }
]

更多內置欄位類型

所有的欄位類型見官方文檔: Field Types
上面欄位主要是這幾個:

1. 標準的: Standard Library Types
2. pydantic定義的: Pydantic Types
3. 等…

例子:

from datetime import datetime, time, timedelta

from typing import Optional

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: Optional[datetime] = Body(None),
   end_datetime: Optional[datetime] = Body(None),
   repeat_at: Optional[time] = Body(None),
   process_after: Optional[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,
   }

也可以在BaseModel子類中定義

更多驗證方式

pydantic擁有更加細的自定義驗證器定義方法, 詳情點擊這裡

Form表單

需要安裝python-multipart:

$pip install python-multipart

讀取application/x-www-form-urlencoded

application/x-www-form-urlencoded的數據形如: say=Hi&to=Mom
即, 我們一般的input表單數據

from fastapi import FastAPI, Form

app = FastAPI()


@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
    return {"username": username}

發送數據:

POST //localhost:8000/login/
Content-Type: application/x-www-form-urlencoded

username=lczmx&password=123456

返回數據:

{
  "username": "lczmx"
}

注: Form是Body的子類, 具有通用的方法, 更多參數見: Body

讀取multipart/form-data

即上傳文件

使用pycharm HTTP Client發送數據:

POST /test.html HTTP/1.1
Host: example.org
Content-Type: multipart/form-data;boundary="boundary"

--boundary
Content-Disposition: form-data; name="field1"

value1
--boundary
Content-Disposition: form-data; name="field2"; filename="example.txt"

value2

有以下兩種接收方式:

使用bytes接收

在接收文件時, 必須使用fastapi.File, 否則, FastAPI 會把該參數當作查詢參數或請求體（JSON）參數。

注意: 文件是二進位數據, 故使用bytes類型. input標籤的name屬性作為變數名
例子:

from typing import List

from fastapi import FastAPI, File

app = FastAPI()


# 接收單個文件直接用bytes, 多個文件使用List
@app.post("/files/")
async def create_file(first: bytes = File(...), second: List[bytes] = File(...)):
    return {
        "firstFileSize": len(first),
        "secondFilesContent": [f.decode("utf-8") for f in second]
    }


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

使用pycharm HTTP Client發送數據:

POST //localhost:8000/files/
Content-Type: multipart/form-data; boundary=boundary

--boundary
Content-Disposition: form-data; name="first"; filename="r.txt"

// 上傳r.txt, 需要本地有r.txt
< ./r.txt

--boundary
Content-Disposition: form-data; name="second"; filename="input-second.txt"

// 內容直接為Text Content1
Text Content1

--boundary
Content-Disposition: form-data; name="second"; filename="input-second.txt"

// 內容直接為Text Content2
Text Content2

響應數據:

{
  "firstFileSize": 30,
  "secondFilesContent": [
    "Text Content1",
    "Text Content2"
  ]
}

注: File是Form的子類, 具有通用的方法, 更多參數見: Body

使用UploadFile接收

由於使用bytes不能處理文件的資訊, 為此在某些情況下使用UploadFile更加方便

from typing import List

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


# 接收單個文件直接用bytes, 多個文件使用List
@app.post("/files/")
async def create_file(first: UploadFile = File(...), second: List[UploadFile] = File(...)):
    return {
        "firstFileName": first.filename,
        "secondFilesContent": [f.file.read() for f in second]
    }


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

使用上面的請求數據, 響應數據為:

{
  "firstFileName": "r.txt",
  "secondFilesContent": [
    "Text Content1",
    "Text Content2"
  ]
}

UploadFile 與 bytes 相比有更多優勢：

1. 使用UploadFile類進行文件上傳時，
   會使用到一種特殊機制「離線文件」（Spooled File）：即是當文件在記憶體讀取超過一定限制後，多出來的部分會寫入磁碟。
2. UploadFile適合用於大文件傳輸, 如: 影像、影片、二進位文件等大型文件，好處是不會佔用所有記憶體；
3. 自帶 file-like async 介面
4. 暴露的Python SpooledTemporaryFile對象, 可直接傳遞給其他預期「file-like」對象的庫。

UploadFile的屬性

UploadFile的方法

使用UploadFile讀取文件數據:

# 1. async方法
contents = await myfile.read()

# 2. 普通方法
contents = myfile.file.read()

Response

response_class參數可以指定響應類, 直接return數據即可, 如 HTML

一般的response

from fastapi import FastAPI, Response

app = FastAPI()


@app.get("/index")
async def index():
    """
	響應的參數
    content 響應體內容
    status_code 狀態碼, 默認200
    headers 響應頭
    media_type 響應類型
    background 後台任務
    """
    f = open("statics/index.html", encoding="utf8")
    response = Response(content=f.read(), media_type="text/html", status_code=200, headers={"x-server": "Test Server"})
    f.close()
    return response


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app="test:app", host="127.0.0.1", port=8000, reload=True)

響應模型

FastAPI可以根據根據請求數據快速返回對應的數據

如:

// Request:
// POST /book
{
	"name": "book1",
	"price": 99
}

// Response:
{
	"name": "book1",
	"price": 99
}

一般使用 輸入同輸出

通過response_model參數指定
但是, 不通過response_model參數直接返回亦可以, 但不能自動生成返回值的doc

程式碼:

from typing import List, Optional

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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


# 這種情況可以省略response_model
# 但是, 省略的話, 不能再doc中顯示
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    return item


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

請求:
使用pycharm HTTP Client發送數據:

POST //localhost:8000/items
Content-Type: application/json

{
  "name": "name1",
  "price": 1000,
  "description": "this is description"
}

響應:

{
  "name": "name1",
  "description": "this is description",
  "price": 1000.0,
  "tax": null,
  "tags": []
}

FastAPI會將resturn的數據自動轉換為Item中的數據
所以需要名稱對應, 缺失欄位的話會報錯!!

注意: 這種使用方法會將全部請求數據作為返回數據, 在某些場合併不適合!

輸入模型與輸出模型分開

from typing import Optional

from pydantic import BaseModel
from fastapi import FastAPI


class UserIn(BaseModel):
    """
    用戶輸入數據
    """
    username: str
    password: str
    age: int
    description: Optional[str] = None


class UserOut(BaseModel):
    """
    用戶輸出數據
    """
    # 剔除password
    username: str
    age: int
    description: Optional[str] = None


app = FastAPI()


@app.post("/user", response_model=UserOut)
def register(data: UserIn):
    return data


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

可以看到: 接收數據模型為UserIn, return data使用輸出數據模型 (UserOut) 接收

為輸出模型作限定

我們可以通過指定參數, 為輸出模型的欄位作修改
也就是說, 我們在某些場合下可以 在只使用一個模型的情況下 過濾敏感數據

1. 不返回默認值 response_model_exclude_unset

   FastAPI默認會將默認值返回

   from typing import Optional
   
   from pydantic import BaseModel
   from fastapi import FastAPI
   
   
   class UserIn(BaseModel):
   	"""
   	用戶輸入數據
   	"""
   	username: str
   	password: str
   	age: int
   	description: Optional[str] = None
   
   
   class UserOut(BaseModel):
   	"""
   	用戶輸出數據
   	"""
   	# 剔除password
   	username: str
   	age: int
   	description: Optional[str] = None
   
   
   app = FastAPI()
   
   
   @app.post("/user", response_model=UserOut, response_model_exclude_unset=True)
   def register(data: UserIn):
   	return data
   
   
   if __name__ == '__main__':
   	import uvicorn
   
   	uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)
   
   

   如發送數據為:

   {
     "username": "lczmx",
     "password": "123456",
     "age": 18
   }
   

   返回數據為:

   {
     "username": "lczmx",
     "age": 18
   }
   

   原理: FastAPI會將輸出模型的.dict()方法的exclude_unset參數指定, 見: pydanticExporting models

2. 不返回與默認值相同的值 response_model_exclude_defaults

   from typing import Optional
   
   from pydantic import BaseModel
   from fastapi import FastAPI
   
   
   class UserIn(BaseModel):
   	"""
   	用戶輸入數據
   	"""
   	username: str
   	password: str
   	age: int
   	description: Optional[str] = None
   
   
   class UserOut(BaseModel):
   	"""
   	用戶輸出數據
   	"""
   	# 剔除password
   	username: str
   	age: int
   	description: Optional[str] = "abc"
   
   
   app = FastAPI()
   
   
   @app.post("/user", response_model=UserOut, response_model_exclude_defaults=True)
   def register(data: UserIn):
   	return data
   
   
   if __name__ == '__main__':
   	import uvicorn
   
   	uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)
   
   

   如發送數據為:

   {
     "username": "lczmx",
     "password": "123456",
     "age": 18,
     "description": "abc"
   }
   

   返回數據為:

   {
     "username": "lczmx",
     "age": 18
   }
   

   原理: FastAPI會將輸出模型的.dict()方法的exclude_defaults參數指定, 見: pydanticExporting models

3. 不返回為None的值 response_model_exclude_none

   from typing import Optional
   
   from pydantic import BaseModel
   from fastapi import FastAPI
   
   
   class UserIn(BaseModel):
   	"""
   	用戶輸入數據
   	"""
   	username: str
   	password: str
   	age: int
   	description: Optional[str] = None
   
   
   class UserOut(BaseModel):
   	"""
   	用戶輸出數據
   	"""
   	# 剔除password
   	username: str
   	age: int
   	description: Optional[str] = "abc"
   
   
   app = FastAPI()
   
   
   @app.post("/user", response_model=UserOut, response_model_exclude_none=True)
   def register(data: UserIn):
   	return data
   
   
   if __name__ == '__main__':
   	import uvicorn
   
   	uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)
   
   

   如發送數據為:

   {
     "username": "lczmx",
     "password": "123456",
     "age": 18,
     "description": null
   }
   

   返回數據為:

   {
     "username": "lczmx",
     "age": 18
   }
   

   原理: FastAPI會將輸出模型的.dict()方法的exclude_none參數指定, 見: pydanticExporting models

4. 只返回某些欄位 response_model_include

   from typing import Optional
   
   from pydantic import BaseModel
   from fastapi import FastAPI
   
   
   class UserIn(BaseModel):
   	"""
   	用戶輸入數據
   	"""
   	username: str
   	password: str
   	age: int
   	description: Optional[str] = None
   
   
   app = FastAPI()
   
   
   @app.post("/user", response_model=UserIn, response_model_include={"password"})
   def register(data: UserIn):
   	return data
   
   
   if __name__ == '__main__':
   	import uvicorn
   
   	uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)
   
   

   例子中: 只返回password欄位
   原理: FastAPI會將輸出模型的.dict()方法的include參數指定, 見: pydanticExporting models

5. 不返回某些欄位 response_model_exclude

   from typing import Optional
   
   from pydantic import BaseModel
   from fastapi import FastAPI
   
   
   class UserIn(BaseModel):
   	"""
   	用戶輸入數據
   	"""
   	username: str
   	password: str
   	age: int
   	description: Optional[str] = None
   
   
   app = FastAPI()
   
   
   @app.post("/user", response_model=UserIn, response_model_exclude={"password"})
   def register(data: UserIn):
   	return data
   
   
   if __name__ == '__main__':
   	import uvicorn
   
   	uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)
   
   

   例子中: 不返回password欄位
   原理: FastAPI會將輸出模型的.dict()方法的exclude參數指定, 見: pydanticExporting models

通過繼承減少程式碼

以註冊為例子

from typing import Optional
from hashlib import md5
import logging
from logging import config

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()
# 秘鑰
SECRET = r"""=+Au+Z]Ho%W@fG6j7gb\`_@=tUG`|6*!yze:=fi(v&125hirNc$('=AH3FC"wj)E"""
# logging配置
config.dictConfig({
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "running": {
            "()": "uvicorn.logging.DefaultFormatter",
            "fmt": "%(levelprefix)s %(message)s",
            "use_colors": None,
        },
    },
    "handlers": {
        "running": {
            "formatter": "running",
            "class": "logging.StreamHandler",
            "stream": "ext://sys.stdout",
        },
    },
    "loggers": {
        "running": {"handlers": ["running"], "level": "INFO"},
    },
})
logger = logging.getLogger("running")
log_level = logging.INFO  # 默認logging級別


class UserBase(BaseModel):
    """
    用做數據模板
    """
    username: str
    email: EmailStr
    full_name: Optional[str] = None


class UserIn(UserBase):
    """
    輸入模型
    """
    password: str


class UserOut(UserBase):
    """
    輸出模型
	同 UserBase
    """
    pass


class UserInDB(UserBase):
    """
    寫入資料庫的模型
    """
    hashed_password: str


def fake_password_hasher(raw_password: str) -> str:
    """
    為明文密碼作hash
    :param raw_password: 明文密碼
    :return: 加密密文
    """
    m = md5()
    m.update(SECRET.encode())
    m.update(raw_password.encode())
    return m.hexdigest()


def create_user(user_in: UserIn):
    """
    創建用戶並保存到資料庫[假裝]
    """
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)
    logger.info("save to db")
    if log_level <= logging.DEBUG:
        logger.setLevel(logging.DEBUG)
    logger.debug(f"hashed password is {hashed_password}")
    logger.setLevel(logging.INFO)

    return user_in_db


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


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True, log_level=log_level)

請求數據:

POST //localhost:8000/user
Content-Type: application/json

{
  "username": "lczmx",
  "email": "[email protected]",
  "full_name": "xxx",
  "password": "123456"
}

響應數據:

{
  "username": "lczmx",
  "email": "[email protected]",
  "full_name": "xxx"
}

使用Union List Dict與模型結合

1. Union
   你可以將一個響應聲明為兩種類型的 Union，這意味著該響應將是兩種類型中的任何一種。

   from typing import Union
   
   from fastapi import FastAPI
   from pydantic import BaseModel
   
   app = FastAPI()
   
   
   class BaseItem(BaseModel):
   	description: str
   	type: str
   
   
   class CarItem(BaseItem):
   	type = "car"
   
   
   class PlaneItem(BaseItem):
   	type = "plane"
   	size: int
   
   
   items = {
   	"item1": {"description": "All my friends drive a low rider", "type": "car"},
   	"item2": {
   		"description": "Music is my aeroplane, it's my aeroplane",
   		"type": "plane",
   		"size": 5,
   	},
   }
   
   
   @app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem])
   async def read_item(item_id: str):
   	return items[item_id]
   
   

2. List
   聲明由對象列表構成的響應

   from typing import List
   
   from fastapi import FastAPI
   from pydantic import BaseModel
   
   app = FastAPI()
   
   
   class Item(BaseModel):
   	name: str
   	description: str
   
   
   items = [
   	{"name": "Foo", "description": "There comes my hero"},
   	{"name": "Red", "description": "It's my aeroplane"},
   ]
   
   
   @app.get("/items/", response_model=List[Item])
   async def read_items():
   	return items
   

3. Dict
   你還可以使用一個任意的普通 dict 聲明響應，僅聲明鍵和值的類型，而不使用 Pydantic 模型。

   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}
   
   

status_code

FastAPI支援修改status code
status_code可以直接用數字表示, 但FastAPI提供了一些內置狀態碼變數:
位於fastpi.status, 需要根據需求確定具體要用哪個狀態碼
HTTP狀態碼可以點擊這裡查看, WebSocket狀態碼可以點擊這裡查看

修改成功響應的狀態碼

from typing import Optional

from fastapi import FastAPI, status
from pydantic import BaseModel

app = FastAPI()


class BookModel(BaseModel):
    name: str
    price: int
    info: Optional[str] = None


@app.post("/books", status_code=status.HTTP_201_CREATED, response_model=BookModel)
def create_book(data: BookModel):
    return data


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

使用pycharm HTTP Client發送數據:

POST //localhost:8000/books/
Content-Type: application/json

{
  "name": "b1",
  "price": 100,
  "info": "book b1 information"
}

響應的數據:

POST //localhost:8000/books/

HTTP/1.1 201 Created
date: Sat, 06 Nov 2021 13:28:06 GMT
server: uvicorn
content-length: 54
content-type: application/json

{
  "name": "b1",
  "price": 100,
  "info": "book b1 information"
}

在執行過程中修改狀態碼

比如: 使用PUT請求, 若數據已經存在, 返回已經存在數據 狀態碼為200, 否則創建, 返回數據 狀態碼為201

from fastapi import FastAPI, Response, status

app = FastAPI()

tasks = {"foo": "Listen to the Bar Fighters"}


@app.put("/get-or-create-task/{task_id}", status_code=200)
def get_or_create_task(task_id: str, response: Response):
    if task_id not in tasks:
        tasks[task_id] = "This didn't exist before"

        response.status_code = status.HTTP_201_CREATED

    return tasks[task_id]


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

即, 通過response.status_code指定

JSON

FastAPI默認返回json格式的數據, 即response_class的默認值為: JSONResponse

將其他數據結構轉化為json, 見這裡: 數據轉換

HTML

通過response_class參數處理響應的類, HTMLResponse即返回html的類

from fastapi import FastAPI
from fastapi.responses import HTMLResponse

app = FastAPI()


@app.get("/", response_class=HTMLResponse)
async def home():
    return """<html>
            <head>
                <title>title</title>
            </head>
            <body>
                <h1>測試HTML</h1>
            </body>
        </html>
        """


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("fast_test:app", host="127.0.0.1", port=8888, reload=True)

除此外, 你還可以使用模板引擎, 如: jinja2, 使用方式如下

1. 安裝jinja2

   $pip install jinja2
   

2. fastapi-jinja2.py

   from fastapi import FastAPI, Request
   from fastapi.responses import HTMLResponse
   from fastapi.templating import Jinja2Templates
   
   app = FastAPI()
   
   # 設置template目錄
   templates = Jinja2Templates(directory="templates")
   
   
   # 設置response_class
   @app.get("/", response_class=HTMLResponse)
   async def root(request: Request):
   	data = {
   		"id": 1,
   		"name": "lczmx",
   		"message": "hello world",
   		"tags": ["tag1", "tag2", "tag3", "tag4"]
   	}
   
   	# !!! 必須帶上request
   	return templates.TemplateResponse("index.html", {"request": request, "data": data})
   
   
   if __name__ == '__main__':
   	import uvicorn
   
   	uvicorn.run(app="fastapi-jinja2:app", host="0.0.0.0", port=8000, reload=True)
   
   

3. templates/index.html

   <!DOCTYPE html>
   <html lang="en">
   <head>
   	
   	<title>Title</title>
   </head>
   <body>
   <p>id: {{ data.id}}</p>
   <p>name: {{ data.name}}</p>
   <p>message: {{ data.message}}</p>
   
   {% for tag in data.tags %}
   <li>{{ tag }}</li>
   {% endfor %}
   </body>
   </html>
   

   假如需要靜態文件, 可以這樣寫:

   <link href="{{ url_for('static', path='/styles.css') }}" rel="stylesheet">
   

   關於jinja2的一般語法, 見: 模板引擎

靜態文件

需要設置靜態文件的路徑

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# 訪問/static/xxx 時 會找 伺服器的statics/xxx
app.mount("/static", StaticFiles(directory="statics"), name="statics")

if __name__ == '__main__':
    import uvicorn

    uvicorn.run("fast_test:app", host="127.0.0.1", port=8888, reload=True)

內部調用的是starlette.staticfiles

重定向

默認307狀態碼 (臨時重定向)

from fastapi import FastAPI, Response
from fastapi.responses import RedirectResponse

app = FastAPI()


@app.get("/")
async def index_redirect():
    """
    url 要跳轉的url
    status_code 狀態碼 默認307
    headers 響應頭
    background 後台任務

    """
    return RedirectResponse("/index")

迭代返迴流式傳輸響應主體

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


async def fake_video_streamer():
    """假裝讀取影片文件, 並yield"""
    for i in range(10):
        yield b"some fake video bytes"


@app.get("/")
async def main():
    return StreamingResponse(fake_video_streamer())


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app="test:app", host="127.0.0.1", port=8000, reload=True)

非同步傳輸文件

from fastapi import FastAPI
from fastapi.responses import FileResponse

# 文件路徑
some_file_path = "large-video-file.mp4"
app = FastAPI()


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app="test:app", host="127.0.0.1", port=8000, reload=True)

異常處理

主動觸發異常

觸發的是用戶的異常, 即以4開頭的狀態碼

例子:

from fastapi import FastAPI, Path, HTTPException, status

app = FastAPI()
book_data = {
    1: {
        "name": "book1",
        "price": 88
    },
    2: {
        "name": "book2",
        "price": 89
    },
    3: {
        "name": "book3",
        "price": 99
    }
}


@app.get("/books/{book_id}")
def book_retrieve(book_id: int):
    book_item = book_data.get(book_id)
    if not book_item:
        # 不存在的book id
        # 主動拋出HTTPException

        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,

                            #  訂製detail資訊和響應頭
                            detail="不存在book id",
                            headers={"X-Error": "book not exists error"})
    return book_item


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

使用pycharm HTTP Client發送數據:

### 請求1
GET //localhost:8000/books/1

### 請求2
GET //localhost:8000/books/4

響應數據

GET //localhost:8000/books/1

HTTP/1.1 200 OK
date: Sat, 06 Nov 2021 16:04:45 GMT
server: uvicorn
content-length: 27
content-type: application/json

{
  "name": "book1",
  "price": 88
}


GET //localhost:8000/books/4

HTTP/1.1 404 Not Found
date: Sat, 06 Nov 2021 16:02:52 GMT
server: uvicorn
x-error: book not exists error
content-length: 29
content-type: application/json

{
  "detail": "不存在book id"
}

自定義異常處理器

步驟:

1. 定義異常類
2. 添加異常處理器

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

app = FastAPI()
book_data = {
    1: {
        "name": "book1",
        "price": 88
    },
    2: {
        "name": "book2",
        "price": 89
    },
    3: {
        "name": "book3",
        "price": 99
    }
}


# 自定義異常類
class NotFoundException(Exception):
    def __init__(self, name):
        self.name = name


# 自定義異常處理器 即處理函數
@app.exception_handler(NotFoundException)
def not_found_handler(request: Request, exc: NotFoundException):
    content = {
        "status": False,
        "message": f"{exc.name} not exists"

    }
    return JSONResponse(status_code=status.HTTP_404_NOT_FOUND,
                        content=content,
                        headers={"X-Error": "not exists error"})


@app.get("/books/{book_id}")
def book_retrieve(book_id: int):
    book_item = book_data.get(book_id)
    if not book_item:
	    # 主動拋出異常
        raise NotFoundException("book id")
    return book_item


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

使用pycharm HTTP Client發送數據:

GET //localhost:8000/books/4

響應數據:

GET //localhost:8000/books/4

HTTP/1.1 404 Not Found
date: Sat, 06 Nov 2021 16:31:40 GMT
server: uvicorn
x-error: not exists error
content-length: 47
content-type: application/json

{
  "status": false,
  "message": "book id not exists"
}

只要觸發了exception_handler中綁定的異常, 就會調用對應的處理函數

修改內置異常處理器

FastAPI 自帶了一些默認異常處理器, 在執行過程中碰到異常時, FastAPI就會根據這些異常處理器處理異常並返回數據

內置異常類, 位於 fastapi.exceptions

關於 ValidationError與 RequestValidationError的關係, 見官網的介紹: RequestValidationError vs ValidationError

內置異常處理器, 位於fastapi.exception_handlers

from fastapi import FastAPI
from fastapi.exceptions import HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()


# 只需要將內置異常類, 添加到異常處理器字典即可
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    content = {
        "status": False,
        "detail": str(exc.detail)
    }
    return JSONResponse(content, status_code=exc.status_code)


@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("test:app", host="127.0.0.1", port=8000, reload=True)

可以與原異常處理器配合使用, return await http_exception_handler(request, exc)這樣使用即可

關於ValidationError的屬性, 見: pydantic官網

數據轉換

FastAPI提供了將其他數據類型轉化為JSON兼容的數據類型的函數: fastapi.encoders.jsonable_encoder
根據源碼, jsonable_encoder提供了以下類型的數據的轉換:

pydantic.BaseModel

dataclasses

enum.Enum

pathlib.PurePath

str, int, float, type(None)

dict

list, set, frozenset, types.GeneratorType, tuple

一般使用

from typing import List, Optional

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()


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


@app.get("/item")
async def read_item():
    data = {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []}
    data_dict = jsonable_encoder(Item(**data))
    print(type(data_dict))  # <class 'dict'>
    return data_dict


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

其他參數

jsonable_encoder有很多參數, 部分參數和get/post/put/delete等方法的參數類似, 見: 為輸出模型作限定

• include 只返回某些欄位

• exclude 不返回某些欄位

• by_alias 欄位別名是否應該用作返回字典中的鍵

• exclude_unset 不返回默認值

• exclude_defaults 不返回與默認值相同的值

• exclude_none 不返回為None的值

• custom_encoder 指定自定義的編碼器
  先看看調用custom_encoder的源碼:

  if custom_encoder:
  	if type(obj) in custom_encoder:
  		return custom_encoder[type(obj)](obj)
  	else:
  		for encoder_type, encoder in custom_encoder.items():
  			if isinstance(obj, encoder_type):
  				return encoder(obj)
  

  也就是說custom_encoder應該是dict, key為類型, value為具體的處理函數
  例子:

  from typing import Optional
  
  from fastapi.encoders import jsonable_encoder
  from pydantic import BaseModel
  
  
  class BookItem(BaseModel):
  	name: Optional[str] = None
  	price: Optional[float] = None
  
  
  class AuthorClass:
  	def __init__(self, name: str, age: int):
  		self.name = name
  		self.age = age
  
  	def __str__(self):
  		return f"{self.name} ({self.age})"
  
  	def __repr__(self):
  		return self.__str__()
  
  
  # 自定義的編碼器
  # 將類屬性轉換為字典
  custom_encoder = {
  	AuthorClass: lambda obj: {"name": obj.name, "age": obj.age}
  }
  
  book_data = BookItem(**{"name": "book1", "price": 50.2}).dict()
  author_instance = AuthorClass(name="lczmx", age=18)
  # 更新數據
  book_data.update({"author": author_instance})
  
  print(book_data)
  # {'name': 'book1', 'price': 50.2, 'author': lczmx (18)}
  
  data_dict = jsonable_encoder(book_data, custom_encoder=custom_encoder)
  print(data_dict)
  # {'name': 'book1', 'price': 50.2, 'author': {'name': 'lczmx', 'age': 18}}
  
  

  你亦可以在BaseModel中指定json_encoders作為編碼器, 若想知道如何使用見: json_encoders

• sqlalchemy_safe 暫不知道該參數有什麼用 (待補充)

ORM

下面舉一個完整的項目, 說明如何在FastAPI中使用ORM
使用的是SQLAlchemy這個框架

項目結構

+--- test_app
|   +--- __init__.py
|   +--- crud.py
|   +--- database.py
|   +--- main.py
|   +--- models.py
|   +--- schemas.py
+--- run.py

項目依賴:

fastapi==0.63.0
pydantic==1.7.3
requests==2.25.1
SQLAlchemy==1.3.22

程式碼

• run.py程式的入口

  import uvicorn
  from fastapi import FastAPI
  
  from test_app import application
  
  app = FastAPI(
     title='Fast ORM 測試',
     description='FastAPI 使用SQlAlchemy框架',
     version='1.0.0',
     docs_url='/docs',
     redoc_url='/redocs',
  )
  
  app.include_router(application, prefix='/test_app', tags=['FastAPI ORM'])
  
  if __name__ == '__main__':
     uvicorn.run('run:app', host='0.0.0.0', port=8000, reload=True, debug=True, workers=1)
  
  
  

• test_app/__init__.py 用作run.py導入

  from .main import application
  

• test_app/database.py 用於創建連接和生成創建表的公共基類

  from sqlalchemy import create_engine
  from sqlalchemy.ext.declarative import declarative_base
  from sqlalchemy.orm import sessionmaker
  
  SQLALCHEMY_DATABASE_URL = 'sqlite:///./coronavirus.sqlite3'
  # MySQL或PostgreSQL的連接方法:
  # SQLALCHEMY_DATABASE_URL = "postgresql://username:password@host:port/database_name"
  
  engine = create_engine(
  	# echo=True表示引擎將用repr()函數記錄所有語句及其參數列表到日誌
  	# 由於SQLAlchemy是多執行緒，指定check_same_thread=False來讓建立的對象任意執行緒都可使用。這個參數只在用SQLite資料庫時設置
  	SQLALCHEMY_DATABASE_URL, encoding='utf-8', echo=True, connect_args={'check_same_thread': False}
  )
  
  # 在SQLAlchemy中，CRUD都是通過會話(session)進行的，所以我們必須要先創建會話，每一個SessionLocal實例就是一個資料庫session
  # flush()是指發送資料庫語句到資料庫，但資料庫不一定執行寫入磁碟；commit()是指提交事務，將變更保存到資料庫文件
  SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=True)
  
  # 創建基本映射類
  Base = declarative_base(bind=engine, name='Base')
  
  

• test_app/crud.py 用於增刪改查

  """
  數據增刪改查介面
  """
  from sqlalchemy.orm import Session
  
  from test_app import models, schemas
  
  
  def get_city(db: Session, city_id: int):
  	return db.query(models.City).filter(models.City.id == city_id).first()
  
  
  def get_city_by_name(db: Session, name: str):
  	return db.query(models.City).filter(models.City.province == name).first()
  
  
  def get_cities(db: Session, skip: int = 0, limit: int = 10):
  	return db.query(models.City).offset(skip).limit(limit).all()
  
  
  def create_city(db: Session, city: schemas.CreateCity):
  	db_city = models.City(**city.dict())
  	db.add(db_city)
  	db.commit()
  	db.refresh(db_city)
  	return db_city
  
  
  def get_data(db: Session, city: str = None, skip: int = 0, limit: int = 10):
  	if city:
  		return db.query(models.Data).filter(
  			models.Data.city.has(province=city))  # 外鍵關聯查詢，這裡不是像Django ORM那樣Data.city.province
  	return db.query(models.Data).offset(skip).limit(limit).all()
  
  
  def create_city_data(db: Session, data: schemas.CreateData, city_id: int):
  	db_data = models.Data(**data.dict(), city_id=city_id)
  	db.add(db_data)
  	db.commit()
  	db.refresh(db_data)
  	return db_data
  
  

• test_app/schemas.py定義 傳入或返回的數據

  from datetime import date as date_
  from datetime import datetime
  
  from pydantic import BaseModel
  
  
  class CreateData(BaseModel):
  	date: date_
  	confirmed: int = 0
  	deaths: int = 0
  	recovered: int = 0
  
  
  class CreateCity(BaseModel):
  	province: str
  	country: str
  	country_code: str
  	country_population: int
  
  
  class ReadData(CreateData):
  	id: int
  	city_id: int
  	updated_at: datetime
  	created_at: datetime
  
  	class Config:
  		orm_mode = True
  
  
  class ReadCity(CreateCity):
  	id: int
  	updated_at: datetime
  	created_at: datetime
  
  	class Config:
  		orm_mode = True
  
  

• test_app/main.py 定義網站的邏輯程式碼

  from typing import List
  import requests
  from pydantic import HttpUrl
  from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks
  
  from sqlalchemy.orm import Session
  
  from test_app import crud, schemas
  from test_app.database import engine, Base, SessionLocal
  from test_app.models import City, Data
  
  application = APIRouter()
  
  # 創建表
  Base.metadata.create_all(bind=engine)
  
  
  def get_db():
  	db = SessionLocal()
  	try:
  		yield db
  	finally:
  		db.close()
  
  
  def bg_task(url: HttpUrl, db: Session):
  	"""創建數據
  	根據返回數據解析成 需要的格式
  	"""
  	city_data = requests.get(url=f"{url}?source=jhu&country_code=CN&timelines=false")
  
  	if 200 == city_data.status_code:
  		db.query(City).delete()  # 同步數據前先清空原有的數據
  		for location in city_data.json()["locations"]:
  			city = {
  				"province": location["province"],
  				"country": location["country"],
  				"country_code": "CN",
  				"country_population": location["country_population"]
  			}
  			crud.create_city(db=db, city=schemas.CreateCity(**city))
  
  	coronavirus_data = requests.get(url=f"{url}?source=jhu&country_code=CN&timelines=true")
  
  	if 200 == coronavirus_data.status_code:
  		db.query(Data).delete()
  		for city in coronavirus_data.json()["locations"]:
  			db_city = crud.get_city_by_name(db=db, name=city["province"])
  			for date, confirmed in city["timelines"]["confirmed"]["timeline"].items():
  				data = {
  					"date": date.split("T")[0],  # 把'2020-12-31T00:00:00Z' 變成 『2020-12-31』
  					"confirmed": confirmed,
  					"deaths": city["timelines"]["deaths"]["timeline"][date],
  					"recovered": 0  # 每個城市每天有多少人痊癒，這種數據沒有
  				}
  				# 這個city_id是city表中的主鍵ID，不是coronavirus_data數據里的ID
  				crud.create_city_data(db=db, data=schemas.CreateData(**data), city_id=db_city.id)
  
  
  @application.get("/gen_data/jhu", description="在後台生成數據")
  def gen_data(background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
  	"""在後灘自動生成數據"""
  	background_tasks.add_task(bg_task, "//coronavirus-tracker-api.herokuapp.com/v2/locations", db)
  	return {"message": "正在後台同步數據..."}
  
  
  @application.post("/create_city", response_model=schemas.ReadCity, description="創建一個城市數據")
  def create_city(city: schemas.CreateCity, db: Session = Depends(get_db)):
  	db_city = crud.get_city_by_name(db, name=city.province)
  	if db_city:
  		raise HTTPException(status_code=400, detail="City already registered")
  	return crud.create_city(db=db, city=city)
  
  
  @application.get("/get_city/{city}", response_model=schemas.ReadCity, description="獲取一個城市的數據")
  def get_city(city: str, db: Session = Depends(get_db)):
  	db_city = crud.get_city_by_name(db, name=city)
  	if db_city is None:
  		raise HTTPException(status_code=404, detail="City not found")
  	return db_city
  
  
  @application.get("/get_cities", response_model=List[schemas.ReadCity], description="獲取全部城市的數據")
  def get_cities(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
  	cities = crud.get_cities(db, skip=skip, limit=limit)
  	return cities
  
  
  @application.post("/create_data", response_model=schemas.ReadData, description="創建一個城市的數據")
  def create_data_for_city(city: str, data: schemas.CreateData, db: Session = Depends(get_db)):
  	db_city = crud.get_city_by_name(db, name=city)
  	data = crud.create_city_data(db=db, data=data, city_id=db_city.id)
  	return data
  
  
  @application.get("/get_data", description="獲取一個城市的數據")
  def get_data(city: str = None, skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
  	data = crud.get_data(db, city=city, skip=skip, limit=limit)
  	return data
  
  

認證

即確認, 你到底是不是你?

OAUTH2.0

OAuth是一個驗證授權(Authorization)的開放標準, 詳情見: 理解OAuth 2.0

OAuth2的授權原理圖:

OAuth2.0的授權模式有三種:

1. 授權碼模式 Authoriztion Code Grant
2. 隱授權碼模式 Implicit Grant
3. 密碼授權模式 Resource Owner Password Credentials Grant
4. 客戶端憑證授權模式 client Credentials Grant

這裡的例子用的是第三種模式: 密碼授權模式

使用密碼授權模式需要兩個類:

1. fastapi.security.OAuth2PasswordBearer
OAuth2PasswordBearer是接收URL作為參數的一個類, 這並 不會 創建相應的URL路徑操作，只是指明客戶端用來請求Token的URL地址
   客戶端會向該URL發送username和password參數，然後得到一個Token值
   作為依賴注入時, 表明該URL需要進行驗證: 當請求到來的時候，FastAPI會檢查請求的Authorization頭資訊,
   若: 無Authorization頭資訊，或者頭資訊的內容不是Bearer token, 它會拋出異常:

   raise HTTPException(
   	status_code=HTTP_401_UNAUTHORIZED,
   	detail="Not authenticated",
   	headers={"WWW-Authenticate": "Bearer"},
   )
   

   檢驗成功返回token
   注: 沒有這檢驗token的合法性, 只是檢驗有無請求頭, 所以需要我們手寫檢驗token的邏輯!!

2. fastapi.security.OAuth2PasswordRequestForm
OAuth2PasswordRequestForm可用於接收登錄數據, 數據類型為Form, 即application/x-www-form-urlencoded

   OAuth2PasswordRequestForm的欄位有:

   • grant_type 授權模式, passwrod
   • username 登陸的用戶名
   • password 登陸的密碼
   • scope 用來限制客戶端的訪問範圍，如果為空（默認）的話，那麼客戶端擁有全部的訪問範圍
     格式形如: items:read items:write users:read profile openid
   • client_id 客戶端密鑰
   • client_secret 客戶端ID

例子:

from typing import Optional

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

app = FastAPI()

# 告知客戶端 請求Token的URL地址是 /token
oauth2_schema = OAuth2PasswordBearer(tokenUrl="/token")

# 模擬資料庫的數據
fake_users_db = {
    "john snow": {
        "username": "john snow",
        "full_name": "John Snow",
        "email": "[email protected]",
        "hashed_password": "fakehashedsecret",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "[email protected]",
        "hashed_password": "fakehashedsecret2",
        "disabled": True,
    },
}


# hash 密碼
def fake_hash_password(password: str):
    return "fakehashed" + password


class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None


class UserInDB(User):
    hashed_password: str


# 登錄
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user_dict = fake_users_db.get(form_data.username)
    if not user_dict:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect username or password")
    user = UserInDB(**user_dict)
    hashed_password = fake_hash_password(form_data.password)
    # 檢驗密碼
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect username or password")
    return {"access_token": user.username, "token_type": "bearer"}


# 獲取用戶
def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


# 檢驗token的合法性
def fake_decode_token(token: str):
    user = get_user(fake_users_db, token)
    return user


# 檢驗是否 已經驗證了
async def get_current_user(token: str = Depends(oauth2_schema)):
    # 這裡的token是用戶名
    user = fake_decode_token(token)
    if not user:
        # UNAUTHORIZED 的 固定寫法
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            # OAuth2的規範，如果認證失敗，請求頭中返回「WWW-Authenticate」
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user


async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user")
    return current_user


# 獲得 active的用戶
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("fastapi-test:app", port=8000, reload=True)

主要注意/users/me和/token路由, 以及fake_decode_token函數, 上面程式碼看起來比較複雜, 只是由於使用了依賴注入 一層套一層而已.

JWT

JWT介紹

jwt是我們常用的認證方式, jwt由三部分組成: 頭部 (header) 載荷 (payload) 簽證 (signature)

1. 頭部 header
   jwt的頭部承載兩部分資訊: 聲明類型和聲明加密的演算法, 形如:

   {
   	'typ': 'JWT',
   	'alg': 'HS256'
   }
   

   然後將頭部進行base64加密, 變為: eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9

2. 載荷 payload
   載荷就是存放有效資訊的地方, 即我們存放數據的地方, 由三部分組成: 標準中註冊的聲明 公共的聲明 私有的聲明

   標準中註冊的聲明, 即已經預定的標識

   名稱 key	描述
   iss	jwt簽發者
   sub	jwt所面向的用戶
   aud	接收jwt的一方
   exp	jwt的過期時間，這個過期時間必須要大於簽發時間
   nbf	定義在什麼時間之前，該jwt都是不可用的
   iat	jwt的簽發時間
   jti	jwt的唯一身份標識，主要用來作為一次性token, 從而迴避重放攻擊

   公共的聲明
   公共的聲明可以添加任何的資訊, 一般添加用戶的相關資訊或其他業務需要的必要資訊

   私有的聲明
   私有聲明是提供者和消費者所共同定義的聲明

   不建議在JWT中存放敏感資訊, 因為base64是對稱解密的, 意味著該部分資訊可以歸類為明文資訊

   假如payload數據為:

   {
   "sub": "1234567890",
   "name": "John Doe",
   "admin": true
   }
   

   對其進行base64加密, 得到: eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiYWRtaW4iOnRydWV9

3. 簽證 signature
   即對數據的簽證, 由三部分組成: header (base64後的) payload (base64後的) secret

   這個部分需要base64加密後的header和base64加密後的payload連接組成的字元串
   然後通過header中聲明的加密方式進行加鹽secret組合加密，然後就構成了jwt的第三部分

最終得到jwt: header.payload.signature
訪問時通過指定請求頭Authorization: Bearer token訪問伺服器.

安裝依賴

安裝生成和校驗 JWT 令牌的庫:

$pip install python-jose[cryptography]

安裝生成hash密碼的庫:

$pip install passlib[bcrypt]

passlib一般使用

from passlib.context import CryptContext

# 加密演算法為: bcrypt, 沒有安裝的話需要 pip install bcrypt
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# 獲得hash後的密文
password = "123456"

# hash(self, secret, scheme=None, category=None):
hash_str = pwd_context.hash(password)
print(f"hash password {hash_str}")

# 檢驗密碼是否符合
# verify(self, secret, hash, scheme=None, category=None)
is_verify = pwd_context.verify(password, hash_str)
print(f"is verify? {is_verify}")

FastAPI使用JWT

步驟:

1. 生成秘鑰
2. 定義加密演算法和令牌過期時間
3. 指定哈希加密演算法和token url
4. 調用jwt.encode生成jwt
5. 通過依賴注入獲取jwt令牌

你需要先安裝依賴, 如上文
生成安全秘鑰:

$openssl rand -hex 32
09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7

例子:

from datetime import datetime, timedelta
from typing import Optional

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

app = FastAPI()

SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"  # jwt加密演算法
ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 訪問令牌過期分鐘
# 模擬當前用戶數據
fake_users_db = {
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "[email protected]",
        "hashed_password": "fakehashedsecret2",
        "disabled": True,
    },
    "john snow": {
        "username": "john snow",
        "full_name": "John Snow",
        "email": "[email protected]",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    }
}


class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None


class UserInDB(User):
    hashed_password: str


class Token(BaseModel):
    """返回給用戶的Token"""
    access_token: str
    token_type: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_schema = OAuth2PasswordBearer(tokenUrl="/jwt/token")


def verity_password(plain_password: str, hashed_password: str):
    """對密碼進行校驗"""
    return pwd_context.verify(plain_password, hashed_password)


def jwt_get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


# 檢驗jwt是否合法
def jwt_authenticate_user(db, username: str, password: str):
    # 獲取當前用戶
    user = jwt_get_user(db=db, username=username)
    if not user:
        return False
    # 檢驗密碼是否合法
    if not verity_password(plain_password=password, hashed_password=user.hashed_password):
        return False
    return user


# 生成jwt token
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    # data => payload
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    # 標準中註冊的聲明 過期時間
    to_encode.update({"exp": expire})

    # jwt.encode 的參數
    # claims     指定payload
    # key        指定signature的加密秘鑰
    # algorithm  指定signature的加密演算法
    encoded_jwt = jwt.encode(claims=to_encode, key=SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


@app.post("/jwt/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    """
    登錄 返回 jwt token
    通過依賴注入 OAuth2PasswordRequestForm
    獲得 username 和 password
    """
    user = jwt_authenticate_user(db=fake_users_db, username=form_data.username, password=form_data.password)
    if not user:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}


async def jwt_get_current_user(token: str = Depends(oauth2_schema)):
    """
    獲取當前請求的jwt token
    通過 OAuth2PasswordBearer 獲得
    """
    credentials_exception = HTTPException(
        status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # 獲取 數據

        # decode jwt token
        # 得到payload, 即 create_access_token 中的 to_encode
        payload = jwt.decode(token=token, key=SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = jwt_get_user(db=fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user


# 獲取 active用戶
async def jwt_get_current_active_user(current_user: User = Depends(jwt_get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user")
    return current_user


@app.get("/jwt/users/me")
async def jwt_read_users_me(current_user: User = Depends(jwt_get_current_active_user)):
    return current_user


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("fastapi-test:app", port=8000, reload=True)

這個例子的 username為john snow, password為 secret
訪問時通過指定請求頭Authorization: Bearer token訪問伺服器

session

即使用傳統的session-cookie方式進行認證, FastAPI用於前後端分離的項目居多, 所以不舉例子了
總的來說, 你需要Starlette的SessionMiddleware中間件, 然後通過request.session獲取session

關於SessionMiddleware, 見: SessionMiddleware
第三方SessionMiddleware庫: starsessions

許可權

即確認, 你能不能訪問?

一般通過依賴注入完成簡單的許可權驗證
例子 (用戶名: alice和john, 密碼都為123456):

from datetime import datetime, timedelta
from typing import Optional, List

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

app = FastAPI()

SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"  # jwt加密演算法
ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 訪問令牌過期分鐘
# 模擬當前用戶數據
fake_users_db = {
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "[email protected]",
        "hashed_password": "$2b$12$tCUwz5MrDTgnugd3AKBBr..jZpFBRBIc321iBrbmEA3flPaxWmMwO",
        "disabled": True,
        "role": ["role1"]
    },
    "john": {
        "username": "john",
        "full_name": "John",
        "email": "[email protected]",
        "hashed_password": "$2b$12$Z5xEfIb1sD487A8IdT3.seUGaBAIVpZtwe5/MXhLu4dKzhaeiF.OC",
        "disabled": True,
        "role": ["role2"]
    }
}


class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None
    role: List[str]


class UserInDB(User):
    hashed_password: str


class Token(BaseModel):
    """返回給用戶的Token"""
    access_token: str
    token_type: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_schema = OAuth2PasswordBearer(tokenUrl="/jwt/token")


def verity_password(plain_password: str, hashed_password: str):
    """對密碼進行校驗"""
    return pwd_context.verify(plain_password, hashed_password)


def jwt_get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


# 檢驗用戶名和密碼是否合法
def jwt_authenticate_user(db, username: str, password: str):
    # 獲取當前用戶
    user = jwt_get_user(db=db, username=username)
    hash_str = pwd_context.hash(password)

    if not user:
        return False
    # 檢驗密碼是否合法
    if not verity_password(plain_password=password, hashed_password=user.hashed_password):
        return False

    return user


# 生成jwt token
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(claims=to_encode, key=SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


@app.post("/jwt/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    """
    生成jwt token
    """
    user = jwt_authenticate_user(db=fake_users_db, username=form_data.username, password=form_data.password)
    if not user:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}


async def jwt_get_current_user(token: str = Depends(oauth2_schema)):
    """
    獲取當前已經登陸的用戶數據
    """
    credentials_exception = HTTPException(
        status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token=token, key=SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = jwt_get_user(db=fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user


async def verify_user(user: UserInDB = Depends(jwt_get_current_user)):
    """
    驗證當前用戶是否可以訪問
    """
	# 通過判斷角色來判斷是否有無訪問許可權
    if "role1" not in user.role:
        # 檢驗不可以訪問
        raise HTTPException(
            status_code=403, detail="Forbidden"
        )


# 通過以來注入的方式
@app.get("/items", dependencies=[Depends(verify_user)])
async def get_items():
    return {"data": "items"}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("fastapi-test:app", port=8000, reload=True)

以上例子中, 使用JWT認證用戶, 登錄alice可以訪問/items, 而john無法訪問/items

Cookie

設置

調用response.set_cookie方法
不主動返回response時, 需要在參數中指定Response參數, 否則會解析成查詢參數

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

app = FastAPI()


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


# !!!!!!!! 返回response
@app.post("/cookie/")
def create_cookie():
    content = {"message": "Come to the dark side, we have cookies"}
    response = JSONResponse(content=content)
    response.set_cookie(key="fakesession", value="fake-cookie-session-value")
    return response


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

set_cookie參數:

獲取

Cookie指定要獲取的cookie
注: Cookie是Param的子類, 具有通用的方法, 更多參數見: Param

from typing import Optional

from fastapi import FastAPI, Cookie

app = FastAPI()


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


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

刪除

調用response.delete_cookie方法

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

app = FastAPI()


# !!!!!!!! 不返回response
@app.post("/cookie-and-object/")
def create_cookie(response: Response):
    response.delete_cookie(key="fakesession")
    return {"message": "Come to the dark side, we have cookies"}


# !!!!!!!! 返回response
@app.post("/cookie/")
def create_cookie():
    content = {"message": "Come to the dark side, we have cookies"}
    response = JSONResponse(content=content)
    response.delete_cookie(key="fakesession")
    return response


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

delete_cookie參數:

delete_cookie源碼:

def delete_cookie(self, key: str, path: str = "/", domain: str = None) -> None:
	self.set_cookie(key, expires=0, max_age=0, path=path, domain=domain)

Header

設置

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

app = FastAPI()


# !!!!!!!! 不返回response
@app.get("/headers-and-object/")
def set_headers(response: Response):
    response.headers["X-Cat-Dog"] = "alone in the world"
    return {"message": "Hello World"}


# !!!!!!!! 返回response
@app.get("/headers/")
def set_headers():
    content = {"message": "Hello World"}
    headers = {"X-Cat-Dog": "alone in the world", "Content-Language": "en-US"}
    return JSONResponse(content=content, headers=headers)


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

獲取

通過Header指定要獲取的header
注: Header是Param的子類, 具有通用的方法, 更多參數見: Param

注意: HTTP Header的名稱使用-相連, 不符合python變數命名規則, 故FastAPI會將_轉化為-, 如user_agent==>user-agent
一個Header多個值時, 可以使用List接收, 如: x_token: Optional[List[str]] = Header(None)

from typing import Optional

from fastapi import FastAPI, Header

app = FastAPI()


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


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

刪除

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

app = FastAPI()


# !!!!!!!! 不返回response
@app.get("/headers-and-object/")
def delete_headers(response: Response):
    del response.headers["X-Cat-Dog"]
    return {"message": "Hello World"}


# !!!!!!!! 返回response
@app.get("/headers/")
def delete_headers():
    content = {"message": "Hello World"}
    response = JSONResponse(content=content)
    del response.headers["X-Cat-Dog"]
    return response


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("test:app", host="127.0.0.1", port=8000, reload=True)

依賴注入

所謂依賴注入就是 我們在運行程式碼過程中要用到其他依賴 或 子函數 時， 可以在函數定義時聲明

理解起來有點抽象, 就算看了官方文檔的例子也會讓人覺得費解: 明明不用依賴注入也可以做到, 為什麼額外定義一個”依賴”來使用呢?
按我的理解, 依賴注入有以下好處, 值得我們花費時間學習:

依賴注入主要的作用是解耦、 驗證和提高復用率
我們之前使用FastAPI時的主要步驟就是: 1. 定義一堆參數 2. 將參數在函數中接收 3. 在函數中使用
但是, 假如我們需要替換函數中的處理邏輯呢? 那不是整個函數的一部分要重寫, 假如是一個函數還好, 但很多個函數都要修改的話就比較麻煩了.
而且, 假如我們需要為某個鏈接添加某些許可權時, 也不能每次都在函數處理吧.

也就是說: 有了依賴注入，原本接受各種參數來構造一個對象，現在只接受是已經實例化的對象就行了。而且還可在實例化的過程中進行驗證, 如何構造就要看依賴注入中的函數實現了。

使用場景:

1. 共享業務邏輯 (復用相同的程式碼邏輯)
2. 共享資料庫連接
3. 實現安全、驗證、角色許可權
4. 等…

一般使用

舉幾個例子說明依賴注入的一般使用方式。

資料庫連接例子

使用SQLAlchemy連接MYSQL資料庫, 並通過上下文管理協議自動斷開資料庫連接

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy import create_engine

application = APIRouter()

SQLALCHEMY_DATABASE_URL = 'sqlite:///./coronavirus.sqlite3'

engine = create_engine(SQLALCHEMY_DATABASE_URL, encoding='utf-8', echo=True, connect_args={'check_same_thread': False})

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=True)


# 一般來說SessionLocal是從其他py文件中導入
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# 通過依賴注入獲取資料庫session
@application.post("/data")
def get_data(db: Session = Depends(get_db)):
    """
    通過db操作資料庫
    """
    return {}

用到了yield的依賴

許可權驗證例子

一般來說是給路徑注入依賴, 詳見: 許可權

後台任務例子

from fastapi import BackgroundTasks, FastAPI, Depends
from typing import Optional

app = FastAPI()


def write_notification(email: str, message=""):
    # 後台任務的函數為正常的函數
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


def dependency_email(background_tasks: BackgroundTasks, email: Optional[str] = None):
    if email:
        # 添加到後台任務
        background_tasks.add_task(write_notification, email, message="some notification")
    return email


@app.post("/send-notification/")
async def send_notification(email: str = Depends(dependency_email)):
    return {"message": "Notification sent in the background"}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app="test:app", port=8000, reload=True)

類作為依賴

from fastapi import FastAPI, Depends
from typing import Optional

app = FastAPI()


# 定義類依賴
class CommonQueryParams:
    def __init__(self, query: Optional[str] = None, page: int = 1, limit: int = 10):
        self.query = query
        self.page = page
        self.limit = limit


# 使用依賴
@app.get("/")
# 第一種寫法， 比較簡單， 但無法讓ide .出來
# async def index(params=Depends(CommonQueryParams)):

# 第二種寫法，比較複雜， 可以讓ide .出來
# async def index(params: CommonQueryParams = Depends(CommonQueryParams)):

# 第三種寫法，推薦 相當於第二種寫法的縮寫
async def index(params: CommonQueryParams = Depends()):
    return {"params": params}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("fastapi-test:app", port=8000, reload=True)

子依賴

子依賴， 即一個依賴作為其他依賴的參數。

from fastapi import FastAPI, Depends
from typing import Dict

app = FastAPI()


# 子依賴
async def dependency_query(query: str):
    return query


# 在依賴中使用其他依賴
# * ： 將後面的參數變成關鍵字參數
async def sub_dependency_item(*, query: str = Depends(dependency_query), limit: int, skip: int):
    return {
        "query": query,
        "limit": limit,
        "skip": skip,
    }


# 使用依賴
@app.get("/")
def index(params: Dict = Depends(sub_dependency_item)):
    data = {
        "index": "/"
    }
    data.update({"params": params})
    return data


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("fastapi-test:app", port=8000)

路徑依賴

單個路徑的依賴, 即給get/post等添加依賴
給出官方的例子：

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"}]

通過dependencies參數指定，Depends指定依賴

全局依賴

所謂的全局依賴就是給FastAPI和APIRouter添加依賴（通過dependencies參數指定）

from fastapi import FastAPI, Header, Depends, APIRouter


async def global_dependency(x_token: str = Header(..., alias="x-token")):
    # 獲取x-token 請求頭 並 列印
    print(x_token)


# 方式一 FastAPI dependencies參數
app = FastAPI(dependencies=[Depends(global_dependency)])

# 方式二 APIRouter dependencies參數
music_router = APIRouter(prefix="/music", dependencies=[Depends(global_dependency)])


@music_router.get("/")
def index():
    return {"x_token": "1234"}


# 注意 app.include_router 需要在後面， 否則無法導入之前定義的 路由
app.include_router(music_router)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run("fastapi-test:app", port=8000)

yield的依賴注入

我們可以通過yield的依賴，讓其變成上下文管理協議 （利用contextlib.contextmanager和contextlib.asynccontextmanager），上下文管理協議可以讓我們更好地管理資源

例子見上文的: 資料庫連接例子

自定義介面文檔

FastAPI可以自動生成文檔, 你可以訪問連接, /docs (Swagger UI)或/redoc (ReDoc)

文檔資訊

本部分內容包括:

1. 文檔的標題: title
2. 文檔的描述: description
3. 文檔的版本: version
4. 文檔的json路徑: openapi_url
5. 應用的服務條款: terms_of_service
6. 應用的聯繫資訊: contact
7. 應用的許可資訊: license_info
8. 應用的服務列表: servers

例子:

from fastapi import FastAPI


# 聯繫資訊 數據
contact = {
    # 聯繫的名字
    "name": "聯繫名字",
    # 聯繫url
    "url": "//x-force.example.com/contact/",
    # 聯繫的郵箱
    "email": "[email protected]",
}
# 許可資訊數據
license_info = {
    "name": "Apache 2.0",
    "url": "//www.apache.org/licenses/LICENSE-2.0.html",
}
# 服務列表數據
# 將渲染成select元素
servers = [
    # 單個元素 為option元素
    {"url": "//stag.example.com", "description": "Staging environment"},
    {"url": "//prod.example.com", "description": "Production environment"},
]
app = FastAPI(
    # 文檔的標題和描述和版本
    title="測試API", description="描述資訊數據", version="1.1",
    # 文檔的json路徑
    openapi_url="/myapi.json",
    # 文檔的服務條款URL
    terms_of_service="//example.com/terms/",
    # 文檔的聯繫資訊
    contact=contact,
    # 文檔的許可資訊
    license_info=license_info,
    # 文檔的服務列表
    servers=servers
)

標籤與標籤元數據

關於標籤與標籤元數據如下圖

1. 通過FastAPI類的openapi_tags指定標籤元數據
2. 通過APIRouter類或app.include_router或app.get/...的tags參數指定標籤

例子:

from fastapi import FastAPI

tags_metadata = [
    {
        "name": "用戶",
        "description": "操作用戶, **登錄**很重要",
    },
    {
        "name": "數據",
        "description": "管理數據",
        "externalDocs": {
            "description": "fastapi文檔",
            "url": "//fastapi.tiangolo.com/",
        },
    },
]

app = FastAPI(
    # 文檔的標籤元數據
    openapi_tags=tags_metadata)


@app.get("/app/data", tags=["數據"])
async def root():
    return {}


@app.get("/app/user", tags=["用戶"])
async def root():
    return {}

上面是通過get...實現的
下面展示在APIRouter和include_router中定義tags

from fastapi import FastAPI, APIRouter

tags_metadata = [
    {
        "name": "用戶",
        "description": "操作用戶, **登錄**很重要",
    },
    {
        "name": "數據",
        "description": "管理數據",
        "externalDocs": {
            "description": "fastapi文檔",
            "url": "//fastapi.tiangolo.com/",
        },
    },
]

app = FastAPI(
    # 文檔的標籤元數據
    openapi_tags=tags_metadata)

# ----------- APIRouter 的 tags
user_application = APIRouter(
    prefix="/user",
    tags=["用戶"]
)


@user_application.get("/")
async def user_index():
    return {}


data_application = APIRouter(
    prefix="/data",
)


@data_application.get("/")
async def data_index():
    return {}


app.include_router(user_application)

# ----------- include_router中指定 tags
app.include_router(data_application, tags=["數據"])

tags不指定時默認為default

api的概要及描述

包括當前標籤的概要以及標籤的描述資訊

from fastapi import FastAPI, APIRouter

app = FastAPI()


@app.get("/", summary="獲得主頁", description="通過xxx獲取主頁頁面")
async def index():
    return {}


@app.get("/home")
async def index_home():
    """
    獲取home主頁
    """
    return {}

以上程式碼的文檔圖片:

未指定summary時, 概要為函數名.tiltle()並替換_
未指定description時, 描述消息為函數的docstring

補充: docstring的高級用法:
即一些寫法可以被渲染, 主要有以下2個要點

1. \f換頁符, 用於截斷OpenAPI 的輸出
2. 語法為Markdown語法

例子:

from typing import Optional, Set

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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


@app.post("/items/", response_model=Item, summary="創建一個item")
async def create_item(item: Item):
    """
    創建item
    - **name**: 每個item必須要有一個name
    - **description**: item的描述資訊
    - **price**: 必需的參數
    - **tax**: 如果沒有tax參數, 你可以省略它
    - **tags**: item的標籤
    \f
    :param item: User input.
    """

    return item
	

以上程式碼的文檔圖片

api的請求參數

在FastAPI中參數類型有: 路徑參數 (Path), 查詢參數 (Query), 請求體參數 (pydantic和Body), 請求頭參數 (Header), Cookie參數 (Cookie), Form表單參數 (Form), 文件參數 (File)
它們之間的關係, 見: Params

文檔的Parameters

在文檔中的位置:

類型為Path Query Header Cookie會在這裡展示
一般來說我們只需要參數有:

• default
• alias
• description
• example
  這些參數有什麼作用, 見下文的Params

from fastapi import FastAPI
from fastapi import Path, Query, Header, Cookie

app = FastAPI()


@app.get("/data/{id}", summary="獲得數據", description="通過id獲取指定值的數據")
async def index(*,
                did: str = Path(..., description="數據ID的描述資訊",
                                example=1, regex=r"\d+", alias="id"),
                limit: int = Query(10, description="要取得的數據", example=10),
                user_agent: str = Header(..., description="瀏覽器資訊的描述資訊"),
                userid: str = Cookie(..., description="cookie的userid"),
                ):
    return {"id": did, "limit": limit, "user-agent": user_agent, "userid": userid}

以上程式碼對應的文檔:

文檔的Request body

在文檔中的位置:

類型為 pydantic模型 Body``Form File會在這裡展示
一般來說我們只需要參數有:

• default
• title
• alias
• description
• example
  這些參數有什麼作用, 見下文的Params

pydantic模型 Body, 默認類型為: application/json
假如有Form或File, Request body的類型會變為: application/x-www-form-urlencoded或multipart/form-data

例子:

from fastapi import FastAPI
from fastapi import Form, File, UploadFile

app = FastAPI()


@app.post("/update")
async def update(
        username: str = Form(..., description="用戶名的描述資訊", example="lczmx"),
        filename: UploadFile = File(..., description="文件的描述資訊")):
    return {"username": username, "filename": filename.filename}

假如只有pydantic模型和Body的話:

from fastapi import FastAPI
from fastapi import Body
from pydantic import BaseModel, Field

app = FastAPI()


class QueryItem(BaseModel):
    query: str = Field(..., title="查詢字元串", description="查詢字元串詳細資訊", example="東方")


@app.post("/search")
async def search(
        query_item: QueryItem,
        query_charset: str = Body("utf-8", title="編碼方式", description="查詢字元的編碼方式的詳細資訊")):
    return {"query": query_item.query, "query_charset": query_charset}

你還可以直接在pydantic的Config類中統一定義example

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class QueryItem(BaseModel):
    query: str
    charset: str

    class Config:
        schema_extra = {
            "example": {
                "query": "東方",
                "charset": "utf-8"
            }

        }


@app.post("/search")
async def search(query_item: QueryItem):
    return {"query": query_item.query, "query_charset": query_item.charset}

你亦可以在Body中統一定義example

from fastapi import FastAPI
from fastapi import Body
from pydantic import BaseModel

app = FastAPI()


class QueryItem(BaseModel):
    query: str
    charset: str


@app.post("/search")
async def search(
        query_item: QueryItem = Body(..., example={
            "query": "東方",
            "charset": "utf-8"
        })):
    return {"query": query_item.query, "query_charset": query_item.charset}

api的返回值

文檔見: OpenAPI Response 對象

在文檔中所在的位置

我們可以通過FastAPI或APIRouter或app.include_router或app.get...的responses參數指定返回值的資訊 (越後面優先順序越高)
responses的值為字典, key為狀態碼, value為字典 (key有model description content)

使用response_model參數可以為文檔添加狀態碼為200的響應模型
使用response_description參數, 可以為文檔添加狀態碼為200的描述資訊

例子:

from fastapi import FastAPI
from pydantic import BaseModel, Field


class ErrorMessage(BaseModel):
    code: int = Field(..., title="狀態碼", example=401)
    message: str = Field(..., title="錯誤資訊", example="Unauthorized")


class UserData(BaseModel):
    username: str = Field(..., title="用戶名", example="lczmx")
    age: int = Field(..., title="年齡", example=18)


app = FastAPI()
responses = {

    200: {
        # 使用response_model的模型
        "description": "成功響應的描述資訊",
        # 右邊的links
        "links": {"鏈接一": {"operationRef": "www.baidu.com", "description": "鏈接描述資訊"}},

    },
    401: {
        "description": "401的描述資訊",
        # 指定響應模型
        "model": ErrorMessage
    },
    404: {
        "description": "404的描述資訊",
        # 手動定義響應模型
        "content": {
            "application/json": {
                "schema": {
                    # 全部模型都在 #/components/schemas 下
                    "$ref": "#/components/schemas/ErrorMessage"
                },
                # 手動指定example
                "example": {"code": "404", "message": "Not Found"}

            },
            # 其他格式的響應數據 格式如上面一樣
            "multipart/form-data": {

            }
        }, }

}


@app.get("/data", responses=responses, response_model=UserData)
async def root():
    return {}

標記已過時api

我們可以通過FastAPI或APIRouter或app.include_router或app.get...的deprecated參數標記當前路由是否已經過時
在文檔中, 過時的效果如下圖:

程式碼:

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/", tags=["items"])
async def read_items():
    return [{"name": "Foo", "price": 42}]


@app.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "johndoe"}]


@app.get("/elements/", tags=["items"], deprecated=True)
async def read_elements():
    return [{"item_id": "Foo"}]

同樣, 你也可以將一個參數標記為已過時的:

from fastapi import FastAPI
from fastapi import Query

app = FastAPI()


@app.get("/data/")
async def read_data(username: str = Query(..., description="用戶名"),
                    uid: int = Query(..., description="用戶ID", deprecated=True)):
    return {"username": username}

從文檔中排除api

我們可以通過FastAPI或APIRouter或app.include_router或app.get...的include_in_schema參數將當前路由排除出文檔
這對於一些只在測試中的介面十分有用, 需要注意的是: 你仍然可以訪問到該介面, 只是在文檔中不顯示而已

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/", tags=["items"])
async def read_items():
    return [{"name": "Foo", "price": 42}]


@app.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "johndoe"}]


# include_in_schema為False時
# 將 /elements/ 排除出文檔
@app.get("/elements/", tags=["items"], include_in_schema=False)
async def read_elements():
    return [{"item_id": "Foo"}]

依賴注入在文檔中

依賴注入, 也會加入到文檔中

比如:

from fastapi import FastAPI
from fastapi import Depends, Query
from pydantic import BaseModel

app = FastAPI()


class DataItem(BaseModel):
    id: int
    username: str


def get_data(data_id: int = Query(..., description="數據的ID", example=1)):
    return {"id": data_id, "username": "lczmx"}


@app.get("/items")
async def read_elements(data: DataItem = Depends(get_data)):
    return data

後台任務

例子:

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    # 後台任務的函數為正常的函數
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    # 添加到後台任務
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app="test:app", host="127.0.0.1", port=8000, reload=True)

你還可以在依賴注入中, 執行後台任務

from fastapi import BackgroundTasks, FastAPI, Depends
from typing import Optional

app = FastAPI()


def write_notification(email: str, message=""):
    # 後台任務的函數為正常的函數
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


def dependency_email(background_tasks: BackgroundTasks, email: Optional[str] = None):
    if email:
        # 添加到後台任務
        background_tasks.add_task(write_notification, email, message="some notification")
    return email


@app.post("/send-notification/")
async def send_notification(email: str = Depends(dependency_email)):
    return {"message": "Notification sent in the background"}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app="test:app", port=8000, reload=True)

Params

當我們導入Path等類時:即from fastapi import Path, 返回特殊類的函數 (__init__.py文件導入了) , 本質上是fastapi.params下的類

Param

Params類是Pydantic.FieldInfo類的子類, Path/Query/Header/Cookie都繼承Params類, 故而有共同的方法和屬性, 所以寫在一起.

注: Pydantic.Field 也會返回一個FieldInfo的實例。
Path等類也直接返回FieldInfo的一個子類的對象。還有其他一些你之後會看到的類是 Body 類的子類。

由於Param調用的是pydantic的構造函數, 所以實例化的參數類似, 所有參數見官網: Field customization

Body

Body類可用於接收單個請求體參數, 由於請求體編碼可以為application/json/multipart/form-data/application/json。故而分為Form和File和Body三個類.

• Body的media_type: application/json
• Form的media_type: application/x-www-form-urlencoded
• File的media_type: multipart/form-data

Body特有的參數:embed, 見: 嵌入單個請求體參數

其他參數和Param相同

WebSocket

WebSocket概述

注意: 這部分內容轉載於: WebSocket 詳解教程

WebSocket 是什麼?

WebSocket是一種網路通訊協議。RFC6455 定義了它的通訊標準。

WebSocket 是 HTML5 開始提供的一種在單個 TCP 連接上進行全雙工通訊的協議。

為什麼需要 WebSocket?

了解電腦網路協議的人，應該都知道：HTTP 協議是一種無狀態的、無連接的、單向的應用層協議。它採用了請求/響應模型。通訊請求只能由客戶端發起，服務端對請求做出應答處理。
這種通訊模型有一個弊端：HTTP 協議無法實現伺服器主動向客戶端發起消息。
這種單向請求的特點，註定了如果伺服器有連續的狀態變化，客戶端要獲知就非常麻煩。大多數 Web 應用程式將通過頻繁的非同步 JavaScript 和 XML（AJAX）請求實現長輪詢。輪詢的效率低，非常浪費資源（因為必須不停連接，或者 HTTP 連接始終打開）。

因此，工程師們一直在思考，有沒有更好的方法。WebSocket 就是這樣發明的。WebSocket 連接允許客戶端和伺服器之間進行全雙工通訊，以便任一方都可以通過建立的連接將數據推送到另一端。WebSocket 只需要建立一次連接，就可以一直保持連接狀態。這相比於輪詢方式的不停建立連接顯然效率要大大提高。

WebSocket 如何工作

Web 瀏覽器和伺服器都必須實現 WebSockets 協議來建立和維護連接。由於 WebSockets 連接長期存在，與典型的 HTTP 連接不同，對伺服器有重要的影響。
基於多執行緒或多進程的伺服器無法適用於 WebSockets，因為它旨在打開連接，儘可能快地處理請求，然後關閉連接。任何實際的 WebSockets 伺服器端實現都需要一個非同步伺服器。

WebSocket 客戶端

在客戶端，沒有必要為 WebSockets 使用 JavaScript 庫。實現 WebSockets 的 Web 瀏覽器將通過 WebSockets 對象公開所有必需的客戶端功能（主要指支援 Html5 的瀏覽器）。

以下程式碼可以創建一個WebSocket 對象:

var Socket = new WebSocket(url, [protocol] );

• 第一個參數url, 指定連接的URL
• 第二個參數protocol 是可選的，指定了可接受的子協議

WebSocket 屬性
以下是 WebSocket 對象的屬性。假定我們使用了以上程式碼創建了 Socket 對象：

WebSocket 事件
以下是 WebSocket 對象的相關事件。假定我們使用了以上程式碼創建了 Socket 對象：

WebSocket 方法
以下是 WebSocket 對象的相關方法。假定我們使用了以上程式碼創建了 Socket 對象：

例子:

// 初始化一個 WebSocket 對象
var ws = new WebSocket('ws://localhost:9998/echo');

// 建立 web socket 連接成功觸發事件
ws.onopen = function() {
  // 使用 send() 方法發送數據
  ws.send('發送數據');
  alert('數據發送中...');
};

// 接收服務端數據時觸發事件
ws.onmessage = function(evt) {
  var received_msg = evt.data;
  alert('數據已接收...');
};

// 斷開 web socket 連接成功觸發事件
ws.onclose = function() {
  alert('連接已關閉...');
};

FastAPI中使用WebSocket

在FastAPI中使用fastapi.WebSocket (內部使用的是starlette.websockets.WebSocket) 創建一個WebSocket伺服器
簡單例子:

from fastapi import FastAPI, WebSocket

app = FastAPI()


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        # 接收
        data = await websocket.receive_text()
        # 發送
        await websocket.send_text(f"接收到文本: {data}")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app="main:app", host="0.0.0.0", port=8000, reload=True)

接收數據

我們可以使用一些任意方法接收數據:

發送數據

我們可以使用一些任意方法發送數據:

其他方法和屬性

一些常用方法

綜合例子

比如實現一個聊天室

from typing import List

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()


class ConnectionManager:
    """
    用於管理多個ws連接
    """

    def __init__(self):
        # 存放所有ws連接, 主要由於廣播
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        """
        建立連接
        調用accept並添加到active_connections
        """
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        """
        從active_connections移除當前連接
        """
        self.active_connections.remove(websocket)

    async def send_personal_message(self, message: str, websocket: WebSocket):
        """
        為當前ws 發送數據
        """""
        await websocket.send_text(message)

    async def broadcast(self, message: str):
        """
        為所有ws 發送數據
        """""
        for connection in self.active_connections:
            await connection.send_text(message)


manager = ConnectionManager()


# 你同樣可以使用 Path Cookie Header Query Depends Security
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.send_personal_message(f"你發送了: {data}", websocket)
            await manager.broadcast(f"連接 #{client_id} 發送了: {data}")

    # 有用戶斷開連接時觸發
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client #{client_id} left the chat")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app="main:app", host="0.0.0.0", port=8000, reload=True)

運行上面的程式碼, 並在下面建立兩個連接查看聊天室功能

<!--@html-start-->
<!DOCTYPE html>
<html lang="en">
<head>
    
    <title>Title</title>
    <link href="//cdn.jsdelivr.net/npm/[email protected]/dist/css/bootstrap.min.css" rel="stylesheet">
    <link href="//blog-static.cnblogs.com/files/lczmx/websocket_tool.min.css" rel="stylesheet">
    <style>
    </style>
</head>
<body>
<div class="well socketBody">
    <div class="socketTop">
        <div class="socketTopColLeft">
            <div class="btn-group socketSelect">
                <button type="button" class="btn btn-default dropdown-toggle socketSelectBtn" data-toggle="dropdown"
                        aria-expanded="false">
                    <span class="showHeadWS">WS</span>
                    <span class="caret"> </span>
                </button>
                <ul class="dropdown-menu socketSelectshadow">
                    <li><a onclick="showWS('WS')">WS</a></li>
                    <li><a onclick="showWS('WSS')">WSS</a></li>
                </ul>
            </div>
        </div>
        <div class="socketTopColRight">
            <input type="text" list="typelist" class="form-control urlInput"
                   placeholder="請輸入連接地址~  如: 127.0.0.1:8000/ws"
                   oninput="inputChange()">
            <datalist id="typelist" class="inputDatalist">
                <option>127.0.0.1:8000/ws/233333</option>
            </datalist>
        </div>
    </div>
    <div class="socketBG well" id="main"></div>
    <div class="socketBottom row">
        <div class="col-xs-8 socketTextareaBody">
            <textarea rows="5" cols="20" class="form-control socketTextarea" placeholder="請輸入發送資訊~"></textarea>
        </div>
        <div class="col-xs-2 socketBtnSendBody">
            <button type="button" class="btn btn-success socketBtnSend" onclick="sendBtn()">發送</button>
        </div>
        <div class="col-xs-2 socketBtnBody">
            <button type="button" class="btn btn-primary socketBtn" onclick="connectBtn()">連接</button>
            <button type="button" class="btn btn-info socketBtn" onclick="emptyBtn()">清屏</button>
            <button type="button" class="btn btn-warning socketBtn" onclick="closeBtn()">斷開</button>
        </div>
    </div>
    <div class="alert alert-danger socketInfoTips" role="alert">...</div>
 
 
</div>
<script src="//cdn.jsdelivr.net/npm/[email protected]/dist/jquery.min.js"></script>
<script src="//cdn.jsdelivr.net/npm/[email protected]/dist/js/bootstrap.min.js"></script>
<script src="//blog-static.cnblogs.com/files/lczmx/websocket_tool.min.js"></script>
 
</body>
</html>
<!--@html-end-->
 
<!--@css-start-->
/* 已經在link中引入並壓縮了 */
<!--@css-end-->
 
<!--@javascript-start-->
/* 已經在script中引入並壓縮了 */
<!--@javascript-end-->

<!--@html-start-->
<!DOCTYPE html>
<html lang="en">
<head>
    
    <title>Title</title>
    <link href="//cdn.jsdelivr.net/npm/[email protected]/dist/css/bootstrap.min.css" rel="stylesheet">
    <link href="//blog-static.cnblogs.com/files/lczmx/websocket_tool.min.css" rel="stylesheet">
    <style>
    </style>
</head>
<body>
<div class="well socketBody">
    <div class="socketTop">
        <div class="socketTopColLeft">
            <div class="btn-group socketSelect">
                <button type="button" class="btn btn-default dropdown-toggle socketSelectBtn" data-toggle="dropdown"
                        aria-expanded="false">
                    <span class="showHeadWS">WS</span>
                    <span class="caret"> </span>
                </button>
                <ul class="dropdown-menu socketSelectshadow">
                    <li><a onclick="showWS('WS')">WS</a></li>
                    <li><a onclick="showWS('WSS')">WSS</a></li>
                </ul>
            </div>
        </div>
        <div class="socketTopColRight">
            <input type="text" list="typelist" class="form-control urlInput"
                   placeholder="請輸入連接地址~  如: 127.0.0.1:8000/ws"
                   oninput="inputChange()">
            <datalist id="typelist" class="inputDatalist">
                <option>127.0.0.1:8000/ws/666666</option>
            </datalist>
        </div>
    </div>
    <div class="socketBG well" id="main"></div>
    <div class="socketBottom row">
        <div class="col-xs-8 socketTextareaBody">
            <textarea rows="5" cols="20" class="form-control socketTextarea" placeholder="請輸入發送資訊~"></textarea>
        </div>
        <div class="col-xs-2 socketBtnSendBody">
            <button type="button" class="btn btn-success socketBtnSend" onclick="sendBtn()">發送</button>
        </div>
        <div class="col-xs-2 socketBtnBody">
            <button type="button" class="btn btn-primary socketBtn" onclick="connectBtn()">連接</button>
            <button type="button" class="btn btn-info socketBtn" onclick="emptyBtn()">清屏</button>
            <button type="button" class="btn btn-warning socketBtn" onclick="closeBtn()">斷開</button>
        </div>
    </div>
    <div class="alert alert-danger socketInfoTips" role="alert">...</div>
 
 
</div>
<script src="//cdn.jsdelivr.net/npm/[email protected]/dist/jquery.min.js"></script>
<script src="//cdn.jsdelivr.net/npm/[email protected]/dist/js/bootstrap.min.js"></script>
<script src="//blog-static.cnblogs.com/files/lczmx/websocket_tool.min.js"></script>
 
</body>
</html>
<!--@html-end-->
 
<!--@css-start-->
/* 已經在link中引入並壓縮了 */
<!--@css-end-->
 
<!--@javascript-start-->
/* 已經在script中引入並壓縮了 */
<!--@javascript-end-->

中間件

一般中間件

帶yield的依賴的退出部分的程式碼 (finally) 和 後台任務 會在中間件之後運行

from fastapi import FastAPI, Request
import time

app = FastAPI()


@app.middleware('http')
async def add_process_time_header(request: Request, call_next):
    # 處理request
    # ...
    start_time = time.time()
    # call_next 需要await
    # 接收request請求做為參數, 返回response
    response = await call_next(request)
    # 處理response
    # ...
    process_time = time.time() - start_time
    # 添加自定義的以「X-」開頭的請求頭
    response.headers['X-Process-Time'] = str(process_time)
    return response


@app.get("/")
async def index():
    return {"index": "/"}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("fastapi-test:app", port=8000, reload=True)

返回數據的響應頭:

content-length: 13 
content-type: application/json 
date: Wed,29 Dec 2021 14:25:48 GMT 
server: uvicorn 
x-process-time: 0.0010099411010742188 

CORSMiddleware解決跨域問題

用於同源策略, 我們需要特意指定那些源可以跨域請求

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    # 允許跨域請求的源列表
    allow_origins=[
        "//127.0.0.1",
        "//127.0.0.1:8080"
    ],
    # 指示跨域請求支援 cookies。默認是 False
    # 為True時, allow_origins 不能設定為 ['*']，必須指定源。
    allow_credentials=True,
    # 允許跨域請求的 HTTP 方法列表
    allow_methods=["*"],
    # 允許跨域請求的 HTTP 請求頭列表
    allow_headers=["*"],
)


@app.get("/")
async def index():
    return {"index": "/"}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("fastapi-test:app", port=8000, reload=True)

pycharmHttpClient

pycharm HTTP Client是pycharm自帶的工具

使用語法見官網: Exploring the HTTP request in Editor syntax