FastAPI 入門

FastAPI 是一個用於構建 API 的現代、快速（高性能）的 web 框架，使用 Python 3.6+ 並基於標準的 Python 類型提示。

# 安裝與啟動#

首先安裝 FastAPI

pip install "fastapi[all]"

以上安裝還包括了 uvicorn ，可以將其用作運行代碼的伺服器。

執行代碼時，使用一下命令

uvicorn main:app --reload

伺服器上後台啟動

nohup uvicorn main:app --host 0.0.0.0 --port 8000 > /文件路徑/log.log 2>&1 &

接著執行

bg

可以將一個在後台暫停的作業切換到繼續在後台執行

最後執行

disown

用於將一個作業從當前終端會話中分離，使其不會受到終端會話的影響，退出終端仍然運行

uvicorn main:app 命令含義如下:

main：main.py 文件（一個 Python「模塊」）。
app：在 main.py 文件中通過 app = FastAPI() 創建的對象，也可以是其他名稱
--reload：讓伺服器在更新代碼後重新啟動。僅在開發時使用該選項。

此時訪問 localhost:8000 即可調用

# Hello World#

## 新建一個 main.py 文件#

from fastapi import FastAPI

app = FastAPI()

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

代碼解釋

from fastapi import FastAPI

FastAPI 是一個為你的 API 提供了所有功能的 Python 類。

app = FastAPI()

這裡的變量 app 會是 FastAPI 類的一個「實例」。

這個實例將是創建所有 API 的主要交互對象。

這個 app 同樣在啟動的命令中被 uvicorn 所引用

也可以寫成 myapi = FastAPI ()
啟動命令就變成 uvicorn main --reload

@app.get("/")

告訴 FastAPI 在它下方的函數負責處理如下訪問請求

請求的路徑為 /

使用 get 操作調用

同樣可以使用 @app.post () 、 @app.put () 、 @app.delete ()

async def root():

定義一個 Python 函數，函數名為 root

async 表示這是一個異步函數

return {"message": "Hello World"}

返回內容

## 執行文件#

uvicorn main:app --reload

執行成功

# 路徑參數#

## 路徑中傳入參數#

@app.get("/items/{item_id}")
async def read_item(item_id):
    return {"item_id": item_id}

{item_id} 表示這是一個可以傳入的參數

此時請求 localhost:8000/items/1

會返回

{ "item_id": "1"}

## 限制參數類型#

同時，可以給這個參數限制類型

@app.get("/items/{item_id}")
async def read_item(item_id:int):
    return {"item_id": item_id}

只需要在函數入參的後面加上 : 類型

此時如果傳入一個字符串，將會返回 422 錯誤

類型可以使用 str、float、bool 等等

注意！

需要注意順序問題

比如：

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": user_id}

現在有兩個接口，如果 /users/{user_id} 在 /users/me 前面

由於_路徑操作_是按順序依次運行的

那當請求 /users/me 的時候，會執行到 /users/{user_id} 接口，返回 “me” 字符串

所以需要確保 /users/me 在 /users/{user_id} 的前面

## 枚舉類型預設值#

### 創建枚舉類#

from fastapi import FastAPI
# 引入枚舉類型
from enum import Enum

# 創建枚舉類
class ModelNameEnum(str, Enum):
      fire = "fire"
      peach = "peach"
      firepeach = "firepeach"

app = FastAPI()


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

@app.get("/items/{item_id}") 
async def read_item(item_id:str): 
	return {"item_id": item_id}

@app.get("/models/{model_name}")
async def get_model(model_name: ModelNameEnum):
    # 通過枚舉類的值來判斷
    if model_name is ModelNameEnum.fire:
        return {"model_name": model_name, "message": "Fire Fire Fire"}
    if model_name.value == "peach":
        return {"model_name": model_name, "message": "Peach Peach Peach"}
    return {"model_name": model_name, "message": "FirePeach FirePeach FirePeach"}

代碼解釋：

async def get_model(model_name: ModelNameEnum):

定義的入參 model_name 是 ModelNameEnum 枚舉類型

if model_name is ModelNameEnum.fire:

這裡的 is 和 == 的作用是相同的，用來和 枚舉成員 進行比較

但是使用 == 的時候，需要用入參 .value 來獲取值

model_name.value

### 測試請求#

## 文件路徑作為入參#

如果需要傳入的入參是一個路徑

可以使用

/files/{file_path:path}

作為請求路徑

@app.get("/files/{file_path:path}") 
async def read_file(file_path: str): 
	return {"file_path": file_path}

當需要傳入 /home/test/text.txt 路徑的時候

請求的入參為 localhost:8000/files//home/test/text.txt 注意為兩個 /

# 查詢參數#

## 預設值#

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

