Fastapi-1 Pydantic請求參數校驗

Pydantic知識

"""
Data validation and settings management using python type annotations.
使用python的類型注解來進行數據校驗和settings管理

pydantic enforces type hints at runtime, and provides user friendly errors when data is invalid.
pydantic可以在代碼運行時提供類型提示，數據校驗失敗時提供友好的錯誤提示

Define how data should be in pure, canonical python; validate it with pydantic.
定義數據應該如何在純規范的python代碼中保存，並用Pydantic驗證他
"""

from pydantic import BaseModel, ValidationError, constr
from datetime import datetime, date
from typing import List, Optional
from pathlib import Path
from sqlalchemy import Column, Integer, String
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.ext.declarative import declarative_base

print("\033[31m1. ---Pydantic的基本用法。Pycharm可以安裝Pydantic插件 ---\033[0m")


class User(BaseModel):
    id: int  # 必填字段
    name: str = 'lem'  # 有默認值，選填字段
    singup_ts: Optional[datetime] = None  #
    friends: List[int] = []  # 列表中元素是int類型或者可以直接轉換成int類型 "1"


external_data = {
    "id": "123",
    "singup_ts": "2022-12-22 12:22",
    "friends": [1, 2, "3"]  # "3" 是可以int("3")的
}

user = User(**external_data)
print(user.id, user.friends)
print(user.singup_ts)
print(repr(user.singup_ts))
print(user.dict())

print("\033[31m2. --- 校驗失敗處理 ---\033[0m")

try:
    User(id=1, singup_ts=datetime.today(), friends=[1, '2', 'not num'])
except ValidationError as e:
    print(e.json())

print("\033[31m3. --- 模型類的屬性和方法 ---\033[0m")
print(user.dict())
print(user.json())
print(user.copy())  # 這里是淺拷貝
print(User.parse_obj(obj=external_data))
print(User.parse_raw('{"id": "123","singup_ts": "2022-12-22 12:22","friends": [1, 2, "3"]}'))

path = Path('pydantic_tutorial.json')
path.write_text('{"id": "123","singup_ts": "2022-12-22 12:22","friends": [1, 2, "3"]}')
print(User.parse_file(path))  # 解析文件

print(user.schema())
print(user.schema_json())

user_data = {"id": "error", "singup_ts": "2022-12-22 12:22", "friends": [1, 2, "3"]}  # id是字符串
print(User.construct(**user_data))  # 不校驗數據直接創建模型類，不建議在construct方法中傳入未經驗證的數據

print(User.__fields__.keys())  # 定義模型類的時候，所有字段都注明類型，字段順序就不會亂

print("\033[31m4. --- 遞歸模型 ---\033[0m")


class Sound(BaseModel):
    sound: str


class Dog(BaseModel):
    birthday: date
    weight: float = Optional[None]
    sound: List[Sound]  # 不同的狗有不同的叫聲。遞歸模型(Recursive Models)就是只一個嵌套一個


dogs = Dog(birthday=date.today(), weight=6.66, sound=[{"sound": "wang wang ~"}, {"sound": "ying ying ~"}])

print(dogs.dict())

print("\033[31m5. --- ORM模型：從類實例創建符合ORM對象的模型 ---\033[0m")

Base = declarative_base()


class CompanyOrm(Base):
    __tablename__ = 'companies'
    id = Column(Integer, primary_key=True, nullable=False)
    public_key = Column(String(20), index=True, nullable=False, unique=True)
    name = Column(String(64), unique=True)
    domains = Column(ARRAY(String(255)))


class CompanyMode(BaseModel):
    id: int
    public_key: constr(max_length=20)
    name: constr(max_length=64)
    domains: List[constr(max_length=255)]

    class Config:
        orm_mode = True


co_orm = CompanyOrm(
    id=123,
    public_key='foobar',
    name='lem',
    domains=['test.com, example.com']
)

print(CompanyMode.from_orm(co_orm))

print("\033[31m5. --- Pydantic支持的字段類型 ---\033[0m")  # 官方文檔 https://pydantic-docs.helpmanual.io/usage/types/

啟動程序

import uvicorn
from fastapi import FastAPI

from tutorial import app03, app04, app05, app06

app = FastAPI()

app.include_router(app03,prefix='/chapter03', tags=['第三章 請求參數和驗證'])
app.include_router(app04,prefix='/chapter04', tags=['第四章 響應處理和FastAPI配置'])
app.include_router(app05,prefix='/chapter05', tags=['第五章 FastAPI的依賴注入系統'])


if __name__ == '__main__':
    uvicorn.run('run:app', host='0.0.0.0', port=8000, reload=True, debug=True, workers=1)

請求參數和驗證

接口測試：http://127.0.0.1:8000/docs

from enum import Enum
from datetime import date
from typing import Optional, List
from fastapi import APIRouter, Path, Query, Cookie, Header
from pydantic import BaseModel, Field

