FastApi持續更新

本文轉載自查看原文 2021-07-04 21:09 778 FastApi

FastAPI 框架，高性能，易於學習，高效編碼，生產可用
官方文檔： https://fastapi.tiangolo.com
FastAPI 是一個用於構建 API 的現convert_underscores代、快速（高性能）的 web 框架，使用 Python 3.6+ 並基於標准的 Python 類型提示。
關鍵特性:

快速：可與 NodeJS 和 Go 比肩的極高性能（歸功於 Starlette 和 Pydantic）。最快的 Python web 框架之一。
高效編碼：提高功能開發速度約 200％至 300％。
更少 bug：減少約 40％的人為（開發者）導致錯誤。
智能：極佳的編輯器支持。處處皆可自動補全，減少調試時間。
簡單：設計的易於使用和學習，閱讀文檔的時間更短。
簡短：使代碼重復最小化。通過不同的參數聲明實現豐富功能。bug 更少。
健壯：生產可用級別的代碼。還有自動生成的交互式文檔。
標准化：基於（並完全兼容）API 的相關開放標准。

安裝FastApi：

pip3 install fastapi
pip3 install unicorn

開始第一個Demo

# 創建一個main.py文件

from typing import Optional

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello FastApi"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None):
    return {"item_id": item_id, "q": q}

運行服務器：

一、命令運行服務器：

uvicorn main:app --reload --port 8888 # 更改端口號

二、代碼運行服務器調試：

if __name__ == '__main__':
    import uvicorn

    uvicorn.run("main:app", reload=True, port=5555)

FastApi提供交互式Api文檔一：這很方便我們管理自己的接口

現在訪問 http://localhost:8000/docs 就會生成一個Swagger文檔

FastApi提供交互式Api文檔二：這很方便我們管理自己的接口

現在訪問 http://127.0.0.1:8000/redoc 就會生成一個redoc文檔

FastApi中何時用Path、Query、Header、Body

Query:查詢參數用
Path:路徑參數
Body:需要在請求數據傳入countent_type為json的
Form:請求數據為表單的

如果你想即「get」又「post」同時請求，你可以這么做：

@app.api_route("/login", methods=("GET", "POST", "PUT"))
def login():
    """這是一個登陸接口"""
    return {"msg": "login success", "code": 200}

何時用「Form」，何時用「Body」，何時用「Header」呢

如果你你以表單的形式傳遞數據，那你就應該用「Form」，看一下代碼

@app.post("/login1")
def login_form(username=Form(None), password=Form(None)):
    return {"username": username, "password": password}

如果你你以JSON的形式傳遞數據，那你就應該用「Body」，看一下代碼

@app.post("/login")
def login(data=Body(None)):
    return {"data": data}

如果你你想傳遞「Header」數據，那你就應該用「Header」，看一下代碼

@app.get("/user")
def user(id, num=Header(None)):
    return {"id": id, "num": num}

如何定制一個返回信息，看代碼

作用：就是將自己定義好的響應結果返回回來

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()


@app.get("/user")
def user():
    return JSONResponse(content={"msg": "get user"},   # 內容
                        status_code=202,  # 狀態碼，默認為200
                        headers={"a": "b"})


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("04_response:app", reload=True)

如何將自己寫的html網站動態加載到fastapi（jinja2模板返回Html頁面）

我們需要安裝一些依賴庫

1、jinja2
pip install jinjia2
2、aiofiles
pip install aiofiles

廢話不多說，看代碼

from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates   # 需要進入的庫
from starlette.staticfiles import StaticFiles  # 引入靜態文件庫


app = FastAPI()
# 指定靜態文件存放路徑
app.mount("/page", StaticFiles(directory="page"), name="page")
# 指定html 存放目錄
template = Jinja2Templates("page")


@app.get("/home")
def home (req: Request):
    return template.TemplateResponse("index.html", context={"request": req})


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("d05_templates:app", reload=True)

如果你的代碼中有引入到css樣式，你就可以向我這樣，你會發現樣式就被引入進來了

如果你想自定義傳參進來，你可以試試這樣：

如果你想實現這樣的例子

Kapture 2021-07-04 at 11.28.05.gif

# main.py

from fastapi import FastAPI, Request, Form
from fastapi.responses import RedirectResponse
from fastapi.templating import Jinja2Templates

app = FastAPI()
template = Jinja2Templates("pages")
todos = ["寫博客", "看電影", "玩游戲"]


@app.get("/")
async def index(req: Request):
    return template.TemplateResponse("index.html", context={"request": req, "todos": todos})


@app.post("/todo")
async def todo(todo=Form(None)):
    """處理用戶發送過來的 todolist 數據"""
    todos.insert(0, todo)
    return RedirectResponse('/', status_code=302)


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("main:app", reload=True)

在這里我說一下為什么要將狀態碼設置為302，如果你不設置這個status_code，瀏覽器發送給后端的請求狀態碼為307，因為307的狀態碼是不能從post請求跳轉到get請求，原因是post請求如果要跳轉到get請求不通用，如果想進行跳轉，需要將307更改為302。

# index.html    <meta charset="UTF-8">    <title>Title</title><h1>待辦事項</h1><div>    <form action="/todo" method="post">        <input name="todo" type="text" placeholder="請添加待辦事件...">        <input type="submit" value="添加">    </form></div>    {% for todo in todos %}    <p>{{ todo }}</p>    {% endfor %}

關聯數據庫，將數據存儲化

第一步：我們需要安裝依賴庫

pip install tortoise-orm
pip install aiomysq

第二步：電腦需要安裝mysql，安裝調試過程不在贅述

以我為例：先創建一個db為fastapi的庫

create database fastapi;

第三步：配置數據庫

from tortoise.contrib.fastapi import register_tortoise

# 數據庫綁定
register_tortoise(app,
                  db_url="mysql://root:Ranyong_520@localhost:3306/fastapi",
                  modules={"models": []},
                  add_exception_handlers=True,
                  generate_schemas=True)

實例一：將數據存儲到數據庫並返回給前端

# d06_templates.py
from fastapi import FastAPI, Request, Form
from fastapi.responses import RedirectResponse
from fastapi.templating import Jinja2Templates
from tortoise.contrib.fastapi import register_tortoise

from dao.models import Todo

app = FastAPI()
template = Jinja2Templates("pages")

# 數據庫綁定
register_tortoise(app,
                  db_url="mysql://root:Ranyong_520@localhost:3306/fastapi",
                  modules={"models": ['dao.models']},  # 設置模型類
                  add_exception_handlers=True,
                  generate_schemas=True)


@app.get("/")
async def index(req: Request):
    # 從數據庫獲取 todos 的代碼
    # ORM,獲取所有的 todos
    todos = await Todo.all()  # 獲取所有的todos
    print(todos)
    return template.TemplateResponse("index.html", context={"request": req, "todos": todos})


@app.post("/todo")
async def todo(content=Form(None)):
    """處理用戶發送過來的 todolist 數據"""
    await Todo(content=content).save()
    return RedirectResponse('/', status_code=302)


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("d06_templates:app", reload=True)

然后創建一個dao的文件夾里面創建一個models的py文件