比如下面的代碼

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]

在函數入參中

async def read_item(skip: int = 0, limit: int = 10):

定義了預設值 skip 默認為 0，limit 默認為 10

所以

訪問 http://127.0.0.1:8000/items/?skip=0&limit=10 的時候

傳入的參數就是 skip = 0 , limit = 10

訪問 http://127.0.0.1:8000/items/?skip=1 的時候

傳入的參數就是 skip = 1 , limit = 10，因為 limit 不傳時，默認為 10

訪問 http://127.0.0.1:8000/items/?limit=1 的時候

傳入的參數就是 skip = 0 , limit = 1，skip 不傳默認為 0

## 可選值#

需要引入 Union

from typing import Union

@app.get("/items/{item_id}")
async def read_itemUnion(item_id: str, q: Union[str, None] = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

此時，q 就是可傳可不傳的參數

還可以加入 bool 類型的入參

@app.get("/items/{item_id}")
async def read_itemUnion(item_id: str, q: Union[str, None] = None, short: bool = False):
    item = {"item_id": item_id}
    if q:
        # 將 q 作為 item 的一個鍵值對，update 可以將新的鍵值對合併進去
        item.update({"q": q})
    if not short:
         item.update({"description": "This is an amazing item that has a long description"})
    return item

使用 update ({"XX" : "XX"})，可以在原有的鍵值對中新增這個鍵值對拼接上去

測試訪問

## 多個路徑和查詢參數#

@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
    user_id: int, item_id: str, q: Union[str, None] = 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

可以支持多個參數的請求

## 必須的查詢參數#

如果在非路徑的參數中指定了一個沒有預設值的參數，那麼這個參數就是必傳的

也就是說不是在 @app.get (/XXX/{ XXX}) 中指定的，是在函數的入參中指定的

而且沒有配置 Union[str, None] = None , 那麼這個參數就是必傳的

@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 參數就是必傳

# 請求體#

當你需要將數據從客戶端（例如瀏覽器）發送給 API 時，你將其作為「請求體」發送。

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

你的 API 幾乎總是要發送響應體。但是客戶端並不總是需要發送請求體。

## 導入 BaseModel#

from pydantic import BaseModel

需要用到 pydantic 模型來聲明請求體

## 創建數據模型#

Class Item(BaseModel):
	name: str
	description: Union[str,None] = None
	price: float
	tax: Union[float,None] = None

這時候就聲明了一個名稱為 Item 的請求體的結構類型

{
    "name": "Foo",
    "description": "An optional description",
    "price": 45.2,
    "tax": 3.5
}

接著添加一個 post 方法，就可以指定入參為 Item 模型類型

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

測試請求

## 使用模型#

在函數的內部，可以直接訪問模型對象的所有屬性

但是需要通過 模型.dict () 獲取模型的字典才可以進行修改

模型.dict () 就是模型中的每一個屬性單獨拎出來

@app.post("/items/") 
async def create_item(item: Item): 
	item_dict = item.dict() 
	if item.tax: 
		price_with_tax = item.price + item.tax 
		item_dict.update({"price_with_tax": price_with_tax}) 
	return item_dict

## 請求體加路徑參數#

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

可以同時傳入路徑參數和請求體

注意返回 item.dict () 的時候要加上 **

可以看出如果返回 item.dict () 就是把模型中的每一個屬性，單獨的拎出來

## 請求體、路徑參數加查詢參數#

@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item, q: Union[str, None] = None):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result

# 參數校驗#

查詢參數校驗#

首先需要導入 Query

from fastapi import FastAPI, Query

### 長度校驗#

@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, min_length=2, max_length=5)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

可以在參數後加上 Query ()

由於必須用 Query(default=None) 替換預設值 None，Query 的第一個參數同樣也是用於定義預設值。

上面的代碼表示 q 這個字段最小長度為 2 最大長度為 5

### 正則表達式校驗#

@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, min_length=2, max_length=5,regex='^query$')):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

regex='^query$'

上面的代碼指定了正則表達式， q 的內容必須為 query

### 預設值#

同時預設值也可以是指定的

default = 'query'

## 路徑參數的校驗#

需要引入 Path

from fastapi import FastAPI,Query,Path

添加方法

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", ge=0, le=1000),
    q: str,
    size: float = Query(gt=0, lt=10.5),
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

