網(wǎng)站舉報平臺12315怎樣制作網(wǎng)頁設計
英文版文檔:https://fastapi.tiangolo.com/
中文版文檔:https://fastapi.tiangolo.com/zh/
1、FastAPI 教程
簡? 介
FastAPI 和 Sanic 類似,都是 Python 中的異步 web 框架。相比 Sanic,FastAPI 更加的成熟、社區(qū)也更加的活躍。
FastAPI 站在巨人的肩膀上?
很大程度上來說,這個巨人就是指 Flask 框架。
FastAPI 從語法上和 Flask 非常的相似,有異曲同工之妙。
其實不僅僅是 FastAPI ,就連 Sanic 也是基于 Flask 快速開發(fā)的 Web API 框架。
FastAPI?是一個用于構(gòu)建 API 的現(xiàn)代、快速(高性能)的 web 框架,使用 Python 3.6+ 并基于標準的 Python 類型提示。
它具有如下這些優(yōu)點:
- 快速:可與?NodeJS?和?Go?比肩的極高性能(歸功于 Starlette 和 Pydantic)。
Starlette 用于路由匹配,Pydantic 用于數(shù)據(jù)驗證
- 高效編碼:提高功能開發(fā)速度約 200% 至 300%
- 更少 bug:減少約 40% 的人為(開發(fā)者)導致錯誤。
- 智能:極佳的編輯器支持。處處皆可自動補全,減少調(diào)試時間
- 簡單:設計的易于使用和學習,閱讀文檔的時間更短
- 簡短:使代碼重復最小化。通過不同的參數(shù)聲明實現(xiàn)豐富功能。bug 更少
- 健壯:生產(chǎn)可用級別的代碼。還有自動生成的交互式文檔
- 標準化:基于(并完全兼容)API 的相關(guān)開放標準:OpenAPI?(以前被稱為 Swagger) 和?JSON Schema。
FastAPI 最大的特點就是它使用了 Python 的類型注解,使用 FastAPI 需要 Python 版本大于等于 3.6。
安裝?FastAPI
pip install fastapi,會自動安裝 Starlette 和 Pydantic;然后還要 pip install uvicorn,因為 uvicorn 是運行相關(guān)應用程序的服務器?;蛘咭徊降轿?#xff1a;pip install fastapi[all],會將所有依賴全部安裝。
教? 程
官網(wǎng)教程:https://fastapi.tiangolo.com/zh/tutorial/
Python 高性能 web 框架 - FastApi 全面指南:https://zhuanlan.zhihu.com/p/397029492
最受歡迎的異步框架:https://www.cnblogs.com/traditional/p/14733610.html
高級用戶指南:https://fastapi.tiangolo.com/zh/advanced/
- 第一步
- 路徑參數(shù)
- 查詢參數(shù)
- 請求體
- 查詢參數(shù)和字符串校驗
- 路徑參數(shù)和數(shù)值校驗
- 請求體 - 多個參數(shù)
- 請求體 - 字段
- 請求體 - 嵌套模型
- 模式的額外信息 - 例子
- 額外數(shù)據(jù)類型
- Cookie 參數(shù)
- Header 參數(shù)
- 響應模型
- 額外的模型
- 響應狀態(tài)碼
- 表單數(shù)據(jù)
- 請求文件
- 請求表單與文件
- 處理錯誤
- 路徑操作配置
- JSON 兼容編碼器
- 請求體 - 更新數(shù)據(jù)
-
依賴項
-
安全性
- 中間件
- CORS(跨域資源共享)
- SQL (關(guān)系型) 數(shù)據(jù)庫
- 更大的應用 - 多個文件
- 后臺任務
- 元數(shù)據(jù)和文檔 URL
- 靜態(tài)文件
- 測試
- 調(diào)試
Python 類型提示簡介
:https://fastapi.tiangolo.com/zh/python-types/
Python類型注解,你需要知道的都在這里:https://zhuanlan.zhihu.com/p/419955374
對象注解屬性的最佳實踐:https://docs.python.org/zh-cn/3/howto/annotations.html
2、FastAPI?使用
簡單使用示例
示例:同步?代碼
from typing import Optional
from fastapi import FastAPIapp = FastAPI()@app.get("/")
def read_root():return {"Hello": "World"}@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None):return {"item_id": item_id, "q": q}
示例:異步 代碼
新建一個?main.py?文件
import os
import uvicorn
from pathlib import Path
from fastapi import FastAPIapp = FastAPI()@app.get("/")
async def root():return {"message": "Hello World"}if __name__ == '__main__':# os.system('uvicorn main_test:app --reload')uvicorn.run(f'{Path(__file__).stem}:app', host="0.0.0.0", port=5555)pass
FastAPI 推薦使用 uvicorn 來運行服務,Uvicorn 是基于uvloop 和 httptools 構(gòu)建的閃電般快速的 ASGI 服務器。
uvicorn main:app 命令含義如下:
- main:main.py 文件(一個 Python「模塊」)。
- app:在 main.py 文件中通過 app = FastAPI() 創(chuàng)建的對象。
- --reload:讓服務器在更新代碼后重新啟動。僅在開發(fā)時使用該選項。
訪問 http://127.0.0.1:5555? 可以看到 JSON 格式的響應:{"message": "Hello World"}
請求?查詢?參數(shù)
「請求路徑」也通常被稱為「端點」或「路由」。
import uvicorn
from pathlib import Path
from fastapi import FastAPIapp = FastAPI() # 創(chuàng)建 api 對象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]@app.get("/") # 根路由?!刚埱舐窂健挂餐ǔ1环Q為「端點」或「路由」。
async def root():return {"name": "king", 'age': 100}@app.get("/say/{data}")
async def say(data: str, q: int = None):return {"data": data, "q": q}if __name__ == '__main__':# os.system('uvicorn main_test:app --reload')uvicorn.run(f'{Path(__file__).stem}:app', host="127.0.0.1", port=5555)pass
瀏覽器訪問
http://127.0.0.1:5555/
http://127.0.0.1:5555/say/test
http://127.0.0.1:5555/items/?skip=0&limit=10
該查詢是 ? URL中位于關(guān)鍵字之后的一組鍵值對,以&字符分隔。
在 url 中進行查詢:
skip:查詢的起始參數(shù)
limit:查詢的結(jié)束參數(shù)
關(guān)于 @app.get()、@router.get()?區(qū)別
都用于定義 HTTP GET 請求的處理程序,但存在一些區(qū)別。
- @app.get(): 這個方法是在 FastAPI 應用程序?qū)嵗现苯佣x路由的一種簡寫方式。
from fastapi import FastAPIapp = FastAPI()@app.get("/")
async def root():return {"message": "Hello, World!"}
- @router.get(): 這個方法是在 FastAPI 路由器對象(Router)上定義路由的一種方式。你可以創(chuàng)建一個獨立的路由器對象,然后使用 @router.get() 來定義特定路徑的路由。
from fastapi import FastAPI, APIRouterrouter = APIRouter()@router.get("/items")
async def read_items():return {"message": "Read all items"}
上面示例創(chuàng)建一個名為 router 的路由器對象,并使用 router.get() 定義了 "/items" 路徑的 GET 請求處理程序。需要注意的是,如果你使用了多個路由器對象,最終需要將它們添加到 FastAPI 應用程序?qū)嵗?#xff0c;才能生效。
#?通過 app.include_router() 方法,將路由器對象添加到 FastAPI 應用程序?qū)嵗?#xff0c;這樣它的定義的路由才會被正確處理。
app.include_router(router)
總結(jié):app.get() 用于在應用程序?qū)嵗隙x路由,而 router.get() 是在路由器對象上定義路由。兩者的區(qū)別在于定義位置和應用方式,但它們都可以用于處理 HTTP GET 請求。
總 結(jié)
- 導入?
FastAPI
。 - 創(chuàng)建一個?
app
?實例。 - 編寫一個路徑操作裝飾器(如?
@app.get("/")
)。 - 編寫一個路徑操作函數(shù)(如上面的?
def root(): ...
)。 - 運行開發(fā)服務器(如?
uvicorn main:app --reload
)。
請求的 "路徑 參數(shù)"
路徑參數(shù)是必須要體現(xiàn)在參數(shù)中。
「請求路徑」也通常被稱為「端點」或「路由」。
FastAPI 編寫一個簡單的應用程序:
import os
import uvicorn
from pathlib import Path
from fastapi import FastAPI# 類似于 app = Flask(__name__)
app = FastAPI()# 綁定路由和視圖函數(shù)
@app.get("/")
async def root():return {"message": "Hello World"}# 在 Windows 中必須加上 if __name__ == "__main__",
# 否則會拋出 RuntimeError: This event loop is already running
if __name__ == '__main__':# 啟動服務,因為我們這個文件叫做 main_test.py,所以需要啟動 main_test.py 里面的 app# os.system('uvicorn main_test:app --reload')# uvicorn.run("main_test:app", host="0.0.0.0", port=8000)uvicorn.run(f'{Path(__file__).stem}:app', host="0.0.0.0", port=5555)pass
在瀏覽器中輸入 "localhost:5555" 就會顯示相應的輸出,我們看到在視圖函數(shù)中可以直接返回一個字典。當然除了字典,其它的數(shù)據(jù)類型也是可以的。
# -*- coding:utf-8 -*-
from fastapi import FastAPI
import uvicornapp = FastAPI()@app.get("/int")
async def index1():return 666@app.get("/str")
async def index2():return "佛祖保佑,佛祖保佑"@app.get("/bytes")
async def index3():return b"satori"@app.get("/tuple")
async def index4():temp_tuple = ("佛祖保佑", "佛祖保佑")return temp_tuple@app.get("/list")
async def index5():return [{"name": "擒賊先擒王", "age": 18}, {"name": "捉奸先捉雙", "age": 16}]if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
直接使用?requests?發(fā)送請求
import requests
print(requests.get("http://localhost:5555/int").text)
print(requests.get("http://localhost:5555/str").text)
print(requests.get("http://localhost:5555/bytes").text)
print(requests.get("http://localhost:5555/tuple").text)
print(requests.get("http://localhost:5555/list").text)
不過元組自動轉(zhuǎn)成列表返回了。這里我們在路由中指定了路徑,可以看到 FastAPI 中的路徑形式和其它框架并無二致,只不過目前的路徑是寫死的,如果我們想動態(tài)聲明路徑參數(shù)該怎么做呢?
路徑 參數(shù)
可以使用與 Python 格式化字符串相同的語法 來聲明路徑 "參數(shù)" 或 "變量",FastAPI 足以辨別 路徑參數(shù) 和 查詢參數(shù)。
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id):
? ? return {"item_id": item_id}
路徑參數(shù)?item_id
?的值將作為參數(shù)?item_id
?傳遞給你的函數(shù)。運行并訪問 http://127.0.0.1:5555/items/foo? 將會看到響應:{"item_id":"foo"}
?示例:
from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str = None, short: bool = False):item = {"item_id": item_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
路徑參數(shù) item_id 的值將作為參數(shù) item_id 傳遞給你的函數(shù)。聲明不屬于路徑參數(shù)的其他函數(shù)參數(shù)時,它們將被自動解釋為 "查詢字符串" 參數(shù):
看看其訪問路徑,執(zhí)行以下的任何一種 url 訪問方式
http://127.0.0.1:8000/items/老王睡隔壁?short=1
http://127.0.0.1:8000/items/老王睡隔壁?short=True
http://127.0.0.1:8000/items/老王睡隔壁?short=true
http://127.0.0.1:8000/items/老王睡隔壁?short=on
http://127.0.0.1:8000/items/老王睡隔壁?short=yes
可以發(fā)現(xiàn)任何大小寫的字母等都會被轉(zhuǎn)換成 bool 值的參數(shù) True,這就是所謂模糊驗證參數(shù),對于開發(fā)者來說是個好消息。注意:如果 short 參數(shù)沒有默認值,則必須傳參,否則 FastAPI 將報錯。
{
? ? "detail": [
? ? ? ? {
? ? ? ? ? ? "loc": [
? ? ? ? ? ? ? ? "query",
? ? ? ? ? ? ? ? "needy"
? ? ? ? ? ? ],
? ? ? ? ? ? "msg": "field required",
? ? ? ? ? ? "type": "value_error.missing"
? ? ? ? }
? ? ]
}
示例:
from fastapi import FastAPIapp = 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]
查詢字符串是鍵值對的集合,這些鍵值對位于 URL 的 ? 之后,并以 &符號 分隔。
可以使用 Query 對查詢進行額外的校驗:
from typing import Optionalfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: Optional[str] = Query(None, max_length=50)):results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}if q:results.update({"q": q})return results
Query 有如下這些字段校驗:
- min_length?最小長度
- max_length?最大長度
- regex?正則匹配
- Query?第一個參數(shù)為默認值,
...
表示是必需的
Path 和 Query 用法一樣,也能對查詢字段進行校驗。
而且你還可以聲明數(shù)值校驗:
from fastapi import FastAPI, Path, Queryapp = 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
gt
:大于ge
:大于等于lt
:小于le
:小于等于
類似的還有 Cookie:
from typing import Optionalfrom fastapi import Cookie, FastAPIapp = FastAPI()@app.get("/items/")
async def read_items(ads_id: Optional[str] = Cookie(None)):return {"ads_id": ads_id}
以及 Header:
from typing import Optionalfrom fastapi import FastAPI, Headerapp = FastAPI()@app.get("/items/")
async def read_items(user_agent: Optional[str] = Header(None)):return {"User-Agent": user_agent}
請求主體+路徑+查詢參數(shù),在請求主體的基礎上加入 url 動態(tài)路徑參數(shù) 和 查詢參數(shù)
from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: str = Noneprice: floattax: float = Noneapp = FastAPI()@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item, q: str = None):result = {"item_id": item_id, **item.dict()}if q:result.update({"q": q})return result
關(guān)于模板引擎
FastAPI 不像 Flask 那樣自帶 模板引擎(Jinja2),也就是說沒有默認的模板引擎,從另一個角度上說,FastAPI 在模板引擎的選擇上變得更加靈活,極度舒適。
以 Jinja2 模板為例,安裝依賴
pip install jinja2
pip install aiofiles # 用于 fastapi 的異步靜態(tài)文件
# -*- coding:utf-8 -*-
from fastapi import FastAPI, Request
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
import uvicornapp = FastAPI()app.mount("/static", StaticFiles(directory="static"), name="static") # 掛載靜態(tài)文件,指定目錄templates = Jinja2Templates(directory="templates") # 模板目錄@app.get("/data/{data}")
async def read_data(request: Request, data: str):return templates.TemplateResponse("index.html", {"request": request, "data": data})if __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)
html 文件渲染
<html>
<head>
? ? <title>士可殺不可辱(you can kill me, but you can't fuck me)</title>
</head>
<body>
? ? <h1>高呼: {{ data }}</h1>
</body>
</html>
在瀏覽器鍵入 http://127.0.0.1:8000/data/士可殺不可辱
值得注意的是,在返回的 TemplateRespone 響應時,必須帶上 request 的上下文對象,傳入?yún)?shù)放在同一字典。這樣一來,又可以像 Flask 一樣的使用 Jinja2。
為路徑設置 tags 標簽進行分組:
from typing import Optional, Set
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattax: Optional[float] = Nonetags: Set[str] = []@app.post("/items/", response_model=Item, tags=["items"])
async def create_item(item: Item):return item@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"}]
還可以設置 summary 和 description:
from typing import Optional, Set
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattax: Optional[float] = Nonetags: Set[str] = []@app.post("/items/",response_model=Item,summary="Create an item",description="Create an item with all the information, name, description, price, tax and a set of unique tags",
)
async def create_item(item: Item):return item
多行注釋:
from typing import Optional, Set
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattax: Optional[float] = Nonetags: Set[str] = []@app.post("/items/", response_model=Item, summary="Create an item")
async def create_item(item: Item):"""Create an item with all the information:- **name**: each item must have a name- **description**: a long description- **price**: required- **tax**: if the item doesn't have tax, you can omit this- **tags**: a set of unique tag strings for this item"""return item
?廢棄路由:
from fastapi import FastAPIapp = 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"}]
有類型的路徑參數(shù)
可以使用標準的 Python 類型標注為函數(shù)中的路徑參數(shù)聲明類型。
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
? ? return {"item_id": item_id}
在這個例子中,item_id 被聲明為 int 類型。
運行示例,并打開瀏覽器訪問 http://127.0.0.1:5555/items/3
將得到如下響應:{"item_id":3}
注意函數(shù)接收(并返回)的值為 3,是一個 Python int 值,而不是字符串 "3"。
所以,FastAPI 通過上面的類型聲明提供了對請求的自動"解析"。
# -*- coding:utf-8 -*-from fastapi import FastAPI
import uvicornapp = FastAPI()@app.get("/apple/{item_id}")
async def get_item(item_id: int):"""Flask 定義類型是在路由當中,也就是在 <> 里面,變量和類型通過 : 分隔FastAPI 是使用類型注解的方式,此時的 item_id 要求一個整型(準確的說是一個能夠轉(zhuǎn)成整型的字符串)"""return {"item_id": item_id}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
順序很重要
由于路徑操作是按順序依次運行的,所以在定義路由的時候需要注意一下順序
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/me
?在?/users/{user_id}
?的前面,否則的話只會匹配到?/users/{user_id}
,此時如果訪問?/users/me
,那么會返回一個解析錯誤,因為字符串?"me" 無法解析成整型。
預設值 (?枚舉 )
可以將某個路徑參數(shù)通過類型注解的方式聲明為指定的類型( 準確的說是可以轉(zhuǎn)成指定的類型,因為默認都是字符串 ),但如果我們希望它只能是我們規(guī)定的幾個值之一該怎么做呢?
這時可以使用標準的 Python?Enum
?類型。導入?Enum
?并創(chuàng)建一個繼承自?str
?和?Enum
?的子類。通過從?str
?繼承,API 文檔將能夠知道這些值必須為?string
?類型并且能夠正確地展示出來。然后創(chuàng)建具有固定值的類屬性,這些固定值將是可用的有效值:
from enum import Enumfrom fastapi import FastAPIclass ModelName(str, Enum):alexnet = "alexnet"resnet = "resnet"lenet = "lenet"app = FastAPI()@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):if model_name is 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"}
路徑中包含 /
假設有這樣一個路由:/files/{file_path},而用戶傳遞的 file_path 中顯然是可以帶 / 的,假設 file_path 是 /root/test.py,那么路由就變成了 /files//root/test.py,顯然這是有問題的。
OpenAPI 不支持任何方式去聲明路徑參數(shù)以在其內(nèi)部包含路徑,因為這可能會導致難以測試和定義的情況出現(xiàn)。不過,仍然可以通過 Starlette 的一個內(nèi)部工具在?FastAPI?中實現(xiàn)它。
可以使用直接來自 Starlette 的選項來聲明一個包含路徑的路徑參數(shù):/files/{file_path:path}
在這種情況下,參數(shù)的名稱為?file_path
,結(jié)尾部分的?:path
?說明該參數(shù)應匹配任意的路徑。
因此,你可以這樣使用它:
from fastapi import FastAPI
app = FastAPI()
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
? ? return {"file_path": file_path}
你可能會需要參數(shù)包含?/home/johndoe/myfile.txt
,以斜杠(/
)開頭。
在這種情況下,URL 將會是?/files//home/johndoe/myfile.txt
,在files
?和?home
?之間有一個雙斜杠(//
)。
from fastapi import FastAPI
import uvicornapp = FastAPI()# 聲明 file_path 的類型為 path,這樣它會被當成一個整體
@app.get("/files/{file_path:path}")
async def get_file(file_path: str):return {"file_path": file_path}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
設置?127.0.0.1:8000/docs
至于?"127.0.0.1:8000/docs" 頁面本身,也是可以進行設置的:
from fastapi import FastAPI
import uvicornapp = FastAPI(title="測試文檔",description="這是一個簡單的 demo",docs_url="/my_docs",openapi_url="/my_openapi"
)@app.get("/apple/{item_id}")
async def get_item(item_id: int):return {"item_id": item_id}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
路徑參數(shù)、數(shù)據(jù)校驗
查詢參數(shù)數(shù)據(jù)校驗使用的是 Query,路徑參數(shù)數(shù)據(jù)校驗使用的是 Path,兩者的使用方式一模一樣,沒有任何區(qū)別
from fastapi import FastAPI, Path
import uvicornapp = FastAPI()@app.get("/items/{item-id}")
async def read_items(item_id: int = Path(..., alias="item-id")):return {"item_id": item_id}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
因為路徑參數(shù)是必須的,它是路徑的一部分,所以我們應該使用 ... 將其標記為必傳參數(shù)。當然即使不這么做也無所謂,因為指定了默認值也用不上,因為路徑參數(shù)不指定壓根就匹配不到相應的路由。至于一些其它的校驗,和查詢參數(shù)一模一樣,所以這里不再贅述了。
不過我們之前說過,路徑參數(shù)應該在查詢參數(shù)的前面,盡管 FastAPI 沒有這個要求,但是這樣寫明顯更舒服一些。但是問題來了,如果路徑參數(shù)需要指定別名,但是某一個查詢參數(shù)不需要,這個時候就會出現(xiàn)問題:
@app.get("/items/{item-id}")
async def read_items(q: str,item_id: int = Path(..., alias="item-id")):return {"item_id": item_id, "q": q}
顯然此時 Python 的語法就決定了 item_id 就必須放在 q 的后面,當然這么做是完全沒有問題的,FastAPI 對參數(shù)的先后順序沒有任何要求,因為它是通過參數(shù)的名稱、類型和默認值聲明來檢測參數(shù),而不在乎參數(shù)的順序。但此時我們就要讓 item_id 在 q 的前面要怎么做呢?
@app.get("/items/{item-id}")
async def read_items(*, item_id: int = Path(..., alias="item-id"),q: str):return {"item_id": item_id, "q": q}
此時就沒有問題了,通過將第一個參數(shù)設置為 *,使得 item_id 和 q 都必須通過關(guān)鍵字傳遞,所以此時默認參數(shù)在非默認參數(shù)之前也是允許的。當然我們也不需要擔心 FastAPI 傳參的問題,你可以認為它所有的參數(shù)都是通過關(guān)鍵字參數(shù)的方式傳遞的。
請求的 "查詢 參數(shù)"
- 查詢參數(shù)在 FastAPI 中可以通過類型注解的方式進行聲明,查詢參數(shù)也可以不寫,那么請求相關(guān)的所有信息都會進入到這個 Request? 對象中
- 如果函數(shù)中定義了不屬于路徑參數(shù)的參數(shù)時,那么它們將會被自動解釋會查詢參數(shù)。
示例:
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]
查詢字符串是鍵值對的集合,這些鍵值對位于 URL 的??
?之后,并以?&
?符號分隔。
例如,在以下 url 中:http://127.0.0.1:5555/items/?skip=0&limit=10
查詢參數(shù)為:
skip
:對應的值為?0
limit
:對應的值為?10
示例:
from fastapi import FastAPI
import uvicornapp = FastAPI()@app.get("/user/{user_id}")
async def get_user(user_id: str, name: str, age: int):"""函數(shù)中參數(shù)定義了 user_id、name、age 三個參數(shù)顯然 user_id 和 路徑參數(shù)中的 user_id 對應,然后 name 和 age 會被解釋成查詢參數(shù)這三個參數(shù)的順序沒有要求,但是一般都是路徑參數(shù)在前,查詢參數(shù)在后"""return {"user_id": user_id, "name": name, "age": age}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
可選參數(shù)、必需參數(shù)
- 當?非路徑參數(shù) 默認值設置為 None,則參數(shù)是可選的
- 當?非路徑參數(shù)?沒有默認值?時,則參數(shù)是必需傳遞的
將它們的默認值設置為?None
?來聲明可選查詢參數(shù):
from typing import Union
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: str, q: Union[str, None] = None):
? ? if q:
? ? ? ? return {"item_id": item_id, "q": q}
? ? return {"item_id": item_id}
在這個例子中,函數(shù)參數(shù) q 默認值為 None,所以q可選的。FastAPI 能夠分辨出參數(shù) item_id 是路徑參數(shù)而 q 不是,因此 q 是一個查詢參數(shù)。
多類型?參數(shù) (?Union )
指定多個類型。比如 user_id 按照整型解析、解析不成功退化為字符串。
from typing import Union, Optional
from fastapi import FastAPI
import uvicornapp = FastAPI()@app.get("/user/{user_id}")
async def get_user(user_id: Union[int, str], name: Optional[str] = None):"""通過 Union 來聲明一個混合類型,int 在前、str 在后。會先按照 int 解析,解析失敗再變成 str然后是 name,它表示字符串類型、但默認值為 None(不是字符串),那么應該聲明為 Optional[str]"""return {"user_id": user_id, "name": name}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
所以 FastAPI 的設計還是非常不錯的,通過 Python 的類型注解來實現(xiàn)參數(shù)類型的限定可以說是非常巧妙的,因此這也需要我們熟練掌握 Python 的類型注解。
bool 類型自動轉(zhuǎn)換
對于布爾類型,FastAPI 支持自動轉(zhuǎn)換
from fastapi import FastAPI
import uvicornapp = FastAPI()@app.get("/{flag}")
async def get_flag(flag: bool):return {"flag": flag}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
print(requests.get("http://localhost:5555/1").json())
print(requests.get("http://localhost:5555/True").json())
print(requests.get("http://localhost:5555/true").json())
print(requests.get("http://localhost:5555/False").json())
print(requests.get("http://localhost:5555/false").json())
print(requests.get("http://localhost:5555/on").json())
print(requests.get("http://localhost:5555/yes").json())
print(requests.get("http://localhost:5555/off").json())
print(requests.get("http://localhost:5555/no").json())
多個路徑、查詢參數(shù)
可以同時聲明多個路徑參數(shù)和查詢參數(shù),FastAPI 能夠識別它們。而且你不需要以任何特定的順序來聲明。它們將通過名稱被檢測到:
from typing import Union
from fastapi import FastAPI
app = FastAPI()
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
? ? user_id: int, item_id: str, q: 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
FastAPI 可以定義任意個路徑參數(shù),只要動態(tài)的路徑參數(shù)在函數(shù)的參數(shù)中都出現(xiàn)即可。當然查詢參數(shù)也可以是任意個,FastAPI 可以處理的很好。
from typing import Optional
from fastapi import FastAPI
import uvicornapp = FastAPI()@app.get("/postgres/{schema}/v1/{table}")
async def get_data(schema: str,table: str,select: str = "*",where: Optional[str] = None,limit: Optional[int] = None,offset: Optional[int] = None):"""標準格式是:路徑參數(shù)按照順序在前,查詢參數(shù)在后但其實對順序是沒有什么要求的"""query = f"select {select} from {schema}.{table}"if where:query += f" where {where}"if limit:query += f" limit {limit}"if offset:query += f" offset {offset}"return {"query": query}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
http://localhost:5555/postgres/ods/v1/staff
http://localhost:5555/postgres/ods/v1/staff?select=id, name&where=id > 3&limit=100
Depends (?依賴注入 )
from typing import Optional
import uvicorn
from fastapi import Depends, FastAPIapp = FastAPI()async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):return {"q": q, "skip": skip, "limit": limit}@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):# common_parameters 接收三個參數(shù):q、skip、limit# 然后在解析請求的時候,會將 q、skip、limit 傳遞到 common_parameters 中,然后將返回值賦值給 commons# 但如果解析不到某個參數(shù)時,那么會判斷函數(shù)中參數(shù)是否有默認值,沒有的話就會返回錯誤,而不是傳遞一個 None 進去return commons@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):return commonsif __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
requests.get("http://localhost:5555/items").json()
{q': None,'skip':@,limit': 100}
requests.get("http://localhost:5555/items?g=id,name").json()
{g':id, name',skip': 0,limit': 100}
所以 Depends 能夠很好的實現(xiàn)依賴注入,而且我們特意寫了兩個路由,就是想表示它們是彼此獨立的。因此當有共享的邏輯、或者共享的數(shù)據(jù)庫連接、增強安全性、身份驗證、角色權(quán)限等等,會非常的實用。
FastAPI 提供了簡單易用,但功能強大的依賴注入系統(tǒng),可以讓開發(fā)人員輕松地把組件集成至FastAPI。
什么是「依賴注入」?
依賴注入是一種消除類之間依賴關(guān)系的設計模式。把有依賴關(guān)系的類放到容器中,解析出這些類的實例,就是依賴注入。目的是實現(xiàn)類的解耦。
示例:
from typing import Optional
from fastapi import Depends, FastAPIapp = FastAPI()async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):return {"q": q, "skip": skip, "limit": limit}@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):return commons@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):return commons
本例中的依賴項預期接收如下參數(shù):
- 類型為?
str
?的可選查詢參數(shù)?q
- 類型為?
int
?的可選查詢參數(shù)?skip
,默認值是?0
- 類型為?
int
?的可選查詢參數(shù)?limit
,默認值是?100
然后,依賴項函數(shù)返回包含這些值的?dict
。
使用Class作為依賴:
from typing import Optional
from fastapi import Depends, FastAPIapp = FastAPI()fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]class CommonQueryParams:def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):self.q = qself.skip = skipself.limit = limit@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):response = {}if commons.q:response.update({"q": commons.q})items = fake_items_db[commons.skip: commons.skip + commons.limit]response.update({"items": items})return response
使用嵌套子依賴:
from typing import Optional
from fastapi import Cookie, Depends, FastAPIapp = FastAPI()def query_extractor(q: Optional[str] = None):return qdef query_or_cookie_extractor(q: str = Depends(query_extractor), last_query: Optional[str] = Cookie(None)
):if not q:return last_queryreturn q@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):return {"q_or_cookie": query_or_default}
在路徑中使用依賴:
from fastapi import Depends, FastAPI, Header, HTTPExceptionapp = 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"}]
全局依賴項,可以為所有路徑操作應用該依賴項:
from fastapi import Depends, FastAPI, Header, HTTPExceptionasync 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_keyapp = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])@app.get("/items/")
async def read_items():return [{"item": "Portal Gun"}, {"item": "Plumbus"}]@app.get("/users/")
async def read_users():return [{"username": "Rick"}, {"username": "Morty"}]
查詢參數(shù)、數(shù)據(jù)校驗
FastAPI 支持我們進行更加智能的數(shù)據(jù)校驗,比如一個字符串,我們希望用戶在傳遞的時候只能傳遞長度為 6 到 15 的字符串該怎么做呢?
from typing import Optional
from fastapi import FastAPI, Query
import uvicornapp = FastAPI()@app.get("/user")
async def check_length(# 默認值為 None,應該聲明為 Optional[str],當然聲明 str 也是可以的。只不過聲明為 str,那么默認值應該也是 str# 所以如果一個類型允許為空,那么更規(guī)范的做法應該是聲明為 Optional[類型]。password: Optional[str] = Query(None, min_length=6, max_length=15)
):return {"password": password}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
password 是可選的,但是一旦傳遞則必須傳遞字符串、而且還是長度在 6 到 15 之間的字符串。所以如果傳遞的是 None,那么在聲明默認值的時候 None 和 Query(None) 是等價的,只不過 Query 還支持其它的參數(shù)來對參數(shù)進行限制。
requests.get("http://localhost:5555/user?password=12345").json()
requests.get("http://localhost:5555/user?password=123456").json()
Query 里面除了限制最小長度和最大長度,還有其它的功能
查詢參數(shù)和字符串校驗
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/")
async def read_items(q: str | None = None):
? ? results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
? ? if q:
? ? ? ? results.update({"q": q})
? ? return results
查詢參數(shù)?q
?的類型為?str
,默認值為?None
,因此它是可選的。
額外的校驗
添加約束條件:即使?q
?是可選的,但只要提供了該參數(shù),則該參數(shù)值不能超過50個字符的長度。
從 fastapi 導入 Query:
from typing import Union
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, max_length=50)):
? ? results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
? ? if q:
? ? ? ? results.update({"q": q})
? ? return results
添加更多校驗
還可以添加?min_length
?參數(shù):
from typing import Union
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/")
async def read_items(
? ? q: Union[str, None] = Query(default=None, min_length=3, max_length=50)
):
? ? results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
? ? if q:
? ? ? ? results.update({"q": q})
? ? return results
添加正則表達式
可以定義一個參數(shù)值必須匹配的正則表達式:
from typing import Union
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/")
async def read_items(
? ? q: Union[str, None] = Query(
? ? ? ? default=None, min_length=3, max_length=50, pattern="^fixedquery$"
? ? )
):
? ? results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
? ? if q:
? ? ? ? results.update({"q": q})
? ? return results
聲明為必須參數(shù)
... 是 Python 中的一個特殊的對象,通過它可以實現(xiàn)該參數(shù)是必傳參數(shù)。
from fastapi import FastAPI, Query
import uvicornapp = FastAPI()@app.get("/user")
async def check_length(password: str = Query(..., min_length=6)):"""將第一個參數(shù)換成 ... 即可實現(xiàn)該參數(shù)是必傳參數(shù)"""return {"password": password}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
查詢參數(shù)變成一個列表
如果我們指定了?a=1&a=2
,那么我們在獲取 a 的時候如何才能得到一個列表呢?
from typing import Optional, List
from fastapi import FastAPI, Query
import uvicornapp = FastAPI()@app.get("/items")
async def read_items(a1: str = Query(...),a2: List[str] = Query(...),b: List[str] = Query(...)
):return {"a1": a1, "a2": a2, "b": b}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
首先?"a2" 和?"b" 都是對應列表,然后?"a1" 只獲取了最后一個值。另外可能有人覺得我們這樣有點啰嗦,在函數(shù)聲明中可不可以這樣寫呢?
@app.get("/items")
async def read_items(a1: str,a2: List[str],b: List[str]
):return {"a1": a1, "a2": a2, "b": b}
對于 a1 是可以的,但是 a2 和 b 不行。對于類型為 list 的查詢參數(shù),無論有沒有默認值,你都必須要顯式的加上 Query 來表示必傳參數(shù)。如果允許為 None(或者有默認值)的話,那么應該這么寫:
from typing import Optional, List
from fastapi import FastAPI, Query
import uvicornapp = FastAPI()@app.get("/items")
async def read_items(a1: str,a2: Optional[List[str]] = Query(None),b: List[str] = Query(["1", "嘿嘿"])
):return {"a1": a1, "a2": a2, "b": b}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
給參數(shù)起別名
假設我們定義的查詢參數(shù)名叫 item-query,那么由于它要體現(xiàn)在函數(shù)參數(shù)中、而這顯然不符合 Python 變量的命名規(guī)范,這個時候要怎么做呢?答案是起一個別名。
from typing import Optional, List
from fastapi import FastAPI, Query
import uvicornapp = FastAPI()@app.get("/items")
async def read_items(# 通過 url 的時候使用別名即可item1: Optional[str] = Query(None, alias="item-query"),item2: str = Query("哈哈", alias="@@@@"),item3: str = Query(..., alias="$$$$") # item3 是必傳的
):return {"item1": item1, "item2": item2, "item3": item3}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
數(shù)值檢測
Query 不僅僅支持對字符串的校驗,還支持對數(shù)值的校驗,里面可以傳遞 gt、ge、lt、le 這幾個參數(shù),相信這幾個參數(shù)不用說你也知道是干什么的,我們舉例說明:
from fastapi import FastAPI, Query
import uvicornapp = FastAPI()@app.get("/items")
async def read_items(# item1 必須大于 5item1: int = Query(..., gt=5),# item2 必須小于等于 7item2: int = Query(..., le=7),# item3 必須必須等于 10item3: int = Query(..., ge=10, le=10)
):return {"item1": item1, "item2": item2, "item3": item3}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
Request?對象
Request 是什么?首先我們知道任何一個請求都對應一個 Request 對象,請求的所有信息都在這個 Request 對象中,FastAPI 也不例外。
路徑參數(shù)是必須要體現(xiàn)在參數(shù)中,但是查詢參數(shù)可以不寫了 因為我們定義了 request: Request,那么請求相關(guān)的所有信息都會進入到這個 Request 對象中
from fastapi import FastAPI, Request
import uvicornapp = FastAPI()@app.get("/girl/{user_id}")
async def read_girl(user_id: str,request: Request):"""路徑參數(shù)是必須要體現(xiàn)在參數(shù)中,但是查詢參數(shù)可以不寫了因為我們定義了 request: Request,那么請求相關(guān)的所有信息都會進入到這個 Request 對象中"""header = request.headers # 請求頭method = request.method # 請求方法cookies = request.cookies # cookiesquery_params = request.query_params # 查詢參數(shù)return {"name": query_params.get("name"), "age": query_params.get("age"), "hobby": query_params.getlist("hobby")}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
通過 Request 對象可以獲取所有請求相關(guān)的信息,我們之前當參數(shù)傳遞不對的時候,FastAPI 會自動幫我們返回錯誤信息,但通過 Request 我們就可以自己進行解析、自己指定返回的錯誤信息了。
Response?對象
既然有 Request,那么必然會有 Response,盡管我們可以直接返回一個字典,但 FastAPI 實際上會幫我們轉(zhuǎn)成一個 Response 對象。
Response 內(nèi)部接收如下參數(shù):
content:返回的數(shù)據(jù)
status_code:狀態(tài)碼
headers:返回的請求頭
media_type:響應類型(就是 HTML 中 Content-Type,只不過這里換了個名字)
background:接收一個任務,Response 在返回之后會自動異步執(zhí)行
示例
from fastapi import FastAPI, Request, Response
import uvicorn
import orjsonapp = FastAPI()@app.get("/girl/{user_id}")
async def read_girl(user_id: str, request: Request):query_params = request.query_params # 查詢參數(shù)data = {"name": query_params.get("name"), "age": query_params.get("age"), "hobby": query_params.getlist("hobby")}# 實例化一個 Response 對象response = Response(# content,我們需要手動轉(zhuǎn)成 json 字符串,如果直接返回字典的話,那么在包裝成 Response 對象的時候會自動幫你轉(zhuǎn)orjson.dumps(data),# status_code,狀態(tài)碼201,# headers,響應頭{"Token": "xxx"},# media_type,就是 HTML 中的 Content-Type"application/json",)# 如果想設置 cookie 的話,那么通過 response.set_cookie 即可# 刪除 cookie 則是 response.delete_cookiereturn responseif __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
通過 Response 我們可以實現(xiàn)請求頭、狀態(tài)碼、cookie 等自定義。
另外除了 Response 之外還有很多其它類型的響應,它們都在 fastapi.responses 中,比如:FileResponse、HTMLResponse、PlainTextResponse 等等。它們都繼承了 Response,只不過會自動幫你設置響應類型,舉個栗子:
from fastapi import FastAPI
from fastapi.responses import Response, HTMLResponse
import uvicornapp = FastAPI()@app.get("/index")
async def index():response1 = HTMLResponse("<h1>你好呀</h1>")response2 = Response("<h1>你好呀</h1>", media_type="text/html")# 以上兩者是等價的,在 HTMLResponse 中會自動將 media_type 設置成 text/htmlreturn response1if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
?使用?response_model
參數(shù)來聲明用于響應的模型:
from typing import List, Optional
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattax: Optional[float] = Nonetags: List[str] = []@app.post("/items/", response_model=Item)
async def create_item(item: Item):return item
response_model_exclude_unset=True
:響應中將不會包含那些默認值,而是僅有實際設置的值response_model_include?
包含哪些屬性response_model_exclude?
省略某些屬性
status_code
參數(shù)來聲明用于響應的 HTTP 狀態(tài)碼:
from fastapi import FastAPIapp = FastAPI()@app.post("/items/", status_code=201)
async def create_item(name: str):return {"name": name}
表單字段時,要使用Form
:
from fastapi import FastAPI, Formapp = FastAPI()@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):return {"username": username}
File
用于定義客戶端的上傳文件(接收上傳文件,要預先安裝
):python-multipart
from fastapi import FastAPI, File, UploadFileapp = FastAPI()@app.post("/files/")
async def create_file(file: bytes = File(...)):return {"file_size": len(file)}@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):return {"filename": file.filename}
向客戶端返回 HTTP 錯誤響應,可以使用HTTPException
。
from fastapi import FastAPI, HTTPExceptionapp = FastAPI()items = {"foo": "The Foo Wrestlers"}@app.get("/items/{item_id}")
async def read_item(item_id: str):if item_id not in items:raise HTTPException(status_code=404, detail="Item not found")return {"item": items[item_id]}
使用response_description
設置響應描述:
from typing import Optional, Setfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattax: Optional[float] = Nonetags: Set[str] = []@app.post("/items/",response_model=Item,summary="Create an item",response_description="The created item",
)
async def create_item(item: Item):"""Create an item with all the information:- **name**: each item must have a name- **description**: a long description- **price**: required- **tax**: if the item doesn't have tax, you can omit this- **tags**: a set of unique tag strings for this item"""return item
請求體
當需要將數(shù)據(jù)從客戶端(例如瀏覽器)發(fā)送給 API 時,是將其作為「請求體」發(fā)送。
請求體是客戶端發(fā)送給 API 的數(shù)據(jù)。
響應體是 API 發(fā)送給客戶端的數(shù)據(jù)。
要發(fā)送數(shù)據(jù),你必須使用下列方法之一:
POST
(較常見)、PUT
、DELETE
?或?PATCH
。
顯然對應 POST、PUT 等類型的請求,我們必須要能夠解析出請求體,并且能夠構(gòu)造出響應體。
Model
在 FastAPI 中,請求體和響應體都對應一個 Model
from typing import Optional, List
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicornapp = FastAPI()class Girl(BaseModel):"""數(shù)據(jù)驗證是通過 pydantic 實現(xiàn)的,我們需要從中導入 BaseModel,然后繼承它"""name: strage: Optional[str] = Nonelength: floathobby: List[str] # 對于 Model 中的 List[str] 我們不需要指定 Query(準確的說是 Field)@app.post("/girl")
async def read_girl(girl: Girl):# girl 就是我們接收的請求體,它需要通過 json 來傳遞,并且這個 json 要有上面的四個字段(age 可以沒有)# 通過 girl.xxx 的方式我們可以獲取和修改內(nèi)部的所有屬性return dict(girl) # 直接返回 Model 對象也是可以的if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
示例:
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
? ? name: str
? ? description: str | None = None
? ? price: float
? ? tax: float | None = None
app = FastAPI()
@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
和聲明查詢參數(shù)時一樣,當一個模型屬性具有默認值時,它不是必需的。否則它是一個必需屬性。將默認值設為?None
?可使其成為可選屬性。
使用 Request 對象?傳遞?body
如果你不想使用 Pydantic 模型,你還可以使用?Body?參數(shù)。請參閱文檔?請求體 - 多個參數(shù):請求體中的單一值。
from fastapi import FastAPI, Request
import uvicornapp = FastAPI()@app.post("/girl")
async def read_girl(request: Request):# 是一個協(xié)程,所以需要 awaitdata = await request.body()print(data)if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
使用 requests 模塊發(fā)送 post 請求的時候可以通過 data 參數(shù)傳遞、也可以通過 json 參數(shù)。
- 當通過 json={"name": "satori", "age": 16, "length": 155.5} 傳遞的時候,會將其轉(zhuǎn)成 json 字符串進行傳輸,程序中的 print 打印如下:b'{"name": "satori", "age": 16, "length": 155.5}'
- 如果用?data 參數(shù)發(fā)請求的話(值不變),那么會將其拼接成 k1=v1&k2=v2 的形式再進行傳輸(相當于表單提交,后面說),程序中打印如下:b'name=satori&age=16&length=155.5'
所以我們看到 await request.body() 得到的就是最原始的字節(jié)流,而除了 await request.body() 之外還有一個 await request.json(),只是后者在內(nèi)部在調(diào)用了前者拿到字節(jié)流之后、自動幫你 loads 成了字典。因此使用 await request.json() 也側(cè)面要求我們必須在發(fā)送請求的時候必須使用 json 參數(shù)傳遞(傳遞的是字典轉(zhuǎn)成的 json、所以也能解析成字典),否則使用 await request.json() 是無法正確解析的。
路徑參數(shù)、查詢參數(shù)、請求體
- 如果在?路徑?中也聲明了該參數(shù),它將被用作路徑參數(shù)。
- 如果參數(shù)屬于?單一類型(比如?
int
、float
、str
、bool
?等)它將被解釋為?查詢?參數(shù)。 - 如果參數(shù)的類型被聲明為一個?Pydantic 模型,它將被解釋為?請求體。
# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicornapp = FastAPI()class Girl(BaseModel):name: strage: Optional[str] = Nonelength: floathobby: List[str]@app.post("/girl/{user_id}")
async def read_girl(user_id, q: str, girl: Girl):return {"user_id": user_id, "q": q, **dict(girl)}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
指定了路徑參數(shù)、查詢參數(shù)和請求體,FastAPI 依然是可以正確區(qū)分的,當然我們也可以使用 Request 對象。
from typing import Optional, List, Dict
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicornapp = FastAPI()@app.post("/girl/{user_id}")
async def read_girl(user_id, request: Request):q = request.query_params.get("q")data: Dict = await request.json()data.update({"user_id": user_id, "q": q})return dataif __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
嵌入 單個請求體 參數(shù)
假設你只有一個來自 Pydantic 模型?Item
?的請求體參數(shù)?item
。默認情況下,FastAPI?將直接期望這樣的請求體。但是,如果你希望它期望一個擁有?item
?鍵并在值中包含模型內(nèi)容的 JSON,就像在聲明額外的請求體參數(shù)時所做的那樣,則可以使用一個特殊的?Body
?參數(shù)?embed
:item: Item = Body(embed=True)
from typing import Annotated
from fastapi import Body, FastAPI
from pydantic import BaseModelapp = FastAPI()
class Item(BaseModel):
? ? name: str
? ? description: str | None = None
? ? price: float
? ? tax: float | None = None
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
? ? results = {"item_id": item_id, "item": item}
? ? return results
在這種情況下,FastAPI?將期望像這樣的請求體:
{
? ? "item": {
? ? ? ? "name": "Foo",
? ? ? ? "description": "The pretender",
? ? ? ? "price": 42.0,
? ? ? ? "tax": 3.2
? ? }
}
而不是:
{
? ? "name": "Foo",
? ? "description": "The pretender",
? ? "price": 42.0,
? ? "tax": 3.2
}
可以添加多個請求體參數(shù)到路徑操作函數(shù)中,即使一個請求只能有一個請求體。
還可以聲明將作為請求體的一部分所接收的單一值。
還可以指示?FastAPI?在僅聲明了一個請求體參數(shù)的情況下,將原本的請求體嵌入到一個鍵中。
多個請求體參數(shù)
# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicornapp = FastAPI()class Girl(BaseModel):name: strage: Optional[str] = Noneclass Boy(BaseModel):name: strage: int@app.post("/boy_and_girl")
async def read_boy_and_girl(girl: Girl, boy: Boy):return {"girl": dict(girl), "boy": dict(boy)}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
此時在傳遞的時候,應該按照如下方式傳遞:
應該將兩個 json 嵌套在一起,組成一個更大的 json,至于 key 就是我們的函數(shù)參數(shù)名。因此這種方式其實就等價于:
# -*- coding:utf-8 -*-
# @Author: komeiji satori
from typing import Optional, List, Dict
from fastapi import FastAPI, Request, Response
from pydantic import BaseModel
import uvicornapp = FastAPI()class BoyAndGirl(BaseModel):girl: Dictboy: Dict@app.post("/boy_and_girl")
async def read_boy_and_girl(boy_and_girl: BoyAndGirl):return dict(boy_and_girl)if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
這種方式也是可以實現(xiàn)的,只不過就字典內(nèi)部的字典的不可進行限制了。當然啦,我們?nèi)匀豢梢允褂?Request 對象,得到字典之后自己再進行判斷,因為對于 json 而言,內(nèi)部的字段可能是會變的,而且最關(guān)鍵的是字段可能非常多。這個時候,我個人更傾向于使用 Request 對象。
混合使用?Path
、Query
?和請求體參數(shù)
可以隨意地混合使用?Path
、Query
?和請求體參數(shù)聲明,FastAPI?會知道該如何處理。
from typing import Annotated
from fastapi import FastAPI, Path
from pydantic import BaseModelapp = FastAPI()
class Item(BaseModel):
? ? name: str
? ? description: str | None = None
? ? price: float
? ? tax: float | None = None
@app.put("/items/{item_id}")
async def update_item(
? ? item_id: Annotated[int, Path(title="The ID of the item to get", ge=0, le=1000)],
? ? q: str | None = None,
? ? item: Item | None = None,
):
? ? results = {"item_id": item_id}
? ? if q:
? ? ? ? results.update({"q": q})
? ? if item:
? ? ? ? results.update({"item": item})
? ? return results
請注意,在這種情況下,將從請求體獲取的?item
?是可選的。因為它的默認值為?None
。
多個 請求體 參數(shù)
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()
class Item(BaseModel):
? ? name: str
? ? description: str | None = None
? ? price: float
? ? tax: float | None = None
class User(BaseModel):
? ? username: str
? ? full_name: 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"
? ? }
}
請求體中的單一值
from typing import Annotated
from fastapi import Body, FastAPI
from pydantic import BaseModelapp = FastAPI()
class Item(BaseModel):
? ? name: str
? ? description: str | None = None
? ? price: float
? ? tax: float | None = None
class User(BaseModel):
? ? username: str
? ? full_name: str | None = None
@app.put("/items/{item_id}")
async def update_item(
? ? item_id: int, item: Item, user: User, importance: Annotated[int, Body()]
):
? ? results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
? ? return results
請求體:
{
? ? "item": {
? ? ? ? "name": "Foo",
? ? ? ? "description": "The pretender",
? ? ? ? "price": 42.0,
? ? ? ? "tax": 3.2
? ? },
? ? "user": {
? ? ? ? "username": "dave",
? ? ? ? "full_name": "Dave Grohl"
? ? },
? ? "importance": 5
}
多個 請求體參數(shù) 和 查詢參數(shù)
默認情況下單一值被解釋為查詢參數(shù),因此你不必顯式地添加?Query
from typing import Annotated
from fastapi import Body, FastAPI
from pydantic import BaseModelapp = FastAPI()
class Item(BaseModel):
? ? name: str
? ? description: str | None = None
? ? price: float
? ? tax: float | None = None
class User(BaseModel):
? ? username: str
? ? full_name: str | None = None
@app.put("/items/{item_id}")
async def update_item(
? ? *,
? ? item_id: int,
? ? item: Item,
? ? user: User,
? ? importance: Annotated[int, Body(gt=0)],
? ? q: str | None = None,
):
? ? results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
? ? if q:
? ? ? ? results.update({"q": q})
? ? return results
Form 表單
調(diào)用 requests.post,如果參數(shù)通過 data 傳遞的話,則相當于提交了一個 form 表單,那么在 FastAPI 中可以通過 await request.form() 進行獲取,注意:內(nèi)部同樣是先調(diào)用 await request.body()。
from fastapi import FastAPI, Request, Response
import uvicornapp = FastAPI()@app.post("/girl")
async def girl(request: Request):# 此時 await request.json() 報錯,因為是通過 data 參數(shù)傳遞的,相當于 form 表單提交# 如果是通過 json 參數(shù)傳遞,那么 await request.form() 會得到一個空表單form = await request.form()return [form.get("name"), form.getlist("age")]if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
也可以通過其它方式:
# -*- coding:utf-8 -*-
# @Author: komeiji satori
from fastapi import FastAPI, Form
import uvicornapp = FastAPI()@app.post("/user")
async def get_user(username: str = Form(...),password: str = Form(...)):return {"username": username, "password": password}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
像 Form 表單,查詢參數(shù)、路徑參數(shù)等等,都可以和 Request 對象一起使用,像上面的例子,如果我們多定義一個 request: Request,那么我們?nèi)匀豢梢酝ㄟ^ await request.form() 拿到相關(guān)的表單信息。所以如果你覺得某個參數(shù)不適合類型注解,那么你可以單獨通過 Request 對象進行解析。
文件上傳
FastAPI 如何接收用戶的文件上傳呢?首先如果想使用文件上傳功能,那么你必須要安裝一個包 python-multipart,直接?pip install python-multipart 即可。
from fastapi import FastAPI, File, UploadFile
import uvicornapp = FastAPI()@app.post("/file1")
async def file1(file: bytes = File(...)):return f"文件長度: {len(file)}"@app.post("/file2")
async def file1(file: UploadFile = File(...)):return f"文件名: {file.filename}, 文件大小: {len(await file.read())}"if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
我們看到一個直接獲取字節(jié)流,另一個是獲取類似于文件句柄的對象。如果是多個文件上傳要怎么做呢?
from typing import List
from fastapi import FastAPI, UploadFile, File
import uvicornapp = FastAPI()@app.post("/file")
async def file(files: List[UploadFile] = File(...)):"""指定類型為列表即可"""for idx, f in enumerate(files):files[idx] = f"文件名: {f.filename}, 文件大小: {len(await f.read())}"return filesif __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
此時就實現(xiàn)了 FastAPI 文件上傳,當然文件上傳并不影響我們處理表單,可以自己試一下同時處理文件和表單。
返回靜態(tài)資源
需要安裝 aiofiles,直接 pip 安裝即可。
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
import uvicornapp = FastAPI()# name 參數(shù)只是起一個名字,FastAPI 內(nèi)部使用
app.mount("/static", StaticFiles(directory=r"C:\Users\satori\Desktop\bg"), name="static")if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
瀏覽器輸入:localhost:5555/static/1.png,那么會返回?C:\Users\satori\Desktop\bg 下的 1.png 文件。
?子應用
如果你有2個獨立的FastAPI的應用,你可以設置一個為主應用,另外一個為子應用:
from fastapi import FastAPIapp = FastAPI()@app.get("/app")
def read_main():return {"message": "Hello World from main app"}subapi = FastAPI()@subapi.get("/sub")
def read_sub():return {"message": "Hello World from sub API"}app.mount("/subapi", subapi)
代理
可以使用root_path
來設置代理。
使用命令行:uvicorn main:app --root-path /api/v1?
或者在代碼中設置:
from fastapi import FastAPI, Requestapp = FastAPI(root_path="/api/v1")@app.get("/app")
def read_main(request: Request):return {"message": "Hello World", "root_path": request.scope.get("root_path")}
使用模板
你可以在FastAPI中使用任何模板,常用的選擇是Jinja2。安裝:pip install jinja2
使用:
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFilesfrom fastapi.templating import Jinja2Templatesapp = FastAPI()app.mount("/static", StaticFiles(directory="static"), name="static")templates = Jinja2Templates(directory="templates")@app.get("/items/{id}", response_class=HTMLResponse)
async def read_item(request: Request, id: str):return templates.TemplateResponse("item.html", {"request": request, "id": id})
模板文件templates/item.html
:
<html>
<head>
? ? <title>Item Details</title>
? ? <link href="{{ url_for('static', path='/styles.css') }}" rel="stylesheet">
</head>
<body>? ? <h1>Item ID: {{ id }}</h1>
</body>
</html>
錯誤處理
from fastapi import FastAPI, HTTPException
import uvicornapp = FastAPI()@app.get("/items/{item_id}")
async def read_item(item_id: str):if item_id != "foo":# 里面還可以傳入 headers 設置響應頭raise HTTPException(status_code=404, detail="item 沒有發(fā)現(xiàn)")return {"item": "bar"}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
HTTPException 是一個普通的 Python 異常類(繼承了 Exception),它攜帶了 API 的相關(guān)信息,既然是異常,那么我們不能 return、而是要 raise。這種方式返回錯誤,因為它能夠攜帶的信息太少了。
自定義異常
FastAPI 內(nèi)部提供了一個?HTTPException,但是我們也可以自定義,但是注意:我們自定義完異常之后,還要定義一個 handler,將異常和 handler 綁定在一起,然后引發(fā)該異常的時候就會觸發(fā)相應的 handler。
from fastapi import FastAPI, Request
from fastapi.responses import ORJSONResponse
import uvicornapp = FastAPI()class ASCIIException(Exception):""""""pass# 通過裝飾器的方式,將 ASCIIException 和 ascii_exception_handler 綁定在一起
@app.exception_handler(ASCIIException)
async def ascii_exception_handler(request: Request, exc: ASCIIException):"""當引發(fā) ASCIIException 的時候,會觸發(fā) ascii_exception_handler 的執(zhí)行同時會將 request 和 exception 傳過去"""return ORJSONResponse(status_code=404, content={"code": 404, "message": "你必須傳遞 ascii 字符串"})@app.get("/items/{item_id}")
async def read_item(item_id: str):if not item_id.isascii():raise ASCIIExceptionreturn {"item": f"get {item_id}"}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
關(guān)于 Request、Response,我們除了可以通過 fastapi 進行導入,還可以通過?starlette 進行導入,因為 fastapi 的路由映射是通過?starlette 來實現(xiàn)的。
自定義 404
當訪問一個不存在的 URL,我們應該提示用戶,比如:您要找到頁面去火星了。
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
from fastapi.exceptions import StarletteHTTPException
import uvicornapp = FastAPI()@app.exception_handler(StarletteHTTPException)
async def not_found(request, exc):return ORJSONResponse({"code": 404, "message": "您要找的頁面去火星了。。。"})if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
此時當我們訪問一個不存在的 URL 時,就會返回我們自定義的 JSON 字符串。
Background Tasks (?后臺任務 )
background tasks 就是在返回響應之后立即運行的任務。
如果一個請求耗時特別久,那么我們可以將其放在后臺執(zhí)行,而 FastAPI 已經(jīng)幫我們做好了這一步。我們來看一下:
import time
from fastapi import FastAPI, BackgroundTasks
from starlette.background import BackgroundTask
from fastapi import Response, Request
import uvicorn
import orjsonapp = FastAPI()def send_email(email: str, message: str = ""):"""發(fā)送郵件,假設耗時三秒"""time.sleep(3)print(f"三秒之后郵件發(fā)送給 {email!r}, 郵件信息: {message!r}")@app.get("/user/{email}")
async def order(email: str, bg_tasks: BackgroundTasks):"""這里需要多定義一個參數(shù)此時任務就被添加到后臺,當 Response 對象返回之后觸發(fā)"""bg_tasks.add_task(send_email, email, message="這是一封郵件")# 我們在之前介紹 Response 的時候說過,里面有一個參數(shù) background# 所以我們也可以將任務放在那里面# 因此我們還可以:# return Response(# orjson.dumps({"message": "郵件發(fā)送成功"}), # background=BackgroundTask(send_email, email, message="這是一封郵件")# )return {"message": "郵件發(fā)送成功"}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
APIRouter
APIRouter 類似于 Flask 中的藍圖,可以更好的組織大型項目,舉個栗子:
在我當前的工程目錄中有一個 app 目錄和一個 main.py,其中 app 目錄中有一個 app01.py,然后我們看看它們是如何組織的。
app/app01.py
# app/app01.py
from fastapi import APIRouterrouter = APIRouter(prefix="/router")# 以后訪問的時候要通過 /router/v1 來訪問
@router.get("/v1")
async def v1():return {"message": "hello world"}
main.py
# main.py
from fastapi import FastAPI
from app.app01 import router
import uvicornapp = FastAPI()# 將 router 注冊到 app 中,相當于 Flask 中的 register_blueprint
app.include_router(router)if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
然后可以在外界通過 /router/v1 的方式來訪問。
示例:
from fastapi import APIRouterrouter = APIRouter()@router.get("/users/", tags=["users"])
async def read_users():return [{"username": "Rick"}, {"username": "Morty"}]@router.get("/users/me", tags=["users"])
async def read_user_me():return {"username": "fakecurrentuser"}@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):return {"username": username}
為所有路徑進行同樣的操作:
from fastapi import APIRouter, Depends, HTTPExceptionfrom ..dependencies import get_token_headerrouter = APIRouter(prefix="/items",tags=["items"],dependencies=[Depends(get_token_header)],responses={404: {"description": "Not found"}},
)fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}@router.get("/")
async def read_items():return fake_items_db@router.get("/{item_id}")
async def read_item(item_id: str):if item_id not in fake_items_db:raise HTTPException(status_code=404, detail="Item not found")return {"name": fake_items_db[item_id]["name"], "item_id": item_id}@router.put("/{item_id}",tags=["custom"],responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):if item_id != "plumbus":raise HTTPException(status_code=403, detail="You can only update the item: plumbus")return {"item_id": item_id, "name": "The great Plumbus"}
該示例,就為所有的路徑添加了前綴,標簽、依賴和返回,而不用在每個路徑上單獨聲明,簡化了代碼。
中間件
中間件在 web 開發(fā)中可以說是非常常見了,說白了中間件就是一個函數(shù)或者一個類。在請求進入視圖函數(shù)之前,會先經(jīng)過中間件(被稱為請求中間件),而在中間件里面,我們可以對請求進行一些預處理,或者實現(xiàn)一個攔截器等等;同理當視圖函數(shù)返回響應之后,也會經(jīng)過中間件(被稱為響應中間件),在中間件里面,我們也可以對響應進行一些潤色。
自定義中間件
在 FastAPI 里面也支持像 Flask 一樣自定義中間件,但是 Flask 里面有請求中間件和響應中間件,但是在 FastAPI 里面這兩者合二為一了,我們看一下用法。
from fastapi import FastAPI, Request, Response
import uvicorn
import orjsonapp = FastAPI()@app.get("/")
async def view_func(request: Request):return {"name": "古明地覺"}@app.middleware("http")
async def middleware(request: Request, call_next):"""定義一個協(xié)程函數(shù),然后使用 @app.middleware("http") 裝飾,即可得到中間件"""# 請求到來時會先經(jīng)過這里的中間件if request.headers.get("ping", "") != "pong":response = Response(content=orjson.dumps({"error": "請求頭中缺少指定字段"}),media_type="application/json",status_code=404)# 當請求頭中缺少 "ping": "pong",在中間件這一步就直接返回了,就不會再往下走了# 所以此時就相當于實現(xiàn)了一個攔截器return response# 然后,如果條件滿足,則執(zhí)行 await call_next(request),關(guān)鍵是這里的 call_next# 如果該中間件后面還有中間件,那么 call_next 就是下一個中間件;如果沒有,那么 call_next 就是對應的視圖函數(shù)# 這里顯然是視圖函數(shù),因此執(zhí)行之后會拿到視圖函數(shù)返回的 Response 對象# 所以我們看到在 FastAPI 中,請求中間件和響應中間件合在一起了response: Response = await call_next(request)# 這里我們在設置一個響應頭response.headers["status"] = "success"return responseif __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
內(nèi)置的中間件
通過自定義中間件,我們可以在不修改視圖函數(shù)的情況下,實現(xiàn)功能的擴展。但是除了自定義中間件之外,FastAPI 還提供了很多內(nèi)置的中間件。
from fastapi import FastAPIapp = FastAPI()# 要求請求協(xié)議必須是 https 或者 wss,如果不是,則自動跳轉(zhuǎn)
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware
app.add_middleware(HTTPSRedirectMiddleware)# 請求中必須包含 Host 字段,為防止 HTTP 主機報頭攻擊,并且添加中間件的時候,還可以指定一個 allowed_hosts,那么它是干什么的呢?
# 假設我們有服務 a.example.com, b.example.com, c.example.com
# 但我們不希望用戶訪問 c.example.com,就可以像下面這么設置,如果指定為 ["*"],或者不指定 allow_hosts,則表示無限制
from starlette.middleware.trustedhost import TrustedHostMiddleware
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["a.example.com", "b.example.com"])# 如果用戶的請求頭的 Accept-Encoding 字段包含 gzip,那么 FastAPI 會使用 GZip 算法壓縮
# minimum_size=1000 表示當大小不超過 1000 字節(jié)的時候就不壓縮了
from starlette.middleware.gzip import GZipMiddleware
app.add_middleware(GZipMiddleware, minimum_size=1000)
除了這些,還有其它的一些內(nèi)置的中間件,可以自己查看一下,不過不是很常用。
CORS (?跨域設置 )
CORS 過于重要,我們需要單獨拿出來說。
CORS(跨域資源共享)是指瀏覽器中運行的前端里面擁有和后端通信的 JavaScript 代碼,而前端和后端處于不同源的情況。源:協(xié)議(http、https)、域(baidu.com、app.com、localhost)以及端口(80、443、8000),只要有一個不同,那么就是不同源。比如下面都是不同的源:
http://localhost
https://localhost
http://localhost:8080
即使它們都是 localhost,但是它們使用了不同的協(xié)議或端口,所以它們是不同的源。假設你的前端運行在 localhost:8080,并且嘗試與 localhost:5555 進行通信;然后瀏覽器會向后端發(fā)送一個 HTTP OPTIONS 請求,后端會發(fā)送適當?shù)?headers 來對這個源進行授權(quán);所以后端必須有一個 "允許的源" 列表,如果前端對應的源是被允許的,瀏覽器才會允許前端向后端發(fā)請求,否則就會出現(xiàn)跨域失敗。
而默認情況下,前后端必須是在同一個源,如果不同源那么前端就會請求失敗。而前后端分離早已成為了主流,因此跨域問題是必須要解決的。
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import uvicornapp = FastAPI()
app.add_middleware(CORSMiddleware,# 允許跨域的源列表,例如 ["http://www.example.org"] 等等,["*"] 表示允許任何源allow_origins=["*"],# 跨域請求是否支持 cookie,默認是 False,如果為 True,allow_origins 必須為具體的源,不可以是 ["*"]allow_credentials=False,# 允許跨域請求的 HTTP 方法列表,默認是 ["GET"]allow_methods=["*"],# 允許跨域請求的 HTTP 請求頭列表,默認是 [],可以使用 ["*"] 表示允許所有的請求頭# 當然 Accept、Accept-Language、Content-Language 以及 Content-Type 總之被允許的allow_headers=["*"],# 可以被瀏覽器訪問的響應頭, 默認是 [],一般很少指定# expose_headers=["*"]# 設定瀏覽器緩存 CORS 響應的最長時間,單位是秒。默認為 600,一般也很少指定# max_age=1000
)if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
以上即可解決跨域問題。
使用CORSMiddleware
來配置跨域:
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddlewareapp = FastAPI()origins = ["http://localhost.tiangolo.com","https://localhost.tiangolo.com","http://localhost","http://localhost:8080",]app.add_middleware(CORSMiddleware,allow_origins=origins,allow_credentials=True,allow_methods=["*"],allow_headers=["*"],)@app.get("/")
async def main():return {"message": "Hello World"}
它支持以下參數(shù):
allow_origins
?- 一個允許跨域請求的源列表。例如?['https://example.org', 'https://www.example.org']
。你可以使用?['*']
?允許任何源。allow_origin_regex
?- 一個正則表達式字符串,匹配的源允許跨域請求。例如?'https://.*\.example\.org'
。allow_methods
?- 一個允許跨域請求的 HTTP 方法列表。默認為?['GET']
。你可以使用?['*']
?來允許所有標準方法。allow_headers
?- 一個允許跨域請求的 HTTP 請求頭列表。默認為?[]
。你可以使用?['*']
?允許所有的請求頭。Accept
、Accept-Language
、Content-Language
?以及?Content-Type
?請求頭總是允許 CORS 請求。allow_credentials
?- 指示跨域請求支持 cookies。默認是?False
。另外,允許憑證時?allow_origins
?不能設定為?['*']
,必須指定源。expose_headers
?- 指示可以被瀏覽器訪問的響應頭。默認為?[]
。max_age
?- 設定瀏覽器緩存 CORS 響應的最長時間,單位是秒。默認為?600
。
高階操作
看一些 FastAPI 的高階操作,這些操作有的不一定能用上,但用上了確實會方便許多。
其它的響應
返回 json 數(shù)據(jù)可以是:JSONResponse、UJSONResponse、ORJSONResponse,Content-Type 是 application/json;返回 html 是?HTMLResponse,Content-Type 是 text/html;返回?PlainTextResponse,Content-Type 是 text/plain。但是我們還可以有三種響應,分別是返回重定向、字節(jié)流、文件。
重定向
from fastapi import FastAPI
from fastapi.responses import RedirectResponse
import uvicornapp = FastAPI()@app.get("/index")
async def index():return RedirectResponse("https://www.bilibili.com")if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
頁面中訪問 /index 會跳轉(zhuǎn)到 bilibili。
字節(jié)流
返回字節(jié)流需要使用異步生成器的方式:
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import uvicornapp = FastAPI()async def some_video():for i in range(5):yield f"video {i} bytes ".encode("utf-8")@app.get("/index")
async def index():return StreamingResponse(some_video())if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
如果有文件對象,那么也是可以直接返回的。
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import uvicornapp = FastAPI()@app.get("/index")
async def index():return StreamingResponse(open("main.py", encoding="utf-8"))if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
文件
返回文件的話,還可以通過 FileResponse:
from fastapi import FastAPI
from fastapi.responses import FileResponse
import uvicornapp = FastAPI()@app.get("/index")
async def index():# filename 如果給出,它將包含在響應的 Content-Disposition 中。return FileResponse("main.py", filename="這不是main.py")if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
HTTP 驗證
如果當用戶訪問某個請求的時候,我們希望其輸入用戶名和密碼來確認身份的話該怎么做呢?
from fastapi import FastAPI, Depends
from fastapi.security import HTTPBasic, HTTPBasicCredentials
import uvicornapp = FastAPI()security = HTTPBasic()@app.get("/index")
async def index(credentials: HTTPBasicCredentials = Depends(security)):return {"username": credentials.username, "password": credentials.password}if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
測試
輸入完畢之后,信息會保存在?credentials,我們可以獲取出來進行驗證。
websocket
FastAPI 如何實現(xiàn) websocket:
from fastapi import FastAPI
from fastapi.websockets import WebSocket
import uvicornapp = FastAPI()@app.websocket("/ws")
async def ws(websocket: WebSocket):await websocket.accept()while True:# websocket.receive_bytes()# websocket.receive_json()data = await websocket.receive_text()await websocket.send_text(f"收到來自客戶端的回復: {data}")if __name__ == "__main__":uvicorn.run("main:app", host="0.0.0.0", port=5555)
然后我們通過瀏覽器進行通信:
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>Title</title>
</head>
<body><script>ws = new WebSocket("ws://localhost:5555/ws");//如果連接成功, 會打印下面這句話, 否則不會打印ws.onopen = function () {console.log('連接成功')};//接收數(shù)據(jù), 服務端有數(shù)據(jù)過來, 會執(zhí)行ws.onmessage = function (event) {console.log(event)};//服務端主動斷開連接, 會執(zhí)行.//客戶端主動斷開的話, 不執(zhí)行ws.onclose = function () { }</script>
</body>
</html>
測試
示例:
from fastapi import FastAPI, WebSocketfrom fastapi.responses import HTMLResponseapp = FastAPI()html = """
<!DOCTYPE html>
<html><head><title>Chat</title></head><body><h1>WebSocket Chat</h1><form action="" onsubmit="sendMessage(event)"><input type="text" id="messageText" autocomplete="off"/><button>Send</button></form><ul id='messages'></ul><script>var ws = new WebSocket("ws://localhost:8000/ws");ws.onmessage = function(event) {var messages = document.getElementById('messages')var message = document.createElement('li')var content = document.createTextNode(event.data)message.appendChild(content)messages.appendChild(message)};function sendMessage(event) {var input = document.getElementById("messageText")ws.send(input.value)input.value = ''event.preventDefault()}</script></body>
</html>
"""@app.get("/")
async def get():return HTMLResponse(html)@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):await websocket.accept()while True:data = await websocket.receive_text()await websocket.send_text(f"Message text was: {data}")
FastAPI 服務的部署
使用異步框架,最重要的是要搭配一個異步驅(qū)動去訪問數(shù)據(jù)庫,因為 web 服務的瓶頸都是在數(shù)據(jù)庫上面。
上面介紹了 FastAPI 的絕大部分內(nèi)容,然后我們來看看 FastAPI 服務的部署,其實部署很簡單,直接 uvicorn.run 即可。但是這里面有很多的參數(shù),我們主要是想要介紹這些參數(shù)。
def run(app, **kwargs):config = Config(app, **kwargs)server = Server(config=config)......
看到 app 和 **kwargs 都傳遞給了 Config,所以我們只需要看 Config 里面都有哪些參數(shù)即可。這里選出一部分:
app:第一個參數(shù),不需要解釋
host:監(jiān)聽的ip
port:監(jiān)聽的端口
uds:綁定的 unix domain socket,一般不用
fd:從指定的文件描述符中綁定 socket
loop:事件循環(huán)實現(xiàn),可選項為 auto|asyncio|uvloop|iocp
http:HTTP 協(xié)議實現(xiàn),可選項為 auto|h11|httptools
ws:websocket 協(xié)議實現(xiàn),可選項為 auto|none|websockets|wsproto
lifespan:lifespan 實現(xiàn),可選項為 auto|on|off
env_file:環(huán)境變量配置文件
log_config:日志配置文件
log_level:日志等級
access_log:是否記錄日志
use_colors:是否帶顏色輸出日志信息
interface:應用接口,可選 auto|asgi3|asgi2|wsgi
debug:是否開啟 debug 模式
reload:是否自動重啟
reload_dirs:要自動重啟的目錄
reload_delay:多少秒后自動重啟
workers:工作進程數(shù)
limit_concurrency:并發(fā)的最大數(shù)量
limit_max_requests:能 hold 住的最大請求數(shù)
3、示例:fastapi?開發(fā)接口 (?只是api接口,不帶web渲染?)
示例:豆瓣電影top250
import requests
from scrapy.http import HtmlResponse
import uvicorn
from pathlib import Path
from fastapi import FastAPIapp = FastAPI()headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ""(KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36 Edg/111.0.1661.62",
}def douban_movie_top250(page_num: int):index = (page_num - 1) * 25# https://movie.douban.com/top250?start=50&filter=url = f"https://movie.douban.com/top250?start={index}&filter="__resp = requests.get(url, headers=headers)if 200 == __resp.status_code:resp = HtmlResponse(url, body=__resp.content, encoding='utf-8')movie_name_list = resp.xpath('//div[@class="item"]//div[@class="hd"]/a/span[1]/text()').extract()movie_url_list = resp.xpath('//div[@class="item"]//div[@class="hd"]/a/@href').extract()movie_info_list = list(zip(movie_name_list, movie_url_list))return movie_info_listelse:return {'請求失敗': f" status_code ---> {__resp.status_code}"}@app.get("/douban/movie_top250")
async def get_item(page_num):"""和 Flask 不同,Flask 是使用 <>,而 FastAPI 使用 {}"""try:page_num_int = int(page_num)except BaseException as be:return {"錯誤信息": "頁碼必須是數(shù)字"}data = douban_movie_top250(page_num_int)return {"data": data}if __name__ == '__main__':'''http://127.0.0.1:5555/douban/movie_top250?page_num=1'''print(f'{Path(__file__).stem}:app')uvicorn.run(f'{Path(__file__).stem}:app', host="0.0.0.0", port=5555)pass
訪問:http://127.0.0.1:5555/douban/movie_top250?page_num=5
示例:B 站視頻 字幕 獲取
import uvicorn
from pathlib import Path
from fastapi import FastAPI
import requests
import json
from scrapy.http import HtmlResponseapp = FastAPI()headers = {'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6','cookie': 'buvid3=31D606A5-C08F-CF7F-6345-DFE18CDF1FCA25535infoc; b_nut=1681180325; CURRENT_FNVAL=4048; _uuid=FA2D2910A-1F4C-9ED10-24FB-710C228FE726630982infoc; buvid_fp=dd9cac90362a92030f254a522e274486; buvid4=F13E840C-6245-40B7-9B63-5FDD1992821229542-023041110-0bxmZkxc4Ip6QXGeEfs0Og%3D%3D; CURRENT_PID=130742c0-d811-11ed-ac38-37fb01852f74; rpdid=|(JYYkY~RRYu0J\'uY)uk~lJY|; i-wanna-go-back=-1; header_theme_version=CLOSE; home_feed_column=5; is-2022-channel=1; nostalgia_conf=-1; DedeUserID=384760568; DedeUserID__ckMd5=8fd50449771672ee; b_ut=5; FEED_LIVE_VERSION=V_NO_BANNER_1; bsource=search_bing; browser_resolution=1863-969; bp_video_offset_384760568=787947892852654100; b_lsid=C105138DE_187B25E90A6; SESSDATA=d3f7b6a0%2C1697876752%2C413d0%2A42; bili_jct=e41d9dfdbd372b0cb95222cfa0d33199; sid=59e50ddx; innersign=1; PVID=1; innersign=1'
}def get_subtitle(video_id=None):if video_id:url = f'https://www.bilibili.com/video/{video_id}/'resp = requests.get(url, headers=headers)if 200 == resp.status_code:scrapy_resp = HtmlResponse(url, body=resp.content, encoding='utf-8')try:temp = scrapy_resp.css('html').re('"subtitle_url":"(.*?)"')[0]except BaseException as be:return {'請求失敗': str(be)}subtitle_url = temp.replace(r'\u002F', '/')print(subtitle_url)r = requests.get(subtitle_url)if 200 == r.status_code:return r.json()else:return {'請求失敗, 失敗狀態(tài)碼': resp.status_code}else:return {'請求失敗, 失敗狀態(tài)碼': resp.status_code}else:return {"請求失敗": '視頻 id 錯誤'}@app.get("/bilibili/{video_id}")
async def get_item(video_id):"""和 Flask 不同,Flask 是使用 <>,而 FastAPI 使用 {}"""data = get_subtitle(video_id)return {"data": data}if __name__ == '__main__':'''http://127.0.0.1:5555/bilibili/BV1bW411n7fY'''print(f'{Path(__file__).stem}:app')uvicorn.run(f'{Path(__file__).stem}:app', host="0.0.0.0", port=5555)pass
4、示例:fastapi?開發(fā)?web?渲染?網(wǎng)站
FastAPI 這個 Python Web 框架并沒有帶 "渲染網(wǎng)頁的模板引擎",但是也正因為如此,它可以使用任何網(wǎng)頁模板。官方例子是 jinjia2 。
模板是全棧 Web 開發(fā)的重要組成部分。使用 Jinja,您可以構(gòu)建豐富的模板,為?Python Web 應用程序的前端提供支持。
Jinja 是一個用 Python 編寫的模板引擎,旨在幫助 API 響應的渲染過程。在每種模板語言中,都有變量被替換為實際傳遞給它們的值,當模板被渲染時,有控制模板邏輯的標簽。
安裝 jinja2
安裝:pip install jinja2 aiofiles
Jinja 模板只是一個文本文件。 Jinja 可以生成任何基于文本的格式(HTML、XML、CSV、LaTeX 等)。 Jinja 模板不需要有特定的擴展名:.html、.xml 或任何其他擴展名都可以。
關(guān)于模版的擴展名:任何文件都可以作為模板加載,無論文件擴展名如何。添加 .jinja 擴展名,如 user.html.jinja 可能會使某些 IDE 或編輯器插件更容易,但這不是必需的。自動轉(zhuǎn)義可以基于文件擴展名應用,因此在這種情況下您需要考慮額外的后綴。
識別模板的另一個很好的啟發(fā)式方法是它們位于模板 templates 文件夾中,而不管擴展名是什么。這是項目的常見布局。
Jinja 模板引擎使用花括號 {} 來區(qū)分其表達式和語法,以及與常規(guī) HTML、文本和模板文件中的任何其他變量。{{}} 語法稱為變量塊。{% %} 語法包含控制結(jié)構(gòu),如 if/else 、循環(huán)和宏。Jinja 模板語言中使用的三種常見語法塊包括以下內(nèi)容:
- {% ... %}:這種語法用于控制結(jié)構(gòu)等語句。
- {{ todo.item }}:這個語法用于打印出傳遞給它的表達式的值。
- {# Test #}: 這種語法在寫評論時使用,不在網(wǎng)頁上顯示。
Jinja2 是一種流行的模板語言,被 Flask、Bottle、Pelican 使用,也可被 Django 使用。
渲染第一個 Jinja 模板
代碼:
import jinja2
environment = jinja2.Environment()
template = environment.from_string("Hello, {{ name }}!")
result = template.render(name="渲染第一個jinja2模板")
print(result)
Jinja 的核心組件是 Environment() 類。在此示例中,創(chuàng)建了一個不帶任何參數(shù)的 Jinja 環(huán)境。然后通過environment.from_string?來自定義環(huán)境。這里是創(chuàng)建一個普通環(huán)境,并在其中加載字符串 Hello, {{ name }}! 作為模板。
這個例子顯示了在使用 Jinja 時通常會執(zhí)行的兩個重要步驟:
- 加載模板:加載包含占位符變量的源。默認情況下,它們包含在一對大括號 {{ }} 中。
- 渲染模板:用內(nèi)容填充占位符。您可以提供字典或關(guān)鍵字參數(shù)作為上下文。
執(zhí)行結(jié)果如下:
使用外部文件作為模板
與上述方式同理,我們可以使用外部文件作為我們的模版來源,在我們的項目中創(chuàng)建一個新文件夾。在工作目錄中,創(chuàng)建一個名為 templates/ 的文件夾。
然后,您可以在 template?目錄中創(chuàng)建 index.html 模板文件,并使用 Jinja2 語法來呈現(xiàn)它們。例如,在template/index.html 中寫入如下內(nèi)容:
<!DOCTYPE html>
<html>
?<head>
?<title>Welcome</title>
?<link href="{{ url_for('static', path='/styles.css') }}" rel="stylesheet">
?</head>
?<body>
?<h1>Hello, {{ name }}</h1>
?</body>
</html>
回到?main.py 中:
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templatesapp = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")@app.get("/{name}")
async def home(request: Request, name: str):return templates.TemplateResponse("index.html", {"request": request,"name": name})if __name__ == '__main__':pass
整個文件的目錄結(jié)構(gòu)如下:
啟動 FastAPI 服務:uvicorn main:app --reload --port 8888
然后另外打開一個終端,執(zhí)行 curl 127.0.0.1:8888/Yuzhou1su 命令,可以看到如下 name 被渲染出來的結(jié)果:
通過瀏覽器訪問這個 http://127.0.0.1:8888/Yuzhou1su 就能看到 css 渲染的顏色:
Jinja 模板變量可以是任何 Python 類型或?qū)ο?#xff0c;只要它們可以轉(zhuǎn)換為字符串??梢詫⒛P?、列表或字典類型傳遞到模板中,并通過將這些屬性放置在先前列出的第二個塊中來顯示其屬性。在下一節(jié)中,我們將看一下過濾器。過濾器是每個模板引擎的重要組成部分,在 Jinja 中,過濾器使我們能夠執(zhí)行某些函數(shù),例如從列表中連接值和檢索對象的長度,等等。Jinja 中常用的功能:變量、過濾器、if 語句、循環(huán)、宏和模板繼承。
變量
模板變量由傳遞給模板的上下文字典定義。
在模板中,只要應用程序傳遞了變量,您就可以隨意操作這些變量。變量可能還具有您可以訪問的屬性或元素。變量具有哪些屬性取決于提供該變量的應用程序。
除了標準的 Python __getitem__ “下標”語法( [] )之外,您還可以使用點(. )來訪問變量的屬性。
以下行執(zhí)行相同的操作:
{{ foo.bar }}
{{ foo['bar'] }}
Filters
盡管 Python 和 Jinja 的語法非常相似,但是像連接字符串、將字符串的第一個字符設置為大寫等修改操作不能使用Python 的語法在 Jinja 中完成。因此,為了執(zhí)行這樣的修改操作,我們在 Jinja 中使用過濾器。
變量可以被過濾器修改。過濾器與變量用管道符號(|)分隔,并且可以在括號中包含可選參數(shù)??梢枣溄佣鄠€過濾器。一個過濾器的輸出應用于下一個。過濾器的定義格式如下:
{{ variable | filter_name(*args) }}
不加參數(shù)的過濾器:
{{ variable | filter_name }}
{{ name|striptags|title }}
default 過濾器: 如果該值未定義,它將返回傳遞的默認值,否則返回變量的值:
{{ my_variable | default('my_variable is not defined') }}
escape 過濾器: 這個過濾器用于渲染原始 HTML 輸出:將字符串 s 中的字符 & < > ' ” 轉(zhuǎn)換為 HTML 安全序列。如果您需要在 HTML 中顯示可能包含此類字符的文本,請使用此選項。將返回值標記為標記字符串。
{{ "<title>Todo Application</title>" | escape }}
<title>Todo Application</title>
類型轉(zhuǎn)換過濾器: 這些過濾器包括 int 和 float 過濾器,用于從一種數(shù)據(jù)類型轉(zhuǎn)換到另一種數(shù)據(jù)類型:
{{ 3.142 | int }}
3
{{ 20 | float }}
20.0
join 過濾器:join(*value*, *d=u''* , *attribute=None*)返回一個字符串,它是序列中字符串的串聯(lián)。元素之間的分隔符默認為空字符串,您可以使用可選參數(shù)定義它:
{{ [1, 2, 3] | join('|') }}-> 1|2|3
{{ [1, 2, 3] | join }}-> 123
也可以連接對象的某些屬性:
{{ users|join(', ', attribute='username') }}
長度 filter: 這個過濾器返回一個序列或集合的長度,它的作用與 Python 中 len() 函數(shù)的作用相同:
Todo count: {{ todos | length }}
Todo count: 4
if 條件
Jinja 中 if 語句的用法與 Python 中的用法類似。在 {% %} 控制塊中使用。讓我們看一個例子:
{% if todos %}
<ul>
{% for todo in todos %}<li>{{ todo.name|e }}</li>
{% endfor %}
</ul>
{% endif %}
Loop 條件
我們也可以在Jinja中對變量進行迭代。這可以是一個列表或一個一般的函數(shù)、 比如說下面這個,例如
{% for todo in todos %}<li>{{ todo.name|e }}</li>
{% endfor %}
你可以在 for 循環(huán)中訪問特殊的變量,比如 loop.index ,它給出了當前迭代的索引。
宏
宏可與常規(guī)編程語言中的函數(shù)相媲美。它們有助于將常用的習語放入可重用的函數(shù)中,以免重復自己(“DRY” 原則)。
{% macro input(name, value='', type='text', size=20) %}<div class="form"><input type="{{ type }}" name="{{ name }}"value="{{ value|escape }}" size="{{ size }}"></div>
{% endmacro %}
現(xiàn)在,為了在你的表單中快速創(chuàng)建一個輸入,調(diào)用了這個宏:
{{ input('item') }}
渲染完成后,將會返回:
<div class="form"><input type="text" name="item" value="" size="20" /></div>
FastAPI 中的 Jinja
FastAPI 實際上是為構(gòu)建 API 和微服務而設計的。它可用于構(gòu)建使用 Jinja 提供 HTML 服務的 Web 應用程序,但這并不是它真正優(yōu)化的目的。
如果您想構(gòu)建一個在服務器上呈現(xiàn)大量 HTML 的大型網(wǎng)站,Django 可能是更好的選擇。
但是,如果您正在使用 React、Angular 或 Vue 等前端框架構(gòu)建現(xiàn)代網(wǎng)站,那么從 FastAPI 獲取數(shù)據(jù)是一個不錯的選擇。
5、FastAPI、vue?開發(fā)?web 網(wǎng)站
5.1 關(guān)于 FastAPI 與 Vue3 的通信
:https://zhuanlan.zhihu.com/p/632387477
基于Vue3和FastAPI對數(shù)據(jù)庫進行操作:https://zhuanlan.zhihu.com/p/632393099
5.2 連接 Vue.js 作為前端,Fastapi 作為后端
:https://www.cnblogs.com/hahaha111122222/p/15904405.html
目錄結(jié)構(gòu)
├── main.py
└── templates
? ? └── home.html
后端?fastapi
pip install fastapi[all]
pip install jinja2
main.py
- 我們在 / 中服務于我們的前端,并在該路徑中呈現(xiàn)我們的home.html。
- 我們使用templates文件夾保存我們的HTML并將其傳遞給Jinja。
- 另外,我們將從我們的front-end向/add發(fā)送一個請求。
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from pydantic import BaseModeltemplates = Jinja2Templates(directory="templates") app = FastAPI()class TextArea(BaseModel):content: str@app.post("/add")
async def post_textarea(data: TextArea):print(data.dict())return {**data.dict()}@app.get("/")
async def serve_home(request: Request):return templates.TemplateResponse("home.html", {"request": request})
前端 - home.html
- 讓我們創(chuàng)建一個有文本區(qū)域和按鈕的虛擬應用程序。
- 我們正在使用Axios將請求發(fā)送到后端。
- 因為它們在同一個端口上運行,所以我們可以直接將/add傳遞給Axios。
<html>
<title></title>
<script src="https://cdn.jsdelivr.net/npm/vue/dist/vue.js"></script>
<script src="https://unpkg.com/axios/dist/axios.min.js"></script><body><div id="app"><textarea name="" id="content" cols="30" rows="10" v-model="content"></textarea><button @click="addText" id="add-textarea">click me</button></div><script>new Vue({el: "#app",data: {title: '',content: ''},methods: {addText() {return axios.post("/add", {content: this.content}, {headers: {'Content-type': 'application/json',}}).then((response) => {console.log("content: " + this.content);});}}});</script>
</body></html>
運行,訪問測試
命令:uvicorn main:app --reload
最后,你會有一個可怕的文本區(qū)和一個按鈕。但它會幫助你更好地理解事情。
5.3?FastApi+Vue+LayUI實現(xiàn)簡單的前后端分離 demo
實際使用中,通常建議前后端項目分離。下面使用FastApi+Vue+LayUI做一個前后端分離的Demo。
后端
后端采用 FastApi,代碼 test.py
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from fastapi.responses import JSONResponse
from pathlib import Path
import uvicorn
import subprocessapp = FastAPI()
templates = Jinja2Templates(directory="templates")@app.get('/info')
async def user_list():# vue的響應數(shù)據(jù)ret_list = [{'id': '1', 'value': 'one'},{'id': '2', 'value': 'two'},{'id': '3', 'value': 'three'},]return JSONResponse(content=ret_list)@app.get("/check")
async def home(request: Request):return templates.TemplateResponse("index.html", {"request": request})if __name__ == '__main__':'''http://127.0.0.1:5555/check'''# print(f'{Path(__file__).stem}:app')uvicorn.run(f'{Path(__file__).stem}:app', host="0.0.0.0", port=5555)pass
前端
前端直接導入Vue、LayUI、Axios 的 JS 和 CSS 的 CDN 資源,在 Vue 實例的 mount 階段,使用axios 調(diào)用后端接口拿到數(shù)據(jù),使用 LayUI 的樣式對 table 元素進行美化。
代碼
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><meta http-equiv="X-UA-Compatible" content="IE=edge"><meta name="viewport" content="width=device-width, initial-scale=1.0"><script src="https://unpkg.com/vue@3/dist/vue.global.js"></script><script src="https://unpkg.com/axios/dist/axios.min.js"></script><!-- 引入 layui.css --><link rel="stylesheet" href="https://www.layuicdn.com/layui/css/layui.css"/><!-- 引入 layui.js --><script src="https://www.layuicdn.com/layui/layui.js" type="text/javascript" charset="utf-8"></script><title>Home</title>
</head>
<body>
<div id="app"><table class="layui-table"><tr v-for="fo in info_list"><td> [[ fo.id ]] </td><td> [[ fo.value ]] </td></tr></table>
</div>
<table id="test" class="layui-table"></table><script>const {createApp, ref} = Vueconst vue_app = createApp({data() {return {info_list: [{id: 1, name: '默認值'}],// info: "hello vue..."}},mounted() {this.showinfo();},methods: {showinfo() {axios.get('/info').then(response => {this.info_list = response.data;// console.log(response);console.log(`this.info_list ---> ${this.info_list.toString()}`);}, err => {console.log(err);})}}});// vue_app.config.delimiters = ['[[', ']]'];vue_app.config.compilerOptions.delimiters = ['[[', ']]']vue_app.mount('#app');
</script>
</body>
</html>
vue?和?jinja2?默認都使用 "{{內(nèi)容}}"?在前端進行顯示變量的值,所以會造成沖突。
可以修改 vue?顯示值得方式,即修改 "插值符":
vue 2?方式:
<script>
? ? const vue = new Vue({
? ? ? ? el:"#app",
? ? ? ? delimiters: ["[[", "]]"],
? ? ? ? data:{
? ? ? ? ? ? selects:['enu','cha'],
? ? ? ? ? ? userData:[]
? ? ? ? }
</script>
vue 3?方式:
在 Vue 3 中,默認的差值符號是雙大括號({{ }})用于渲染動態(tài)數(shù)據(jù)到模板中。然而,如果你希望修改默認的差值符號,Vue 3 提供了一種簡單的方式來實現(xiàn)。
可以在創(chuàng)建 Vue 應用程序?qū)嵗?#xff0c;使用 createApp 函數(shù)的 config 方法來配置全局的差值符號:
import { createApp } from 'vue';
const app = createApp({});
//app.config.delimiters = ['${', '}'];app.config.compilerOptions.delimiters = ['${', '}'];
app.mount('#app');
上述代碼中,我們通過 app.config.delimiters 來修改差值符號為 ${ }。
修改之后,你可以在模板中使用新的差值符號來顯示動態(tài)數(shù)據(jù):
<template>
? <div>
? ? <p>${ message }</p>
? </div>
</template><script>
export default {
? data() {
? ? return {
? ? ? message: "Hello, world!"
? ? };
? }
};
</script>
在上述示例中,我們使用 ${ } 差值符號來顯示 message 數(shù)據(jù)。
需要注意的是,修改差值符號后,你需要確保新的差值符號與模板中的變量名不會發(fā)生沖突。同時,修改差值符號只在當前應用程序?qū)嵗秶鷥?nèi)有效,不會影響其他應用程序?qū)嵗?/p>
運行項目
啟動 FastApi 后端服務器,訪問 /test/check 接口。
使用 axios 發(fā)送 get 和 post 請求詳解
:https://blog.csdn.net/grand_brol/article/details/108167088
Q&A
Q:為什么在請求/info
接口總會出現(xiàn)一個Temporary Redirect
重定向呢?
A:原因是因為我們在 FastApi
接口定義的時候,uri
的格式不規(guī)范導致,uri
的結(jié)尾不需要/
,如果你接口增加了/
,我們使用瀏覽器訪問 uri
,瀏覽器會忽略結(jié)尾的/
,FastApi
會在內(nèi)部進行查重定向,將瀏覽器不帶/
的請求重定向到我們定義的帶/
的視圖函數(shù)上。
5.4 使用 python fastapi+vue 快速搭建網(wǎng)站
:https://www.elprup.com/2020/09/19/fastapi_vue/
傳統(tǒng)網(wǎng)站由一個 web 框架完全承擔,例如基于 nodejs 的 express,koa,基于 python 的 django,tornado。新型網(wǎng)站演變?yōu)橛?vue 開發(fā)前端系統(tǒng),使用 api 框架開發(fā)后端 api 請求接口的模式。
5.5 用 FastAPI 和 Vue.js 開發(fā)一個單頁應用程序
原文地址:https://testdriven.io/blog/developing-a-single-page-app-with-fastapi-and-vuejs/
源碼地址:https://github.com/testdrivenio/fastapi-vue
翻譯1:https://juejin.cn/post/7113790977848360967
翻譯2:https://www.cnblogs.com/leimu/p/16992966.html