from tortoise import Model, fieldsclass Todo(Model):    """數據庫當中的表 todo"""    id = fields.IntField(pk=True)  # id為int類型的 pk:是將id作為主鍵    content = fields.CharField(max_length=500)  # todo項里面的內容    例如:todos = ["寫博客", "看電影", "玩游戲"]    created_at = fields.DatetimeField(auto_now_add=True)  # auto_now_add 當每次插入新數據的時候會引用本地時間    updated_at = fields.DatetimeField(auto_now=True)  # auto_now 當每次修改數據后會更新本地時間

這時候我們來運行下代碼：
Kapture 2021-07-04 at 16.36.46.gif
可以發現返回了，並沒有返回添加的數據，那我們再去數據庫看,點擊數據庫更新按鈕后，可以發現我們的數據已經存儲到了表中表。
Kapture 2021-07-04 at 16.39.05.gif

可以看到數據庫已經存了我們提交的數據，現在我們只需要改一下index.html文件一個地方就可以解決

最后效果
Kapture 2021-07-04 at 16.43.39.gif

枚舉(Enum)

在python3.4以后才可使用

第一步：需要倒入Enum庫

from enum import Enum

第二步：創建一個Enum的子類

class ModelName(str, Enum):
    A = 'a'
    B = 'b'
    C = 'c'

# 預設值.py
from fastapi import FastAPI                                          
                                                                     
from enum import Enum                                                
                                                                     
                                                                     
# 預設值                                                                
# 如果你有一個接收路徑參數的路徑操作，但是你希望預先設定可能的有效參數值                                
                                                                     
# 創建一個繼承str和Enum的子類                                                  
class ModelName(str, Enum):                                          
    A = 'a'                                                          
    B = 'b'                                                          
    C = 'c'                                                          
                                                                     
                                                                     
app = FastAPI()                                                      
                                                                     
                                                                     
@app.get("/models/{model_name}")                                     
async def get_model(model_name: ModelName):                          
    if model_name == ModelName.A:                                    
        """你可以用類里面ModelName.key來進行比較"""                              
        return {"model_name": model_name, "message": "You chose A!"} 
                                                                     
    if model_name.value == 'b':                                      
        """你也可以用model_name.value來進行比較"""                             
        return {"model_name": model_name, "message": "You chose B"}  
                                                                     
    return {"model_name": model_name, "message": "You chose C"}      
                                                                     
                                                                     
if __name__ == '__main__':                                           
    import uvicorn                                                   
                                                                     
    uvicorn.run("預設值:app", reload=True)

查詢參數

默認值：

顧名思義，默認值就是已經寫好了，你不需要在寫任何參數，不過你寫也不妨礙

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):    """    skip默認值為0    limit默認值為10    """    return fake_items_db[skip:skip + limit]# 傳入 http://127.0.0.1:8000/items/ === http://127.0.0.1:8000/items/?skip=0&limit=10# 返回值："""[{"item_name": "Foo"},{"item_name": "Bar"    },{"item_name": "Baz"}]"""if __name__ == '__main__':    import uvicorn    uvicorn.run("查詢參數:app", reload=True)

可選參數

第一步：導入依賴庫

from typing import Optional

第二步：寫方法引用

from typing import Optional

from fastapi import FastAPI

app = FastAPI()

# 選傳參數
@app.get("/items/{item_id}")
async def read_item(item_id: str, q: Optional[str] = None, short: bool = False):
    """
    :param item_id:
    :param q: 函數q是可選的，默認值為None
    :param short:
    :return:
    """
    item = {"item_id": item_id}
    if q:
        return {"item_id": item_id, "q": q}
    if not short:
        item.update(
            {"description": "這是一個令人驚異的項目，有一個很長的描述"}
        )
    return item

# 傳入http://127.0.0.1:8000/items/1?short=false
# 返回值：
"""
{
 "item_id": "1",
  "description": "這是一個令人驚異的項目，有一個很長的描述"
}
"""
# 傳入 http://127.0.0.1:8000/items/1?short=true
# 返回值：
"""
{
  "item_id": "1"
}
"""

if __name__ == '__main__':
    import uvicorn

    uvicorn.run("查詢參數:app", reload=True)

必傳參數

from fastapi import FastAPIapp = FastAPI()# 必傳參數@app.get("/items1/{item_id}")async def read_user_item(item_id: str, needy: str):    """    :param item_id: 必傳    :param needy: 必傳    :return:    """    item = {"item_id": item_id, "needy": needy}    return item# 傳入 http://127.0.0.1:8000/items1/1?needy=1# 返回值："""{  "item_id": "1",  "needy": "1"}"""if __name__ == '__main__':    import uvicorn    uvicorn.run("查詢參數:app", reload=True)

例子：

from fastapi import FastAPIapp = FastAPI()@app.get("/item2/{item_id}")async def read_user_item(        item_id: str,     	needy: str,     	skip: int = 0,     	limit: Optional[int] = None):    item = {"item_id": item_id, "needy": needy, "skip": skip, "limit": limit}    return item# 傳入：http://127.0.0.1:8000/item2/1?needy=2&skip=0&limit=1# 返回值："""{  "item_id": "1",  "needy": "2",  "skip": 0,  "limit": 1}"""if __name__ == '__main__':    import uvicorn    uvicorn.run("查詢參數:app", reload=True)

請求體

什么是請求體？就是將數據從客戶端發送api數據，這部分就叫請求體
請求體是客戶端發送api的數據，響應是api發送給客戶端的數據
api幾乎總是要發送響應體，但是客戶端並不總是需要發送請求體
⚠️：發送請求體不能用GET，發送數據我們通常使用POST，PUT，DELETE或PATH

第一步：導入Pydantic中的BaseModel

from pydantic import BaseModel

第二步：創建數據模型，將你的數據模型聲明為繼承自BaseModel的類

class Itme(BaseModel):    name: str    description: Optional[str] = None   # Optional 參數為可選的    price: float    tax: Optional[float] = None

完整代碼

from typing import Optional   
from fastapi import FastAPI
from pydantic import BaseModel   # 第一步導包


# 創建數據模型
class Itme(BaseModel):     # 第二步創建數據模型
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None


app = FastAPI()


# 使用模型
# 在函數內部，可以直接訪問模型對象的所有屬性
@app.post("/items/")
async def create_item(item: Itme):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.proice + item.tax
        item_dict.updata({"Proice_with_tax": price_with_tax})
    return item_dict



# 請求體+路徑參數
@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Itme):
    return {"item_id": item_id, **item.dict()}


# 請求體+路徑參數+查詢參數
@app.put("/items1/{item_id}")
async def create_item(item_id: int, item: Itme, q: Optional[str] = None):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("請求體:app", reload=True)

如果在路徑中也聲明了該參數，它將被作用路徑參數
如果參數屬於單一參數類似(int、float、str、bool等) 它將被解釋為查詢參數
如果參數的類型被聲明為一個Pydantic 模型，它將被解釋為請求體

查詢參數和字符串校驗

"""
在FastApi中提供了數據校驗的功能
"""

舉個例子：

from typing import Optionalfrom fastapi import FastAPIapp = FastAPI()@app.get("/items/")async def read_items(q: Optional[str] = None):    """    查詢參數q的類型為str，可以看到默認值為None，因為它是可選的(Optional)    但是我們需要添加約束條件，即使q是可選的，我還想限制它不能超過50個字符串，下面可以看《第一步》    """    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}    if q:        results.update({"q": q})    return resultsif __name__ == '__main__':    uvicorn.run("查詢參數和字符串校驗:app", reload=True)