@app.get("/items4/{item_id}")
async def read_items6(

    *,
    item_id: int = Path(title="The ID of the item to get", ge=0, le=1000),
    q: str,
    size: float = Query(gt=0, lt=10.5),
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

gt：大於（greater than）
ge：大於等於（greater than or equal）
lt：小於（less than）
le：小於等於（less than or equal）

傳遞 * 作為函數的第一個參數。

Python 不會對該 * 做任何事情，但是它將知道之後的所有參數都應作為關鍵字參數（鍵值對），也被稱為 kwargs，來調用。即使它們沒有預設值。

也就是說，如果不加 * 作为第一个参数的话，函數入參里不能只填寫 q: str
因為他沒有預設值

# 請求體中多個 Model 參數#

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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


class User(BaseModel):
    username: str
    full_name: Union[str, None] = 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"
    }
}

## 請求體中的單一值#

如果想在同一個請求體裡面在新增一個鍵值對

比如

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

那麼可以引入 Body

from fastapi import FastAPI,Query,Path,Body

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User, key: str = Body()):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    return results

通過添加下面代碼來實現

key: str = Body()

同樣的 Body () 中也可以和 Query、Path 一樣添加校驗

key: str = Body(gt=1)

測試接口

## 嵌入單個請求體參數#

如果期望實現的是下面這種請求方式

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}

而不是

{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2
}

則可以使用 Body () 的 embed 參數

比如

item: Item = Body(embed=True)

from typing import Union

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = 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

# 請求體中的嵌套模型#

## 具有子類型的 List 字段#

首先引入 List

from typing import List, Union

比如需要傳入多個標籤

可以使用下面代碼來表示 List

# 限制內容的類型，只能傳入 str 類型
tags: List[str] = []

# 不限制內容的類型，str、int等等都可以
tags:List =[]

代碼如下

from typing import List, Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results

## Set 類型#

和 List 不同的是，Set 中的元素不可以重複，所以用於存放標籤會更合適

因為重複的元素會被自動合併

一樣需要從 typing 中引入 Set

from typing import Set

使用代碼

tags: Set[str] = set()

代碼如下

class Message(BaseModel):
    role:str
    content:str
    tags:Set[str] = set()
  
@app.post("/chat")
async def chat(message: Message, model: str = Body()):
    return {"message":message,"model":model}

## 嵌套模型#

可以在模型中，嵌套另一個模型

例如

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

先定義子模型

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

接著添加到原來的模型中

class Message(BaseModel):
    role:str
    content:str
    tags:Set[str] = set()
    
    image: Union[Image,None] = None

## 特殊的類型和校驗#

除了普通的單一值類型（如 str、int、float 等）外，你還可以使用從 str 繼承的更複雜的單一值類型。

外部的特殊類型，可以通過 Pydantic 引入，具體可以訪問 Pydantic 官網

比如 image 中的 URL 應該是個地址，相應的 Pydantic 提供了 HttpUrl 的類型用來校驗

同樣的先引入

from pydantic import BaseModel, HttpUrl

接著修改 image 模型類

class Image(BaseModel):

    url: HttpUrl
    
    name: str

此時如果 URL 傳入的不是 http 或者 https 開頭的地址，都会報錯

## 帶有一組子模型的屬性#

同樣的，還可以將 Pydantic 模型用作 list、set 等的子類型

比如

class Message(BaseModel):
    role:str
    content:str
    tags:Set[str] = set()

    image: Union[List[Image],None] = None

# 響應模型#

## 用戶註冊響應#

可以在任意的_路徑操作_中使用 response_model 參數來聲明用於響應的模型：

比如用戶註冊，註冊成功後返回用戶數據

但是密碼是不能直接返回的，所以可以設置一個沒有密碼的響應模型

用戶返回定的模型的出參

首先引入 Any 類型和 EmailStr

from typing import Any

from pydantic import EmailStr

定義註冊用戶的模型

class User(BaseModel):
	username:str
	full_name: Union[str,None] = None
	email: EmailStr
	password: str

定義註冊成功後返回的響應模型

class UserOut(BaseModel):
	username: str
	full_name: Union[str,None] = None
	email: EmailStr

實現代碼如下

@app.post("/user/", response_model= UserOut):
async def create_user(user: User) -> Any:
	return user

這裡的 Any 是指返回的 user 可以是任何類型

測試接口

## 預設模型響應#

首先定義一個相應模型

class Item2(BaseModel):
    name: str
    description: Union[str,None] = None
    price: float
    tax: float = 10
    tags: List[str] = []

接著設置默認的返回

item2s = {
    "fire": {"name": "Fire", "price": 3.2},
    "peach": {"name": "Peach", "description": "Peach Peach Peach", "price": 2.2,"tags":["aa","bb","cc"]},
    "firepeach": {"name": "firepeach", "description": "firepeach firepeach firepeach", "price": 2.2}
}

接口實現如下

@app.get("/item2s/{item_id}", response_model=Item2)
async def read_item2(item_id: str):
	return item2s[item_id]

此時請求入參發送 item_id 就會返回 item2s 中對應的 item_id 的那條數據

測試調用