app03 = APIRouter()

1.路徑參數和數字驗證

"""Path Parameters and Number Validations 路徑參數和數字驗證"""


@app03.get('/path/{parameters}')  # 函數的順序就是路由的順序 {parameters}是路徑參數
def path_params01(parameters: str):
    return {"message": parameters}


@app03.get('/path/parameters')
def path_params01():
    return {"message": "This is a message"}


class CityName(str, Enum):
    Beijing = 'Beijing China'
    Shanghai = 'Shanghai China'


@app03.get('/enum/{city}')  # 枚舉類型參數
async def latest(city: CityName):
    if city == CityName.Shanghai:
        return {"city_name": city, "confirmed": 1492, "death": 7}
    if city == CityName.Beijing:
        return {"city_name": city, "confirmed": 999, "death": 9}
    return {"city_name": city, "latest": "unknown"}


@app03.get('/files/{file_path:path}')  # 通過path parameters傳遞文件路徑
def filepath(file_path: str):
    return f"The file path is {file_path}"


# 長度和正則表達式的驗證
@app03.get('/path_/{num}')
def path_params_validate(
        num: int = Path(..., ge=1, le=100, title='Your number', description='不可描述')
):
    return num

2.查詢參數和字符串驗證

"""Path Parameters and String Validations 查詢參數和字符串驗證"""


@app03.get('/query')
def page_limit(page: int = 1, limit: Optional[int] = None):  # 給了默認值就是選填參數，沒給默認值就是必填參數
    if limit:
        return {"page": page, "limit": limit}
    return {"page": page}


@app03.get('/query/bool/conversion')
def type_conversion(param: bool = False):
    return param


@app03.get('/query/validations')
def query_params_validate(
        value: str = Query(..., min_length=8, max_length=16, regex="^a"),  # Query查詢參數
        values: List[str] = Query(default=['v1', 'v2'], alias='alias_name')
):  # 多個查詢參數的列表。參數別名
    return value, values

3.請求體和字段

"""Request Body and Fields 請求體和字段"""


class CityInfo(BaseModel):
    name: str = Field(..., example="Beijing")  # Field ...代表必填字段,example是注解的作用，值不會被驗證
    country: str
    country_code: str = None  # 給一個默認值
    country_population: int = Field(default=800, title="人口數量", description="國家的人口數量", ge=800)

    class Config:
        schema_extra = {
            "example": {
                "name": "Shanghai",
                "country": "China",
                "country_code": "CN",
                "country_population": 1400000000,
            }
        }


@app03.post('/request_body/city')
def city_info(city: CityInfo):
    print(city.name, city.country)
    return city.dict()

4.多參數混合

"""Request Body + Path parameters + Query parameters 多參數混合"""


@app03.put('/request_body/city/{name}')
def mix_city_info(
        name: str,
        city01: CityInfo,
        city02: CityInfo,  # Body可以定義時多個
        confirmed: int = Query(ge=0, description="確診數", default=0),
        death: int = Query(ge=0, description="死亡數", default=0),
):
    if name == 'Shanghai':
        return {'Shanghai': {"confirmed": confirmed, "death": death}}
    return city01.dict(), city02.dict()

5.數據格式嵌套的請求體

"""Request Body - Nested Models 數據格式嵌套的請求體"""


class Data(BaseModel):
    city: List[CityInfo] = None  # 這里就是定義數據格式嵌套的請求體
    date: date  # 額外的數據類型，還有uuid，datetime，bytes，frozenset等，具體參考官方文檔
    confirmed: int = Field(ge=0, description="確診數", default=0)
    deaths: int = Field(ge=0, description="死亡數", default=0)
    recovered: int = Field(ge=0, description="痊愈數", default=0)


@app03.put('/request_body/nested')
def nested_models(data: Data):
    return data

6.Cookie 和 Header

"""Cookie 和 Header"""@app03.get('/cookie')  # 效果只能用Postman測試def cookie(cookie_id: Optional[str] = Cookie(None)):    return {'cookie_id': cookie_id}@app03.get('/header')def header(        user_agent: Optional[str] = Header(None, convert_underscores=True),        x_token: List[str] = Header(None),):    """    有些HTTP代理和服務器是不允許在請求頭中有下划線的，所有Header提供convert_underscores=True的屬性讓_變為-    :param user_agent: convert_underscores=True 會把 user_agent 變成 user-agent    :param x_token: x_token是包含多個值的列表    :return:    """    return {"User-Agent": user_agent, "x_token": x_token}

7.總結

• 在使用Pydantic定義請求體數據時，對數據進行校驗使用Field類
• 在使用路徑參數時，對數據進行校驗使用Path類
• 在使用查詢參數時，對數據進行校驗使用Query類
• 定義Cookie參數需要使用Cookie類