第一步：

導入Query

from fastapi import Query

第二步：

將Query作為查詢參數的默認值，並進行限制

import uvicornfrom typing import Optionalfrom fastapi import FastAPI, Queryapp = FastAPI()# 默認值@app.get("/items/")async def read_items(q: Optional[str] = Query(None, min_length=3, max_length=50, regex="^love$")):    """    :param q: 可選參數    ：None = 可填參數為空  如果將None 更換成一個字符串 比如 "test" 該q的值就會成默認值"test"    ：min_length = 最小字符串數量    ：max_length = 最大字符串數量    ：regex = 正則匹配 ^ 以該符號之后的字符開頭； $ 到此結束    :return:    """    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}    if q:        results.update({"q": q})    return resultsif __name__ == '__main__':    uvicorn.run("查詢參數和字符串校驗:app", reload=True)

默認值

"""
向Query 的第一個參數傳入None的作用是默認值，同樣你也可以傳其他的默認值
假設你想要聲明查詢參數q，限制它的最短字符串為3，並默認值為 demo，你可以這么做
"""

import uvicornfrom typing import Optionalfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")async def read_items(q: str = Query("demo", min_length=3)):    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}    if q:        results.update({"q": q})    return resultsif __name__ == '__main__':    uvicorn.run("查詢參數和字符串校驗:app", reload=True)

必填值

"""
當我們要聲明一個參數為必填值時，只需要將Query里面的「None」改成「...」就好了，不用管它是什么含義，你就簡單的理解「...」是必傳的就對了
"""

import uvicornfrom typing import Optionalfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")async def read_items(q: str = Query(..., min_length=3)):    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}    if q:        results.update({"q": q})    return resultsif __name__ == '__main__':    uvicorn.run("查詢參數和字符串校驗:app", reload=True)

參數傳入多個值

"""
當你使用Query定義查詢參數時，你還可以聲明它去接收一組值，簡單來說接收多個值
"""