如果不想返回那些帶有默認值的參數

可以添加一個參數 response_model_exclude_unset=True

修改代碼如下

@app.get("/item2s/{item_id}", response_model=Item2 , response_model_exclude_unset=True)
async def read_item2(item_id: str):
    return item2s[item_id]

此時接口調用，將不會返回 數據內容為默認值，或則內容為空 的數據

與默認值不同的仍然會返回

# 額外的模型#

保存用戶數據的時候，需要新增一個存入數據庫的模型

因為密碼不能通過明文去保存，例如可以用 hash 值去保存

新增 UserDB 模型

class UserDB(BaseModel):
    username: str
    hashed_password: str
    email: EmailStr
    full_name: Union[str,None] = None

定義一個假的密碼加密函數

def fake_password_hasher(raw_password: str):
    return "hashpassword" + raw_password

定義一個拼接 UserDB 模型類型的數據并保存到數據庫的函數

def fake_save_user(user_in: User):
    fake_hash_password = fake_password_hasher(user_in.password);
    # 可以使用 ** 將一個字典解包成關鍵字參數
    # 然後將解包好的 user_in 作為關鍵字參數傳入 user_db
    # 並將密碼加密成 hashed_password 一起賦予 user_db
    user_db = UserDB(**user_in.dict(), hashed_password=fake_hash_password);

    print("用戶假的保存成功！");
    return user_db```

新增一個註冊用戶的接口

``` python
@app.post("/user2/", response_model=UserOut)
async def create_user2(user: User):
	user_save = fake_save_user(user)
	return user_save

測試接口

## 關於 `**user.dict()` #

### Pydantic 模型#

user 是個 User 類的 Pydantic 模型.

Pydantic 模型具有 .dict（） 方法，該方法返回一個擁有模型數據的 dict （字典）。

所以如果創建一個 User 模型的對象是

user_in = UserIn(username="john", password="secret", email="[email protected]")

如果 user.dict ()

user_dict = user.dict()

user_dict 的數據則是

{
    'username': 'john',
    'password': 'secret',
    'email': '[email protected]',
    'full_name': None,
}

也就是所有的屬性都被單獨出來

### 解包 dict#

如果將 user_dict 這樣的 dict 以 **user_dict 形式傳遞給一個函數（或類），Python 將對其進行「解包」。它會將 user_dict 的鍵和值作為關鍵字參數直接傳遞。

所以如果使用以下代碼

UserDB(**user_dict)
# 或者
UserDB(**user.dict())

此時我們得到的 UserDB 則是

UserDB(
	'username': 'john',
    'password': 'secret',
    'email': '[email protected]',
    'full_name': None,
)

或者更確切地，直接使用 user_dict 來表示將來可能包含的任何內容：

UserDB(
    username = user_dict["username"],
    password = user_dict["password"],
    email = user_dict["email"],
    full_name = user_dict["full_name"],
)

然後添加額外的關鍵字參數 hashed_password=hashed_password，例如：

UserDB(**user.dict(),fake_hashed_password = fake_hashed_password)

最終的結果就會是

UserDB(
    username = user_dict["username"],
    password = user_dict["password"],
    email = user_dict["email"],
    full_name = user_dict["full_name"],
    hashed_password = hashed_password,
)

## 關於代碼重複#

代碼重複會增加出現 bug、安全性問題、代碼失步問題（當你在一個位置更新了代碼但沒有在其他位置更新）等的可能性。

上面的這些模型都共享了大量數據，並擁有重複的屬性名稱和類型。

因此，基於上面提到的 dict 和解包，可以直接定義一個 UserBase 的基類

class UserBase(BaseModel):
	username: str
	email: EmailStr
	full_name: Union[str,None] = None

而其他的模型只需要在此基礎上，只賦予新增的屬性

例如

class User(BaseModel):
    password: str


class UserDB(BaseModel):
    hashed_password: str


class UserOut(BaseModel):
    # pass 代表不做任何處理，保持和 UserBase 一致
    pass

因為後面接口的使用中基本所有的字段都會有，只需要解決新增的字段

## Union 模型#

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

這將在文檔中表示為 anyOf

例如

 response_model=Union[PlaneItem, CarItem]

代碼如下

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]

## 模型列表#

可以用同樣的方式聲明由對象列表構成的響應。

response_model=List[Item]

代碼如下

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

# 響應狀態碼#

可以通過 status_code 來指定狀態碼

例如

@app.post("/user2/", response_model=UserOut,status_code=201)
async def create_user2(user: User):
    user_saved = fake_save_user(user);
    return user_saved;

成功返回時，將返回 201 的狀態碼

同時 fastapi 也有提供便捷的變量

引入

from fastapi import status

