web開發(fā)框架之fastapi（二）

這里主要介紹fastapi的Input/Output，或者說Request/Response相關(guān)內(nèi)容，這也是用好fastapi的關(guān)鍵了。
有興趣的讀者，可以詳細(xì)的查看：learn fastapi。

Input

我們知道，HTTP支持 POST/GET/PUT/DELETE?？梢允褂?@app.post(), @app.get(), @app.put(), @app.delete()來定義。

關(guān)于Path參數(shù)

  • 基礎(chǔ)使用

參數(shù)可以在path中聲明：

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

還可以給定類型標(biāo)簽，如：item_id: int。

  • 使用枚舉

需要注意，還可以使用Enum類型，這樣路徑就是一些可以枚舉的類型了：

from enum import Enum
from fastapi import FastAPI

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

app = FastAPI()

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    ......
  • 參數(shù)帶/路徑的場景

比如有一個文件的路徑是 /home/john/myfile.txt，那我們?nèi)绻@么傳遞是有問題的：/files/home/johndoe/myfile.txt。
可以這樣處理：

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

然后傳遞參數(shù)用：/files//home/johndoe/myfile.txt，注意是 // 不是 /。

關(guān)于Query參數(shù)

對于非路徑中的參數(shù)，如下所示 http://127.0.0.1:8000/items/?skip=0&limit=10：
其中query參數(shù)包括了：skip，limit。也可以在參數(shù)中來進(jìn)行定義：

@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
    return fake_items_db[skip : skip + limit]

有一點不同的是可選參數(shù)：

@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str | None = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

表明q可選。
可以這么理解：如果有默認(rèn)值，則表示可選；如果沒有默認(rèn)值，則為必選，如果必選不滿足會報錯。

請求Body

請求body會用到Pydantic，發(fā)送請求body也需要使用POST，PUT，DELETE，PATCH。

使用BaseModel，需要從Pydantic導(dǎo)入。

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

請求傳參就可以使用：

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

示例代碼如下：

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

另外可以直接訪問對象的屬性：

item.price
item.tax
item.dict() # 轉(zhuǎn)化為dict

# 混用 path+body
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, **item.dict()}

Cookie

Cookie在很多場景下都有使用，比如當(dāng)用戶登錄成功后，我們在下次登錄時就自動填入用戶名和密碼。這需要我們使用cookie。cookie一般使用的是key-value來進(jìn)行存儲的。

在fastapi中使用cookie，需要import Cookie：

from fastapi import Cookie, FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(ads_id: Annotated[Union[str, None], Cookie()] = None):
    return {"ads_id": ads_id}
  • 關(guān)于類型

typing 是Python標(biāo)準(zhǔn)庫中的一個模塊，支持類型提示（在代碼中指定變量、函數(shù)參數(shù)、返回值的類型的方法）。

Annotated是一個類型提示，Annotated[T, x] 用元數(shù)據(jù) x 注解類型 T，比如：

Annotated[int, ValueRange(3, 10), ctype("char")]

Annotated 的第一個參數(shù)必須是有效類型，Annotated 至少需要兩個參數(shù)。

Header

需要導(dǎo)入Header模塊：

from typing import Annotated, Union

from fastapi import FastAPI, Header

app = FastAPI()

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

對于上面的代碼，需要注意一下：首先header是case-insensitive的，另外就是我們的header一般里面有-，但是變量不能是user-agent，所以python會將變量里面的_轉(zhuǎn)化為-來進(jìn)行匹配。

另外，可以支持多個重名的header：

from typing import Annotated, List, Union

from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
async def read_items(x_token: Annotated[Union[List[str], None], Header()] = None):
    return {"X-Token values": x_token}

這樣，header傳入和輸出示例如下：

# header
X-Token: foo
X-Token: bar

# output
{
    "X-Token values": [
        "bar",
        "foo"
    ]
}

直接使用request

有時候不希望通過Path、Query、Cookie、Header來獲取，直接通過Request對象也是可以的。

比如希望獲取到 IP 地址：

from fastapi import FastAPI, Request

# 也可以使用如下導(dǎo)入 Request
# from starlette.requests import Request

app = FastAPI()

@app.get("/items/{item_id}")
def read_root(item_id: str, request: Request):
    client_host = request.client.host
    return {"client_host": client_host, "item_id": item_id}

讀者可以參見Request Object的詳情，了解這個對象的一些方法。

Output

下面介紹我們的返回。

返回Model

用類似的方式，可以定義返回的類型，比如：

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

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

@app.get("/items/")
async def read_items() -> list[Item]:
    return [
        Item(name="Portal Gun", price=42.0),
        Item(name="Plumbus", price=32.0),
    ]

這么做，沒有什么太大的問題，不過有時候我們想做一些轉(zhuǎn)換的操作，比如：

from typing import Any

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: str | None = None

class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None

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

這里用到了response_model，會對返回的數(shù)據(jù)進(jìn)行轉(zhuǎn)換。注意，上面故意定義了return type為Any。

返回碼（Response Status Code）

我們還可以為上面的操作定義返回碼：

  • @app.get()
  • @app.post()
  • @app.put()
  • @app.delete()

默認(rèn)的返回碼方式如下：

from fastapi import FastAPI

app = FastAPI()

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

這種方式返回的返回碼是固定的，不太靈活，這個時候Response就要上場了。

from fastapi import FastAPI, Response, status
app = FastAPI()

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


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

總結(jié)

上面我們只是簡單的對fastapi的一些功能做了介紹，用戶可以參考文檔再詳細(xì)了解。

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容