import uvicornfrom typing import Optionalfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")async def read_items(q: Optional[List[str]] = Query(None)):    query_items = {"q": q}    return query_items# 傳入：http://localhost:8000/items/?q=foo&q=bar# 返回："""{  "q": [    "foo",    "bar"  ]}"""if __name__ == '__main__':    uvicorn.run("查詢參數和字符串校驗:app", reload=True)

使用List

"""
你可以直接使用List代替List[str]
兩者區別：
List[int] ：將檢查列表的內容必須為整數
List：將不會做任何檢查
"""

import uvicornfrom typing import Optionalfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")async def read_items(q: Optional[list] = Query(None)):    query_items = {"q": q}    return query_items# 傳入：http://localhost:8000/items/?q=foo&q=bar# 返回："""{  "q": [    "foo",    "bar"  ]}"""if __name__ == '__main__':    uvicorn.run("查詢參數和字符串校驗:app", reload=True)

別名參數

"""
如果你想查詢參數為 text
傳入：http://localhost:8000/items/?text=demo
但是text 不是一個有效的Python變量名稱
這時你可以用 alias 參數聲明一個別名，該別名在URL中查到查詢參數值
"""

import uvicorn
from typing import Optional

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(q: Optional[str] = Query(None, alias="text")):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

# 傳入：http://localhost:8000/items/?text=demo
# 返回：
"""
{
  "items": [
    {
      "item_id": "Foo"
    },
    {
      "item_id": "Bar"
    }
  ],
  "q": "demo"
}
"""

if __name__ == '__main__':
    uvicorn.run("查詢參數和字符串校驗:app", reload=True)

棄用參數

"""
如果假設你不在喜歡這個參數了，但是你又要保留一段時間，你可以引用棄用參數，它可以在文檔中清晰的展示為已棄用
我們只需要傳入一個參數 deprecated=True 傳入Query即可
"""

import uvicornfrom typing import Optionalfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")async def read_items(    q: Optional[str] = Query(        None,        alias="item-query",        title="Query string",        description="Query string for the items to search in the database that have a good match",        min_length=3,        max_length=50,        regex="^fixedquery$",        deprecated=True,    )):    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}    if q:        results.update({"q": q})    return results# 在swagger 里面就會有個紅色的標識if __name__ == '__main__':    uvicorn.run("查詢參數和字符串校驗:app", reload=True)

Swagger文檔：

路徑參數和數值校驗

"""
Path為路徑參數聲明，路徑參數是必需的
所以，聲明的時候「...」將其標記為必填參數
"""

第一步：

導入Path

from fastapi import FastApi, Path, Query

第二步：

from typing import Optionalimport uvicornfrom fastapi import FastAPI, Path, Queryapp = FastAPI()@app.get("/items/{item_id}")async def read_items(        item_id: int = Path(..., title="獲取項目ID"),        q: Optional[str] = Query(None, alies="item-query"),):    results = {"item_id": item_id}    if q:        results.update({"q": q})    return resultsif __name__ == '__main__':    uvicorn.run("路徑參數和數據校驗:app", reload=True)

todo：需要弄懂這個*號含義

數值校驗：大於等於

"""
如果我們想要一個值大於100，我們可以聲明數值約束，
只需要加一個參數「ge=x」，限制一個值大於或等於x
"""

from typing import Optionalimport uvicornfrom 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=100), q: str):    results = {"item_id": item_id}    if q:        results.update({"q": q})    return resultsif __name__ == '__main__':    uvicorn.run("路徑參數和數據校驗:app", reload=True)

數據校驗：大於和小於等於

"""
gt：大於
le：小於等於
"""

from typing import Optionalimport uvicornfrom 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=100),     q: str):    results = {"item_id": item_id}    if q:        results.update({"q": q})    return resultsif __name__ == '__main__':    uvicorn.run("路徑參數和數據校驗:app", reload=True)

數據校驗：浮點數、大於和小於

"""
例如：要求一個值必須大於0，小於1，因此，0.5將是有效的，但是0.0或0不是。
lt：小於
ge：大於等於
"""

from typing import Optionalimport uvicornfrom fastapi import FastAPI, Path, Queryapp = FastAPI()# 數值校驗: 浮點數、大於和小於@app.get("/items6/{item_id}")async def read_items6(        *,        item_id: int = Path(..., title="The ID of the item to get", ge=0, le=1000),        # item_id: 大於0，小於等於1000        q: str,        size: float = Query(..., gt=0, lt=10.5)        # 范圍: 大於0，小於10.5):    results = {"item_id": item_id}    if q:        results.update({"q": q})    return resultsif __name__ == '__main__':    uvicorn.run("路徑參數和數據校驗:app", reload=True)

總結：

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

請求體-多個參數

"""
混合使用Path，Query和請求體參數
"""

import uvicornfrom fastapi import FastAPI, Pathfrom typing import Optionalfrom pydantic import BaseModelapp = FastAPI()class Item(BaseModel):    name: str    description: Optional[str] = None    price: float    tax: Optional[float] = None# 混合使用Path，Query和請求體參數@app.put("/items/{item_id}")async def updata_item(        *,        item_id: int = Path(..., title="The ID of the item to get", ge=0, le=1000),        q: Optional[str] = None,        item: Optional[Item] = None):    results = {"item_id": item_id}    if q:        results.update({"q": q})    if item:        results.update({"item": item})    return resultsif __name__ == '__main__':    uvicorn.run("請求體-多個參數:app", reload=True)

多個請求體參數

import uvicornfrom fastapi import FastAPI, Pathfrom typing import Optionalfrom pydantic import BaseModelapp = FastAPI()class Item(BaseModel):    name: str    description: Optional[str] = None    price: float    tax: Optional[float] = Noneclass User(BaseModel):    username: str    full_name: Optional[str] = None# 多個請求體參數@app.put("/items/{item_id}")async def update_item(item_id: int, item: Item, user: User):    results = {"item_id": item_id, "item": item, "user": user}    return resultsif __name__ == '__main__':    uvicorn.run("請求體-多個參數:app", reload=True)

添加請求體的單一值：

"""
除了請求item和user之外，還想在單獨請求另一個鍵 age，你可以添加請求體的單一值
"""

import uvicornfrom fastapi import FastAPI, Path, Bodyfrom typing import Optionalfrom pydantic import BaseModelapp = FastAPI()class Item(BaseModel):    name: str    description: Optional[str] = None    price: float    tax: Optional[float] = Noneclass User(BaseModel):    username: str    full_name: Optional[str] = None# 請求體的單一值@app.put("/items/{item_id}")async def updata_item(        item_id: int, item: Item, user: User, age: int = Body(...)):    results = {"item_id": item_id, "item": item, "user": user, "age": age}    return results# 這樣就可以單獨的請求就加載了json里"""{  "item": {    "name": "string",    "description": "string",    "price": 0,    "tax": 0  },  "user": {    "username": "string",    "full_name": "string"  },  "age": 0}"""if __name__ == '__main__':    uvicorn.run("請求體-多個參數:app", reload=True)

多個請求體參數和查詢參數

"""
任何時候都可以聲明額外的查詢參數
"""

import uvicorn
from fastapi import FastAPI, Path, Body
from typing import Optional

from pydantic import BaseModel

app = FastAPI()


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


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

# 多個請求體參數和查詢參數
@app.put("/items/{item_id}")
async def updata_item(
        *,
        item_id: int,
        item: Item,
        user: User,
        importance: int = Body(..., gt=0),  # Body具有與Query、Path一樣的相同的數值校驗
        q: Optional[str] = None  # 等價於  q: Optional[str] = Query(None)
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    if q:
        results.update({"q": q})
    return results

if __name__ == '__main__':
    uvicorn.run("請求體-多個參數:app", reload=True)

嵌入單個請求體參數

"""
希望在item鍵並在值中包含模型內容JSON
並需要利用Body方法的embed參數，方能解析出RequestBody 內容
item: Item = Body(..., embed=True)
"""

import uvicornfrom fastapi import FastAPI, Path, Bodyfrom typing import Optionalfrom pydantic import BaseModelapp = FastAPI()class Item(BaseModel):    name: str    description: Optional[str] = None    price: float    tax: Optional[float] = Noneclass User(BaseModel):    username: str    full_name: Optional[str] = None# 嵌入單個請求體參數@app.put("/items/{item_id}")async def updata_item(        item_id: int, item: Item = Body(..., embed=True)):    results = {"item_id": item_id, "item": item}    return results# 返回:"""{    "item": {        "name": "Foo",        "description": "The pretender",        "price": 42.0,        "tax": 3.2    }}"""# 而不是返回"""{    "name": "Foo",    "description": "The pretender",    "price": 42.0,    "tax": 3.2}"""if __name__ == '__main__':    uvicorn.run("請求體-多個參數:app", reload=True)

請求體-字段

"""
Field 字段的意思其實就是類似上面Query，Path，也同樣給給Body內的字段的信息添加相關的校驗，通俗來說就是通過Field來規范提交Body參數信息
"""

第一步：

導入Field

from pydantic import BaseModel, Field

第二步：

舉個例子

import uvicornfrom fastapi import FastAPI, Bodyfrom typing import Optionalfrom pydantic import BaseModel, Fieldapp = FastAPI()class Item(BaseModel):    name: str    description: Optional[str] = Field(        None, title="The description of the item", max_length=300    )    price: float = Field(..., gt=0, description="The price must be greater than zero")    tax: Optional[float] = None@app.put("/items/{item_id}")async def update_item(item_id: int, item: Item = Body(..., embed=True)):    results = {"item_id": item_id, "item": item}    return resultsif __name__ == '__main__':    uvicorn.run("請求體-字段:app", reload=True)

請求體-嵌套模型

"""
定義、校驗、記錄文檔並使用任意深度嵌套的模型
"""

List字段

"""
你可以將一個屬性定義為擁有子元素的類型
"""

第一步：

從typing 導入 List

from typing import List, Optional

import uvicornfrom fastapi import FastAPI, Bodyfrom typing import Optional, Listfrom pydantic import BaseModel, Fieldapp = FastAPI()class Item(BaseModel):    name: str    description: Optional[str] = None    price: float    tax: Optional[float] = None    tags: List = []@app.put("/items/{item_id}")async def update_item(item_id: int, item: Item):    results = {"item_id": item_id, "item": item}    return resultsif __name__ == '__main__':    uvicorn.run("請求體-嵌套模型:app", reload=True)

定義子模型

首先，我們要定義一個Image模型：

import uvicorn

from fastapi import FastAPI, Body
from typing import Optional, List, Set
from pydantic import BaseModel, Field

app = FastAPI()

# 定義一個Image模型
class Image(BaseModel):
    url: str
    name: str
 
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    image: Optional[Image] = None   # 將其用作一個屬性的類型

# 定義子模型    
@app.put("/items2/{item_id}")
async def updata_item2(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results

"""
這意味着FastApi類似於以下內容的請求體：
{
  "item_id": 1,
  "item": {
    "name": "Foo",
    "description": "The Pretender",
    "price": 43,
    "tax": 3.2,
    "image": {
      "url": "https://www.baidu.com",
      "name": "百度一下，你就知道"
    }
  }
}
"""

if __name__ == '__main__':
    uvicorn.run("請求體-嵌套模型:app", reload=True)

特殊的類型和校驗

"""
這里的特殊指檢測url是否合規
例如：在Image模型中我們又一個Url字段，我們可以把它聲明為Pydantic的HttpUrl，而不是str
該字符串將被檢查是否為有效的Url，並在JSON Schema/OpenApi文檔中進行記錄
"""

import uvicorn

from fastapi import FastAPI, Body
from typing import Optional, List, Set
from pydantic import BaseModel, Field, HttpUrl

app = FastAPI()


class Image(BaseModel):
    url: HttpUrl   # HttpUrl 回檢查是否為有效的Url
    name: str


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

 # 定義子模型
@app.put("/items2/{item_id}")
async def updata_item2(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results

# 返回
"""
這意味着FastApi類似於以下內容的請求體：
{
  "item_id": 1,
  "item": {
    "name": "Foo",
    "description": "The Pretender",
    "price": 43,
    "tax": 3.2,
    "image": {
      "url": "https://www.baidu.com",   # 正確的Url回返回正常的200
      "name": "百度一下，你就知道"
    }
  }
}
"""

if __name__ == '__main__':
    uvicorn.run("請求體-嵌套模型:app", reload=True)

深度嵌套模型

"""
可以自定義任意深度的嵌套模型
"""

import uvicorn
from typing import List, Optional, Set
from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


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


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


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


@app.post("/offers/")
async def create_offer(offer: Offer):
    return offer


if __name__ == '__main__':
    uvicorn.run("demo2:app", reload=True)

Swagger UI 上的效果

任意dict構成的請求體

"""
可以將請求聲明為使用某類型的鍵和其他類型的值的dict
例如：接收任意鍵為int類型並且值為float類型的dict
"""

import uvicornfrom typing import List, Optional, Set, Dictfrom fastapi import FastAPIapp = FastAPI()@app.post("/index-weights/")async def create_index_weights(weights: Dict[int, float]): # 接收的名為weight的dict，key為int類型，value為float類型的JSon    return weights# 返回一個類似這樣的一個JSON格式：“”“{  "1": 1.0,  "2": 2.0,  "3": 3.0}”“”if __name__ == '__main__':    uvicorn.run("demo2:app", reload=True)

總結：

JSON僅支持將str作為鍵
但是Pydantic具有自動轉換數據的功能。
API客戶端只能將字符串作為鍵發送，只要這些字符串內容僅包含整數，Pydantic就會進行轉換並校驗
然后你接收的名為weights的dict實際上具有int類型的鍵和floa類型的值

模型的額外信息

from typing import Optionalimport uvicornfrom fastapi import FastAPIfrom pydantic import BaseModelapp = FastAPI()class Item(BaseModel):    name: str    description: Optional[str] = None    price: float    tax: Optional[float] = None    class Config:        schema_extra = {            "example": {                "name": "ranyong",                "description": "A very nice Item",                "price": 35.4,                "tax": 3.2            }        }@app.put("/items/{item_id}")async def updata_item(item_id: int, item: Item):    results = {"item_id": item_id, "item": item}    return results# 返回:"""{  "item_id": 123,  "item": {    "name": "ranyong",    "description": "A very nice Item",    "price": 35.4,    "tax": 3.2  }}"""if __name__ == '__main__':    uvicorn.run("模式的額外信息-例子:app", reload=True)

使用Field參數給JSON聲明額外信息

from typing import Optionalimport uvicornfrom fastapi import FastAPIfrom pydantic import BaseModel, Fieldapp = FastAPI()class Item(BaseModel):    name: str = Field(..., example="這是姓名")    description: Optional[str] = Field(None, example="這是一個描述")    price: float = Field(..., example=36.4)    tax: Optional[float] = Field(None, exapmle=3.2)    age: str = Field(None, example="這是年齡") @app.put("/items/{item_id}")async def updata_item(item_id: int, item: Item):    results = {"item_id": item_id, "item": item}    return resultsif __name__ == '__main__':    uvicorn.run("模式的額外信息-例子:app", reload=True)

Swagger UI展示(注意傳遞的參數不會添加任何驗證，只會添加注釋，用於文檔的目的)

Body 額外參數

"""
可以將請求體的一個example傳遞給Body
"""

from typing import Optionalimport uvicornfrom fastapi import FastAPI, Bodyfrom pydantic import BaseModel, Fieldapp = FastAPI()class Item(BaseModel):    name: str = Field(..., example="這是姓名")    description: Optional[str] = Field(None, example="這是一個描述")    price: float = Field(..., example=36.4)    tax: Optional[float] = Field(None, exapmle=3.2)    age: int = Field(None, example="這是年齡")@app.put("/items/{item_id}")async def update_item(        item_id: int,        item: Item = Body(            ...,            example={                "name": "Foo",                "description": "A very nice Item",                "price": 35.4,                "tax": 3.2,                "age": 18            },        ),):    results = {"item_id": item_id, "item": item}    return resultsif __name__ == '__main__':    uvicorn.run("模式的額外信息-例子:app", reload=True)

額外數據類型

"""
其他數據類型