@app.post("/user2/", response_model=UserOut,status_code=status.HTTP_201_CREATED)
async def create_user2(user: User):
    user_saved = fake_save_user(user);
    return user_saved;

常用的狀態碼有

有關每個狀態代碼以及適用場景的更多信息可以參考 HTTP 響應狀態碼 - HTTP | MDN

# 表單數據#

與 JSON 不同，HTML 表單（<form></form>）向伺服器發送數據通常使用「特殊」的編碼。

接收的不是 JSON，而是表單字段時，要使用 Form。

要使用表單，需預先安裝 python-multipart。

執行

pip install python-multipart

導入 fastapi 提供的 Form

from fastapi import FastAPI, Form
app = FastAPI()

創建表單（Form）參數的方式與 Body 和 Query 一樣

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

例如，OAuth2 規範的 "密碼流" 模式規定要通過表單字段發送 username 和 password。

該規範要求字段必須命名為 username 和 password，並通過表單字段發送，不能用 JSON。

使用 Form 可以聲明與 Body （及 Query、Path、Cookie）相同的元數據和驗證。

表單數據的「媒體類型」編碼一般為 `application/x-www-form-urlencoded`。

但包含文件的表單編碼為 `multipart/form-data`。

# 請求文件#

# 處理錯誤#

某些情況下，需要向客戶端返回錯誤提示。

這裡所謂的客戶端包括前端瀏覽器、其他應用程序、物聯網設備等。

需要向客戶端返回錯誤提示的場景主要如下：

客戶端沒有執行操作的權限
客戶端沒有訪問資源的權限
客戶端要訪問的項目不存在
等等 ...

遇到這些情況時，通常要返回 4XX（400 至 499）HTTP 狀態碼。

4XX 狀態碼與表示請求成功的 2XX（200 至 299） HTTP 狀態碼類似。

只不過，4XX 狀態碼表示客戶端發生的錯誤。

## HTTPException#

向客戶端返回 HTTP 錯誤響應，可以使用 HTTPException。

from fastapi import HTTPException

例如

@app.get("/item2s/{item_id}", response_model=Item2 , response_model_exclude_unset=True)
async def read_item2(item_id: str):
    if item_id not in item2s:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND , detail="沒這個文件")
    return item2s[item_id]

測試調用

## 安裝自定義異常處理器#

添加自定義處理器，要使用 Starlette 的異常工具。

假設要觸發的自定義異常叫作 UnicornException。

且需要 FastAPI 實現全局處理該異常。

此時，可以用 @app.exception_handler() 添加自定義異常控制器：

也可以直接引入 fastapi 封裝好的

from fastapi import  Request
from fastapi.response import JSONResponse

代碼如下

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
	    content={"message": f"哈哈! {exc.name} 這裡出錯了..."},
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name="自定義錯誤名")
    return {"unicorn_name": name}

{"message": f"哈哈! {exc.name} 這裡出錯了..."} 中的 f
可以獲取自定義錯誤名

測試結果

# JSON 兼容編碼器與更新數據#

在某些情況下，您可能需要將數據類型（如 Pydantic 模型）轉換為與 JSON 兼容的數據類型（如dict、list等）。

比如，如果您需要將其存儲在數據庫中。

對於這種要求， FastAPI提供了jsonable_encoder()函數。

## 使用 jsonable_encoder () 更新數據#

假設有一個數據庫名為fake_db，它只能接收與 JSON 兼容的數據。

例如，它不接收datetime這類的對象，因為這些對象與 JSON 不兼容。

因此，datetime對象必須將轉換為包含ISO 格式化的str類型對象。

同樣，這個數據庫也不會接收 Pydantic 模型（帶有屬性的對象），而只接收dict。

對此可以使用 jsonable_encoder。

更新數據請用 HTTP PUT 操作。

它接收一個對象，比如 Pydantic 模型，並會返回一個 JSON 兼容的版本：

把輸入數據轉換為以 JSON 格式存儲的數據（比如，使用 NoSQL 數據庫時），可以使用 jsonable_encoder。例如，把 datetime 轉換為 str。

from datetime import datetime

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

# 假的數據庫
fake_db = {
    "fee": {"title": "foo", "timestamp": datetime.now()}
}

# 定義一個ItemJ 模型
class ItemJ(BaseModel):
    title: str
    timestamp: datetime
    description: str | None = None

# put 用來更新
@app.put("/itemsJ/{id}")
def update_itemJ(id: str, item: ItemJ):
	# 將時間傳喚成 str 類型的 Json格式
    json_compatible_item_data = jsonable_encoder(item)
    fake_db[id] = json_compatible_item_data
    return fake_db[id]

測試調用

# 安全性#

假設後端 API 在某個域。

前端在另一個域，或（移動應用中）在同一個域的不同路徑下。

並且，前端要使用後端的 username 與 password 驗證用戶身份。

固然，FastAPI 支持 OAuth2 身份驗證。

## 概覽#

引入 Depends 依賴和 OAuth2PasswordBearer 身份認證

from fastapi import Depends, FastAPI 
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/oauth/")
async def read_items(token: str = Depends(oauth2_scheme)):
    return {"token": token}

此時在 api 文檔 localhost:8000/docs 中可以看到

頁面右上角出現了一個「Authorize」按鈕。

_路徑操作_的右上角也出現了一個可以點擊的小鎖圖標。

點擊 Authorize 按鈕，彈出授權表單，輸入 username 與 password 及其它可選字段：

如果沒有，需要安裝 python-multipart

執行 pip install python-multipart

是因為 OAuth2 使用 表單數據 发送 `username` 與 `password`。

### 密碼流#

Password 流是 OAuth2 定義的，用於處理安全與身份驗證的方式（流）。

OAuth2 的設計目標是為了讓後端或 API 獨立於伺服器驗證用戶身份。

但在本例中，FastAPI 應用會處理 API 與身份驗證。

下面是簡化的運行流程：

用戶在前端輸入 username 與password，並點擊回車
（用戶瀏覽器中運行的）前端把 username 與password 發送至 API 中指定的 URL（使用 tokenUrl="token" 聲明）
API 檢查 username 與 password，並用令牌（Token）响應（暫未實現此功能）：
令牌只是用於驗證用戶的字符串
一般來說，令牌會在一段時間後過期
- 過時後，用戶要再次登錄
- 這樣一來，就算令牌被人竊取，風險也較低。因為它與永久密鑰不同，在絕大多數情況下不會長期有效
前端臨時將令牌存儲在某個位置
用戶點擊前端，前往前端應用的其它部件
前端需要從 API 中提取更多數據：
- 為指定的端點（Endpoint）進行身份驗證
- 因此，用 API 驗證身份時，要發送值為 Bearer + 令牌的請求頭 Authorization
- 假如令牌為 foobar，Authorization 請求頭就是： Bearer foobar

### OAuth2PasswordBearer#

FastAPI 提供了不同抽象級別的安全工具。

本例使用 OAuth2 的 Password 流以及 Bearer 令牌（Token）

為此要使用 OAuth2PasswordBearer 類。

創建 OAuth2PasswordBearer 的類實例時，要傳遞 tokenUrl 參數。該參數包含客戶端（用戶瀏覽器中運行的前端）的 URL，用於發送 username 與 password，並獲取令牌。

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

在此，tokenUrl="token" 指向的是暫未創建的相對 URL token。這個相對 URL 相當於 ./token。

因為使用的是相對 URL，如果 API 位於 https://example.com/，則指向 https://example.com/token。但如果 API 位於 https://example.com/api/v1/，它指向的就是https://example.com/api/v1/token。

使用相對 URL 非常重要，可以確保應用在遇到使用代理這樣的高級用例時，也能正常運行

### 使用#

oauth2_scheme 變量是 OAuth2PasswordBearer 的實例，也是可調用項。

調用方式為

oauth2_scheme(some, parameters)

因此，Depends （依賴）可以調用 oauth2_scheme 變量。

接下來，使用 Depends 把 oauth2_scheme 傳入依賴項。

@app.get("/items/") 
async def read_items(token: str = Depends(oauth2_scheme)): 
	return {"token": token}

該依賴項使用字符串（str）接收_路徑操作函數_的參數 token

## 獲取當前的用戶#

### 創建用戶模型#

class UserAuth(BaseModel):
    username: str
    # 先使用 Union[str, None]，後面會改成 EmailStr
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None

### 創建依賴項#

創建一個 get_current_user 依賴項。

get_current_user 將具有一個之前所創建的同一個 oauth2_scheme 作為依賴項。

與之前直接在路徑操作中所做的相同，新的依賴項 get_current_user 將從子依賴項 oauth2_scheme 中接收一個 str 類型的 token：

async def get_current_user(token: str = Depends(oauth2_scheme)): 
	user = fake_decode_token(token) 
	return user

### 獲取用戶信息#

將從 oauth2_scheme 獲取到的 token 用來獲取用戶

get_current_user 將使用創建的（偽）工具函數 fake_decode_token ，該函數接收 str 類型的令牌並返回我們的 Pydantic User 模型

def fake_decode_token(token): 
	return User( username=token + "fakedecoded", email="[email protected]", full_name="John Doe" )

### 注入當前用戶#

現在可以在 路徑操作 中使用 get_current_user 作為 Depends 了：

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

## 使用密碼和 Bearer 的簡單 OAuth2#

### 概述#