UUID

一種標准的“通用唯一標識符”，在請求和響應中將以str表示

datetime.datetime

在請求和響應中將表示為ISO 8601格式例如：2021-07-14T14:36:35.171Z

datetime.date

在請求和響應中將表示為ISO 8601格式例如：2021-07-15

datetime.time

在請求和響應中將表示為ISO 8601格式例如：14:36:55.003

datetime.timedelta

在請求和響應中將表示為float代表總秒數

frozenset
bytes
Decimal

"""

第一步：

導包：

from datetime import datetime, time, timedeltafrom uuid import UUID

第二步：

import uvicorn
from fastapi import FastAPI, Body
from datetime import datetime, time, timedelta
from typing import Optional
from uuid import UUID

app = FastAPI()


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


if __name__ == '__main__':
    uvicorn.run("額外數據類型:app", reload=True)

在Swagger UI中顯示

Cookie參數

"""
聲明Cookie參數的結構於聲明Query參數和Path參數相同

"""

第一步：

導入cookie

from fastapi import Cookie, FastApi

第二步：

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 fastapi import FastApi, Header

第二步：

聲明Header參數

import uvicornfrom typing import Optionalfrom fastapi import FastAPI, Header    # 導入Headerapp = FastAPI()@app.get("/items/")async def read_items(user_agent: Optional[str] = Header(None)):    return {"User-Agent": user_agent}# 輸入：http://127.0.0.1:8000/items/# 返回："""{"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.93 Safari/537.36"}"""if __name__ == '__main__':    uvicorn.run("Header參數:app", reload=True)

如果你需要禁用下划線到連字符的自動轉換，設置Header參數convert_underscores=False

@app.get("/items/")async def read_items(user_agent: Optional[str] = Header(None, convert_underscores=False)):    # 禁用下划線到連字符的自動轉換    return {"User-Agent": user_agent}

沒添加：

添加了：

注意：在設置convert_underscores 為 False 之前，請確保一些HTTP代理和服務器不允許使用帶有下划線的headers

重復的headers

"""
收到重復的headers，這意味着header具有很多個值
可以在類型聲明中使用一個list來定義這些情況
"""

import uvicornfrom typing import Optional, Listfrom fastapi import FastAPI, Headerapp = FastAPI()# 重復的headers@app.get("/items1")async def read_items1(x_token: Optional[List[str]] = Header(None)):    return {"X-Token Values": x_token}if __name__ == '__main__':    uvicorn.run("Header參數:app", reload=True)

如果發送了兩個headers，響應就會是這樣

響應模型

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

import uvicornfrom typing import Optional, Listfrom fastapi import FastAPIfrom pydantic import BaseModelapp = FastAPI()class Item(BaseModel):    name: str    description: Optional[str] = None    price: float    tax: Optional[float] = None    tags: List[str] = []@app.post("/items/", response_model=Item)   # 聲明響應模型async def create_item(item: Item):    return itemif __name__ == '__main__':    uvicorn.run("響應模型:app", reload=True)

這里有個坑，如果你按照pip執行不成功的話，你可以這樣做
120831626330160_.pic.jpg
安裝這個模塊：
或者這樣：

返回於輸入相同的數據

"""
例如：要求輸入賬號密碼郵箱名字等等，這邊只需要return 返回賬號郵箱名字密碼不會返回回來
"""

import uvicorn
from typing import Optional, List
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Optional[str] = None


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


# 返回於輸入相同的數據
@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):
    return user


if __name__ == '__main__':
    uvicorn.run("響應模型:app", reload=True)

響應模型編碼參數

"""
這個主要運用到你提前設定好數據，然后輸入可以把提前設定好的數據自動填裝
你需要設置路徑操作裝飾器的 response_model_exclude_unset=True 參數
"""

import uvicornfrom typing import Optional, Listfrom fastapi import FastAPIfrom pydantic import BaseModelapp = FastAPI()class Item(BaseModel):    name: str    description: Optional[str] = None    price: float    tax: float = 10.5    tags: List[str] = []items = {    "foo": {"name": "Foo", "price": 50.2},    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},}@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)async def read_item(item_id: str):    return items[item_id]# 輸入:http://127.0.0.1:8000/items/foo# 返回："""{  "name": "Foo",  "price": 50.2}"""# 輸入：http://127.0.0.1:8000/items/bar# 返回："""{  "name": "Bar",  "description": "The bartenders",  "price": 62,  "tax": 20.2}"""if __name__ == '__main__':    uvicorn.run("響應模型:app", reload=True)

具有與默認值相同值的數據

"""
支持模型包含和模型排除
簡單來說就是你想需要什么值，不想要什么值
response_model_include或response_model_exclude 來省略某些屬性
"""

import uvicornfrom typing import Optionalfrom fastapi import FastAPIfrom pydantic import BaseModelapp = FastAPI()class Item(BaseModel):    name: str    description: Optional[str] = None    price: float    tax: float = 10.5items = {    "foo": {"name": "Foo", "price": 50.2},    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},    "baz": {        "name": "Baz",        "description": "There goes my baz",        "price": 50.2,        "tax": 10.5    },}@app.get(    "/items/{item_id}/name",    response_model=Item,    # response_model_include 需要保留的屬性    response_model_include={"name", "price"})async def read_item(item_id: str):    return items[item_id]@app.get("/items/{item_id}/public", response_model=Item, response_model_exclude={"price"})# response_model_exclude 需要忽略的屬性async def read_item_public_data(item_id: str):    return items[item_id]if __name__ == '__main__':    uvicorn.run("響應模型:app", reload=True)

response_model_include 需要保留的屬性

response_model_exclude 需要忽略的屬性

總結：
使用路徑操作裝飾器 response_model 參數來定義響應模型，特別是確保私有數據被過濾掉
使用response_model_exclude_unset來僅返回顯示設定的值

額外的模型

"""
輸入模型需要擁有密碼屬性
輸出模型不應該包含密碼
數據庫模型很可能需要保存密碼的哈希值
"""

多個模型

todo：沒學會！

響應狀態碼

"""
來聲明用作與響應HTTP狀態碼
類似於：
@app.get()
@app.post()
@app.put()
@app.delete()
status_code 參數接收一個表示HTTP狀態碼的數字
"""

import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.post("/items/", status_code=201)
async def create_item(name: str):
    return {"name": name}


if __name__ == '__main__':
    uvicorn.run("響應狀態碼:app", reload=True)

注意：status_code 是「裝飾器」方法（get，post）的一個參數，不屬於路徑操作函數

介紹一種名稱快捷的方法

第一步：
導包:

from fastapi import FastApi, status

第二步：

import uvicornfrom fastapi import FastAPI, statusapp = FastAPI()@app.post("/items/", status_code=status.HTTP_201_CREATED)  #有足夠多的狀態碼供你選擇async def create_item(name: str):    return {"name": name}if __name__ == '__main__':    uvicorn.run("響應狀態碼:app", reload=True)

表單數據

"""
接收的不是JSON，而是表單字段，要使用Form
如果使用到Form表單，需要安裝庫
pip install python-multipart
"""

# 導入Formimport uvicornfrom fastapi import FastAPI, Formapp = FastAPI()@app.post("/login/")async def login(username: str = Form(...), password: str = Form(...)):    return {"username": username}if __name__ == '__main__':    uvicorn.run("表單數據:app", reload=True)

特別注意的是：

username和password最好通過表單字段來傳遞，因為符合密碼流，不能通過JSON
在一個路徑操作中聲明多個Form參數，但不能同時聲明要接收JSON的Body字段，因為此時請求體的編碼是application/x-www-form-urlencoded，而不是application/json

請求文件

"""
File 用於定義客戶端的上傳文件
因為上傳文件是以「表單數據」形式發送
需要安裝庫
pip install python-multipart
"""

# 導入 Fileimport uvicornfrom fastapi import FastAPI, File, UploadFile # 導包app = 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}if __name__ == '__main__':    uvicorn.run("請求文件:app", reload=True)

特別注意的是：

File是直接繼承自Form的類
聲明文件體必須使用File，否則，FastApi會把該參數當作查詢參數或請求體(JSON)參數
更推薦使用UploadFile
UploadFile 與 bytes 相比有更多優勢：
使用spooled文件：
存儲在內存的文件超出最大上限時，FastApi會把文件存儲到磁盤
這種方式更適於處理圖像、視頻、二進制文件等大型文件，好處是不會占用所有內存
可獲取上傳文件的元數據
自帶async接口
**UploadFile 的屬性如下：
filename:上傳文件名字符串(str)，例如：myimage.jpg
content-type:內容類型字符串(str) 例如：image/jepg
file:其實就是Python文件，可直接傳遞給其他預期file-like 對象的函數或支持庫**
**UploadFile 支持的語法
write(data)：把data寫入文件
read(size)：按指定數量的字節或字符讀取文件內容
seek(offset)：移動至文件offset 字節處的位置
例如：await myfile.seek(0) 移動到文件開頭
執行 await mylife.read() 后,需要再次讀取已讀取內容時
close()：關閉文件**

以上方法都是async方法，要搭配「await」使用
例如，在async路徑操作函數內，要用一下方式讀取文件內容：

contens = myfile.file.read()

什么是「表單數據」

不包含文件時，表單數據一般用application/x-www-form-urlencoded
但是表單包含文件時，編碼為multipart/form-data。使用了File，FastApi就知道從請求體的正確獲取文件

多文件上傳

"""
同時上傳多個文件
可用同一個「表單字段」發送含多個文件的「表單數據」
上傳多個文件時，要聲明含bytes或UploadFile的列表
"""

import uvicornfrom fastapi import FastAPI, File, UploadFilefrom typing import Listfrom starlette.responses import HTMLResponseapp = FastAPI()# 多文件上傳@app.post("/files")async def create_files(files: List[bytes] = File(...)):  # 以byte形式    return {"file_sizes": [len(file) for file in files]}@app.post("/uploadfiles/")async def create_upload_files(files: List[UploadFile] = File(...)): # 以名字的形式    return {"filenames": [file.filename for file in files]}if __name__ == '__main__':    uvicorn.run("請求文件:app", reload=True)

以byte形式展示：

以名字的形式展示：

import uvicornfrom fastapi import FastAPI, File, UploadFilefrom typing import Listfrom starlette.responses import HTMLResponseapp = FastAPI()@app.get("/")async def main():    content = """<form action="/files/" enctype="multipart/form-data" method="post"><input name="files" type="file" multiple=""><input type="submit"></form><form action="/uploadfiles/" enctype="multipart/form-data" method="post"><input name="files" type="file" multiple=""><input type="submit"></form>    """    return HTMLResponse(content=content)if __name__ == '__main__':    uvicorn.run("請求文件:app", reload=True)

請求表單與文件

"""
FastApi 支持同時使用File和Form定義文件和表單字段
需要安裝庫
pip install python-multipart
"""

第一步：

導入 File 與 Form

from fastapi import FastAPI, File, Form, UploadFile

第二步：

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

app = FastAPI()


@app.post("/files")
async def create_file(
        # 定義File與Form參數
        file: bytes = File(...),
        fileb: UploadFile = File(...),
        token: str = Form(...)
):
    return {
        "file_size": len(file),
        "token": token,
        "fileb_content_type": fileb.content_type,
    }


if __name__ == '__main__':
    uvicorn.run("請求表單與文件:app", reload=True)

總結：

在同一個請求中接收數據和文件時，應同時使用File和Form

處理錯誤

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

第一步：

導入：

from fastapi import FastApi, HTTPException

第二步：

# 導入HTTPException
import uvicorn
from fastapi import FastAPI, HTTPException

app = FastAPI()
items = {"foo": "The Foo Wrestlers"}


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


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

# 輸入：http://127.0.0.1:8000/items/foo1
# 返回：
"""
{
  "detail": "Item not found"
}
"""

if __name__ == '__main__':
    uvicorn.run("處理錯誤:app", reload=True)

異常處理器

"""
FastApi 自帶了一些默認異常處理器
當我們接收到無效數據時，FastApi內部會觸發RequestValidationError
該異常內置了默認異常處理器
"""

第一步：

導入RequestValidationError

from fastapi.exceptions import RequestValidationError

並用@app.excption_handler(RequestValidationError)裝飾器處理異常

第二步：

import uvicornfrom fastapi import FastAPI, HTTPExceptionfrom fastapi.responses import PlainTextResponsefrom fastapi.exceptions import RequestValidationErrorfrom starlette.exceptions import HTTPException as StarletteHTTPExceptionapp = FastAPI()@app.exception_handler(StarletteHTTPException)async def http_exception_handler(request, exc):    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)@app.exception_handler(RequestValidationError)async def validation_exception_handler(request, exc):    return PlainTextResponse(str(exc), status_code=400)@app.get("/items/{item_id}")async def read_item(item_id: int):    if item_id == 3:        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")    return {"item_id": item_id}# 輸入：http://127.0.0.1:8000/items/3# 返回："""{"detail": "Nope! I don't like 3."}"""if __name__ == '__main__':    uvicorn.run("處理錯誤:app", reload=True)

請求體-更新數據

"""
用PUT更新數據
"""

更新所有數據（也可更新部分數據）

import uvicorn
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from typing import List, Optional
from pydantic import BaseModel

app = FastAPI()


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


items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}


@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: str):
    """獲取數據"""
    return items[item_id]


@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    """更新數據"""
    update_item_encoded = jsonable_encoder(item)
    items[item_id] = update_item_encoded
    return update_item_encoded


if __name__ == '__main__':
    uvicorn.run("請求體-更新數據:app", reload=True)

先PUT，后GET

只發送要更新的數據，其余數據不變

import uvicorn
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from typing import List, Optional
from pydantic import BaseModel

app = FastAPI()


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


items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

"""
用PATCH進行部分更新
只發送要更新的數據，其余數據保持不變
"""

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: str):
    """獲取數據"""
    return items[item_id]


@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    """部分更新數據"""
    stored_item_data = items[item_id]
    stored_item_model = Item(**stored_item_data)
    update_data = item.dict(exclude_unset=True)
    updated_item = stored_item_model.copy(update=update_data)
    items[item_id] = jsonable_encoder(updated_item)
    return updated_item


if __name__ == '__main__':
    uvicorn.run("請求體-更新數據:app", reload=True)

使用Pydantic的update參數

"""
.copy()：為已有模型創建調用update參數的腳本，該參數為包含更新數據的dict
簡而言之就是復制出一份后，以你這個為准，原先的被覆蓋
"""

import uvicorn
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from typing import List, Optional
from pydantic import BaseModel

app = FastAPI()


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


items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

"""
用 .copy() 為已有模型創建調用update參數的副本，該參數為包含更新數據的dict
例如：stored_item_model.copy(update=update_data):
"""

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: str):
    return items[item_id]


@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    stored_item_data = items[item_id]
    stored_item_model = Item(**stored_item_data)
    update_data = item.dict(exclude_unset=True)
    updated_item = stored_item_model.copy(update=update_data)
    items[item_id] = jsonable_encoder(updated_item)
    return updated_item


if __name__ == '__main__':
    uvicorn.run("請求體-更新數據:app", reload=True)

依賴注入

"""
相同的邏輯判斷處理
共享數據庫連接
用戶身份鑒權，角色管理
業務邏輯復用
提高代碼的復用，減少代碼重復
"""

第一步：

導入Depends：

from fastapi import Depends, FastApi

第二步：

創建一個依賴項，或「被依賴項」

import uvicorn
from fastapi import FastAPI, Depends

app = FastAPI()

# 創建依賴 僅需要2行
async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    """
    common_parameters函數主要是負責接收函數，處理后返回一個字典
    :param q: 可選查詢參數q那是一個str
    :param skip: 默認情況下是0
    :param limit: 默認情況下是100
    :return: 返回一個字典
    """
    return {"q": q, "skip": skip, "limit": limit}

# 聲明依賴
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):  # 聲明了一個依賴關系，表示的是接口參數請求依賴與common_parameters的函數
    commons.update({'小鍾': '同學'})
    return commons

# 聲明依賴
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons


if __name__ == '__main__':
    uvicorn.run("依賴注入:app", reload=True)

注意：您只需要給Depends一個參數，此參數必須類似於函數

把類當作被依賴對象

import uvicorn
from fastapi import FastAPI, Depends

app = FastAPI()

# 把類當作被依賴對象
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


class CommonQueryParams:
    def __init__(self, q: str = None, name: str = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.name = name
        self.skip = skip
        self.limit = limit


@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    """
    :param commons: CommonQueryParams = Depends(CommonQueryParams)和 commons = Depends(CommonQueryParams)是等價的
    :return:
    """
    response = {}
    # 如果q存在
    if commons.q:
        # 我們就把q加到一個新字典
        response.update({"q": commons.q})
        response.update({"小鍾": '同學'})
    elif commons.name:
        response.update({"name": commons.name})
    # 然后在我們的fake_items_db進行截取
    items = fake_items_db[commons.skip: commons.skip + commons.limit]
    response.update({"items": items})
    return response

# 輸入：http://127.0.0.1:8000/items/?q=1&name=admin
# 返回：
"""
{
  "q": "q",
  "小鍾": "同學",
  "name": "admin",
  "items": [
    {
      "item_name": "Foo"
    },
    {
      "item_name": "Bar"
    },
    {
      "item_name": "Baz"
    }
  ]
}
"""

if __name__ == '__main__':
    uvicorn.run("依賴注入:app", reload=True)

有q，name參數：

沒有q，name參數：

多層嵌套依賴

import uvicorn
from fastapi import FastAPI, Depends, Cookie

app = FastAPI()


# 多層嵌套依賴
def query_extractor(q: str = None):
    return q


def query_or_cookie_extractor(
        q: str = Depends(query_extractor), last_query: str = Cookie(None)
):
    """
    query_or_cookie_extractor 依賴於 query_extractor
    然后 query_or_cookie_extractor被注入到接口上也被依賴的對象
    :param q:
    :param last_query:
    :return:
    """
    if not q:
        return last_query
    return q


@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
    return {"q_or_cookie": query_or_default}

# 對於同一個依賴，如果處理的結果是一樣的，就是返回值是一樣的話，我們可以進行多次調用依賴，這時候可以對被依賴的對象設置是否使用緩存機制：
@app.get("/items1")
async def needy_dependency(fresh_value: str = Depends(query_or_cookie_extractor, use_cache=False)):
    """
    use_cache=False 不啟動緩存機制
    """
    return {"fresh_value": fresh_value}

if __name__ == '__main__':
    uvicorn.run("依賴注入:app", reload=True)

官網給的例子：

list列表依賴

"""
list列表的依賴意思就是必須兩條條件都成立才通過
"""

import uvicorn
from fastapi import FastAPI, Depends, Header, HTTPException

app = FastAPI()


# list列表依賴
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")


@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]


if __name__ == '__main__':
    uvicorn.run("依賴注入:app", reload=True)

多依賴對象注入

import uvicorn
from fastapi import FastAPI, Depends, Header, HTTPException

app = FastAPI()


# 多依賴對象注入
async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")

        
@app.get("/items2")
async def items2(xt: str = Depends(verify_token), xk: str = Depends(verify_key)):
    return {"xt": xt, "xk": xk}


if __name__ == '__main__':
    uvicorn.run("依賴注入:app", reload=True)

安全性

"""
/ todo 待補充
"""

第一步：

安裝依賴庫(因為OAuth2使用“form data”發送用戶名和密碼)

pip install python-multipart

第二步：

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

app = FastAPI()
"""
當我們在Tokenurl 參數中創建OAuth2PasswordBearer類的實例時，
此參數包含客戶端(用戶瀏覽器中運行的前端)將用與發送用戶名和秘密啊以獲取令牌的URL
"""
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


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


if __name__ == '__main__':
    uvicorn.run("安全性:app", reload=True)

在Swagger UI 中：

點擊它：

如果沒有輸入username和password，直接try it out 會直接響應401狀態碼(未經授權)

案例一：獲取當前用戶

import uvicorn
from typing import Optional
from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

app = FastAPI()
"""
當我們在Tokenurl 參數中創建OAuth2PasswordBearer類的實例時，此參數包含客戶端(用戶瀏覽器中運行的前端)將用與發送用戶名和秘密啊以獲取令牌的URL
"""
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


# oauth2_scheme變量是OAuth2PasswordBearer的實例，但它也是一個"可調用的"

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


# 創建(偽)工具函數，該函數接收str類型的令牌並返回到User模型
def fake_decode_token(token):
    return User(
        username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
    )


# 創建一個get_current_user 依賴項
async def get_current_user(token: str = Depends(oauth2_scheme)):
    """
    依賴項get_current_user將從子依賴項oauth2_scheme中接收一個str類型的token
    :param token:
    :return:
    """
    user = fake_decode_token(token)
    return user


@app.get("/users/me")
# 注入當前用戶
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user


if __name__ == '__main__':
    uvicorn.run("安全性:app", reload=True)

案例二：使用密碼和Bearer的簡單OAuth2

導入OAuth2PasswordRequestForm，然后在token的路徑操作中通過Depends將其作為依賴項使用

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

# 獲取username和password的代碼
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "fakehashedsecret",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "alice@example.com",
        "hashed_password": "fakehashedsecret2",
        "disabled": True,
    },
}

app = FastAPI()


def fake_hash_password(password: str):
    return "fakehashed" + password


"""
當我們在Tokenurl 參數中創建OAuth2PasswordBearer類的實例時，此參數包含客戶端(用戶瀏覽器中運行的前端)將用與發送用戶名和秘密啊以獲取令牌的URL
"""
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


# oauth2_scheme變量是OAuth2PasswordBearer的實例，但它也是一個"可調用的"


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


class UserInDB(User):
    hashed_password: str


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


def fake_decode_token(token):
    user = get_user(fake_users_db, token)
    return user


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="無效身份驗證",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user


async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="無效用戶")
    return current_user


@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user_dict = fake_users_db.get(form_data.username)
    if not user_dict:
        raise HTTPException(status_code=400, detail="賬號密碼不符")
    user = UserInDB(**user_dict)
    """
    UserInDB(**user_dict) ===
    UserInDB(
        username= user_dict["username"],
        email = user_dict["email"],
        full_name = user_dict["full_name"],
        disabled = user_dict["disabled"],
        hashed_password = user_dict["hashed_password"],
        )
    """
    hashed_password = fake_hash_password(form_data.password)
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="用戶名和密碼不符")
    return {"access_token": user.username, "token_type": "bearer"}


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


if __name__ == '__main__':
    uvicorn.run("安全性:app", reload=True)

在Swagger UI 中

點擊Authorize

點擊登陸后

如果未啟用的用戶(disabled=True),發送get("/users/me")請求就會得到一個「未啟用的用戶」錯誤

使用(哈希)密碼和JWT Bearer 令牌的OAuth2

"""
什么是JWT？
簡單來說就是
會給你一個時效性為一周的令牌，你拿到了令牌你一周可以免登陸，一周過后，你需要重新發起請求獲取令牌
需要安裝依賴庫：
pip install python-jose

**什么是哈希密碼？

**簡單來說就是
將內容轉換成一些像亂碼的字節序列
每次傳入完全相同的密碼時，你就會得到完全相同的亂碼
你只能轉換成亂碼，不能將亂碼轉換成字符
需要安裝依賴庫：
pip install passlib
"""
建議回頭重看
https://fastapi.tiangolo.com/zh/tutorial/security/oauth2-jwt/

中間件

// todo 后續回看

異步、后台任務

"""
需要在request執行之后繼續操作，但終端並不需要等待這些操作完成才能收到response
例如：
1、執行完request之后發送郵件通知
2、收到文件之后對文件進行二次處理
我們可以通過定義后台任務BackgroundTasks來實現這個功能
"""

# 使用BackgroundTasks
import uvicorn
from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: str = None):
    if q:
        message = f"發送內容: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
        email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)
):
    message = f"郵箱來自: {email}\n"
    background_tasks.add_task(write_log, message)
        """
    .add_task()接收的參數
    - 一個在后台運行的任務函數(write_notification)
    - 按照順序傳遞的一個系列參數(email)
    - 任何的關鍵字參數(essage="notification..)
    """
    return {"message": "Message sent", "q": q}


if __name__ == '__main__':
    uvicorn.run("后台任務:app", reload=True)

免責聲明！

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

猜您在找 持續更新 Linux命令（持續更新） CAShapeLayer(持續更新) mpvue的坑，持續更新-.- 總結與心得（持續更新）三.問題匯總（持續更新...） HTTP協議（持續更新） Zabbix故障總結(持續更新) Fabric使用詳解，持續更新。。。 ES6簡述（持續更新…）