現在將使用 FastAPI 的安全性實用工具來獲取 username 和 password。

OAuth2 規定在使用「password 流程」時，客戶端 / 用戶必須將 username 和 password 字段作為表單數據發送。

而且規範明確了字段必須這樣命名。因此 user-name 或 email 是行不通的。

不過數據庫模型也可以使用任何其他名稱。

但是對於登錄 路徑操作，需要使用這些名稱來與規範兼容（以具備例如使用集成的 API 文檔系統的能力）。

規範還寫明了 username 和 password 必須作為表單數據發送（因此，此處不能使用 JSON）。

### 獲取 `username` 和 `password`#

#

需要用到 OAuth2PasswordRequestForm

首先引入

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):

    user_dict = fake_users_db.get(form_data.username)
	# 如果 user_dict 不存在 
    if not user_dict:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

	# 將 User 的數據 傳給 UserDB 模型類
    user = UserDB(**user_dict)
    # 將密碼 hash 化
    hashed_password = fake_hash_password(form_data.password)

	# 如果 密碼不正確
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

	# 返回 用戶名 token的類型
    return {"access_token": user.username, "token_type": "bearer"}

OAuth2PasswordRequestForm 是一個類依賴項，聲明了如下的請求表單：

username。
password。
一個可選的 scope 字段，是一個由空格分隔的字符串組成的大字符串。
一個可選的 grant_type.

#### 密碼哈希化#

「哈希」的意思是：將某些內容（在本例中為密碼）轉換為看起來像雜湊的字節序列（只是一個字符串）。

每次你傳入完全相同的內容（完全相同的密碼）時，你都會得到完全相同的雜湊。

但是你不能從雜湊轉換回密碼。

    # 將密碼 hash 化
    hashed_password = fake_hash_password(form_data.password)

	# 如果 密碼不正確
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

#### 返回 Token#

token 端點的響應必須是一個 JSON 對象。

它應該有個 token_type。在例子中，使用的是「Bearer」令牌，因此令牌類型應為「bearer」。

並且還應該有一個 access_token 字段，它是一個包含我們的訪問令牌的字符串。

對於這個簡單的示例，我們將極其不安全地返回相同的 username 作為令牌。

	# 返回 用戶名 token的類型
    return {"access_token": user.username, "token_type": "bearer"}

#### 更新依賴項#

現在將更新依賴項。

我們想要僅當此用戶處於啟用狀態時才能獲取 current_user。

因此，創建了額外的依賴項 get_current_active_user，而該依賴項又以 get_current_user 作為依賴項。

如果用戶不存在或處於未啟用狀態，則這兩個依賴項都將僅返回 HTTP 錯誤。

因此，在我們的端點中，只有當用戶存在，身份認證通過且處於啟用狀態時，我們才能獲得該用戶：

# 通過 token 獲取用戶信息
async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = fake_decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

# 獲取當前活躍的用戶，登錄了之後就活躍了
# 身份認證通過且處於啟用狀態時，才能獲得此用戶
async def get_current_active_user(current_user: UserAuth = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

#### 創建虛假的數據庫#

# 模擬數據庫

fake_users_db = {
    "bryce": {
        "username": "bryce",
        "full_name": "bryce Yu",
        "email": "[email protected]",
        "hashed_password": "hash123123",
        "disabled": False,
    },

    "yu": {
        "username": "yu",
        "full_name": "yu",
        "email": "[email protected]",
        "hashed_password": "hash123456",
        "disabled": True,
    },
}

### 完整代碼（重點！！！！）#

完整的用戶登錄到獲取當前用戶的代碼如下

# 創建一個 用戶 模型類
class UserAuth(BaseModel):
    username: str
    # 先使用 Union[str, None]，後面會改成 EmailStr
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None

#  創建一個 用戶 數據庫 模型類
class UserDB(UserAuth):
    hashed_password: str

# 模擬將密碼哈希化
def fake_hash_password(raw_password: str):
    # 模擬哈希化 密碼，實際上應該使用 bcrypt 或者其他哈希算法
    return "hash" + raw_password

# 通过 token 獲取用戶信息
def fake_decode_token(token):
    # 通过 token 獲取用戶信息，現在不安全的使用 用戶名 作為 token
    user = get_user(fake_users_db, token)
    return user

# 從 db 數據庫中，獲取用戶名為 username 的用戶信息
def get_user(db, username: str):
    # 現在不安全的使用 用戶名 作為 token
    if username in db:
        user_dict = db[username]
        return UserDB(**user_dict)



# 先通过通过 oauth2_scheme 獲得token ，再使用 token 獲取當前登錄的用戶信息
# oauth2_scheme 是一個認證方案，它會從請求頭中獲取 Authorization 字段的值，然後將其解析成 token
async def get_current_user(token: str = Depends(oauth2_scheme)):
    # 通过token 獲取用戶信息
    user = fake_decode_token(token)
    # 如果用戶不存在
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="無效的認證憑證",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # 如果用戶存在，返回用戶信息
    return user


# 獲取當前活躍的用戶
async def get_current_active_user(current_user: UserAuth = Depends(get_current_user)):
    # 通过 get_current_user 獲取當前登錄的用戶 current_user
    # 接著判斷用戶是否被禁用
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="用戶被禁用")
    return current_user


# 通过表單 登錄用戶
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):

    # 通过用戶名 到 模擬數據庫中獲取用戶
    user_dict = fake_users_db.get(form_data.username)

	# 如果 user_dict 不存在 
    if not user_dict:
        raise HTTPException(status_code=400, detail="用戶名或密碼錯誤")

	# 將 User 的數據 傳給 UserDB 模型類，獲取 UserDB 的實例用來獲取密碼進行比較
    user = UserDB(**user_dict)

    # 將密碼 hash 化，用來和數據庫中的密碼進行比較
    hashed_password = fake_hash_password(form_data.password)

	# 如果 密碼不正確
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="用戶名或密碼錯誤")
    
	# 返回 token 和 token的類型，現在不安全的使用 用戶名 作為 token
    return {"access_token": user.username, "token_type": "bearer"}


# 調用這個接口的時候，需要在 headers 中添加 Authorization: Bearer token，token 就是登錄時返回的 token
# 當使用 openApi 測試的時候，會自動在 headers 中添加 Authorization: Bearer token
# 而實際的請求中，需要手動添加 Authorization: Bearer token
@app.get("/users/me")
async def read_users_me(current_user: UserAuth = Depends(get_current_active_user)):
    # 通过 get_current_active_user 獲取當前登錄並且沒有被禁用的用戶 current_user
    return current_user

#### 流程總結#

首先調用 /token 接口，通過表單提交 username 和 password ，接口會返回一個 token

接著需要在 headers 中的 Authorization 中添加 Token 的值，調用 /user/me 獲取當前登錄用戶

因為本案例中使用的是 Bearer ，所以 Authorization 的值為 "Bearer" + 返回的 Token

而在 openApi 調用測試的時候，為了方便測試，會自動加上 Authorization 的 Token 值

在實際項目的使用中，需要手動添加 headers 中的 Authorization 值

FastAPI 入門

# 安裝與啟動#

# Hello World#

## 新建一個 main.py 文件#

## 執行文件#

# 路徑參數#

## 路徑中傳入參數#

## 限制參數類型#

## 枚舉類型預設值#

### 創建枚舉類#

### 測試請求#

## 文件路徑作為入參#

# 查詢參數#

## 預設值#

## 可選值#

## 多個路徑和查詢參數#

## 必須的查詢參數#

# 請求體#

## 導入 BaseModel#

## 創建數據模型#

## 使用模型#

## 請求體加路徑參數#

## 請求體、路徑參數加查詢參數#

# 參數校驗#

查詢參數校驗#

### 長度校驗#

### 正則表達式校驗#

### 預設值#

## 路徑參數的校驗#

# 請求體中多個 Model 參數#

## 請求體中的單一值#

## 嵌入單個請求體參數#

# 請求體中的嵌套模型#

## 具有子類型的 List 字段#

## Set 類型#

## 嵌套模型#

## 特殊的類型和校驗#

## 帶有一組子模型的屬性#

# 響應模型#

## 用戶註冊響應#

## 預設模型響應#

# 額外的模型#

## 關於 **user.dict() #

### Pydantic 模型#

### 解包 dict#

## 關於代碼重複#

## Union 模型#

## 模型列表#

# 響應狀態碼#

# 表單數據#

# 請求文件#

# 處理錯誤#

## HTTPException#

## 安裝自定義異常處理器#

# JSON 兼容編碼器 與 更新數據#

## 使用 jsonable_encoder () 更新數據#

# 安全性#

## 概覽#

### 密碼流#

### OAuth2PasswordBearer#

### 使用#

## 獲取當前的用戶#

### 創建用戶模型#

### 創建依賴項#

### 獲取用戶信息#

### 注入當前用戶#

## 使用密碼和 Bearer 的簡單 OAuth2#

### 概述#

### 獲取 username 和 password#

#

#### 密碼 哈希 化#

#### 返回 Token#

#### 更新依賴項#

#### 創建虛假的數據庫#

### 完整代碼 （重點！！！！）#

#### 流程總結#

## 關於 `**user.dict()` #

# JSON 兼容編碼器與更新數據#

### 獲取 `username` 和 `password`#

#### 密碼哈希化#

### 完整代碼（重點！！！！）#