背景

• 前面講了可以自定義 Response，那麼這裡就講下請求物件 Request
• 可以通過 Request 來獲取一些資料

獲取請求基礎資訊

@app.get("/base")
async def get_base(*, request: Request):
    res = {
        # 客戶端連線的 host
        "host": request.client.host,
        # 客戶端連線的埠號
        "port": request.client.port,
        # 請求方法
        "method": request.method,
        
 
# 請求路徑
        "base_url": request.base_url,
        # request headers
        "headers": request.headers,
        # request cookies
        "cookies": request.cookies
    }
    return res

請求結果

{
  "host": "127.0.0.1",
  "port": 54364,
  "method": "GET",
  "base_url": {
    "_url": "http://127.0.0.1:8080/
 
"
  },
  "headers": {
    "host": "127.0.0.1:8080",
    "connection": "keep-alive",
    "sec-ch-ua": "\"Chromium\";v=\"94\", \"Google Chrome\";v=\"94\", \";Not A Brand\";v=\"99\"",
    "accept": "application/json",
    "sec-ch-ua-mobile": "?0",
    "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.61 Safari/537.36
 
",
    "sec-ch-ua-platform": "\"macOS\"",
    "sec-fetch-site": "same-origin",
    "sec-fetch-mode": "cors",
    "sec-fetch-dest": "empty",
    "referer": "http://127.0.0.1:8080/docs",
    "accept-encoding": "gzip, deflate, br",
    "accept-language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7,zh-TW;q=0.6",
    "cookie": "test_token=tokenABC"
  },
  "cookies": {
    "test_token": "tokenABC"
  }
}

可以看到 port 並不是 8080

request.url 物件

from fastapi import Query


@app.get("/url/{item_id}")
async def get_url(*,
                  item_id: str,
                  name: str = Query(...),
                  request: Request):
    res = {
        # 請求 url
        "url": request.url,
        # 總的組成
        "components": request.url.components,
        # 請求協議
        "scheme": request.url.scheme,
        # 請求 host
        "hostname": request.url.hostname,
        # 請求埠
        "url_port": request.url.port,
        # 請求 path
        "path": request.url.path,
        # 請求查詢引數
        "query": request.url.query,
        "fragment": request.url.fragment,
        "password": request.url.password
    }
    return res

請求結果

{
  "url": {
    "_url": "http://127.0.0.1:8080/url/123?name=456",
    "_components": [
      "http",
      "127.0.0.1:8080",
      "/url/123",
      "name=456",
      ""
    ]
  },
  "components": [
    "http",
    "127.0.0.1:8080",
    "/url/123",
    "name=456",
    ""
  ],
  "scheme": "http",
  "hostname": "127.0.0.1",
  "url_port": 8080,
  "path": "/url/123",
  "query": "name=456",
  "fragment": "",
  "password": null
}

request.url 是一個物件（URL 類），得到的是一個字典

獲取路徑引數、查詢引數

@app.get("/query_path/{item_id}")
async def get_all(*,
                  item_id: str,
                  name: str = Query(...),
                  request: Request):
    res = {
        # 獲取路徑引數
        "path_params": request.path_params,
        "item_id": request.path_params.get("item_id"),
        # 獲取查詢引數
        "query_params": request.query_params,
        "name": request.query_params["name"]
    }
    return res

請求結果

{
  "path_params": {
    "item_id": "123"
  },
  "item_id": "123",
  "query_params": {
    "name": "小菠蘿"
  },
  "name": "小菠蘿"
}

path_params、query_params返回的都是字典格式的資料

獲取表單資料

@app.post("/form")
async def get_form(*,
                   username: str = Form(...),
                   oassword: str = Form(...),
                   request: Request):
    res = {
        # 獲取表單資料
        "form": await request.form()
    }
    return res

請求結果

{
  "form": {
    "username": "name",
    "oassword": "***"
  }
}

獲取 Request Body

class Item(BaseModel):
    id: str
    title: str


@app.post("/body")
async def get_body(item: Item, request: Request):
    res = {
        # 獲取 Request Body
        "body": await request.json(),
        "body_bytes": await request.body()
    }
    return res

請求結果

{
  "body": {
    "id": "string",
    "title": "string"
  },
  "body_bytes": "{\n  \"id\": \"string\",\n  \"title\": \"string\"\n}"
}

.body() 返回值型別是 bytes

獲取 Request 儲存的附加資訊

async def dep_state(request: Request):
    # 給 request 儲存附加資料
    request.state.db = "Mysql+pymysql//username"


@app.post("/state/", dependencies=[Depends(dep_state)])
async def get_state(request: Request):
    res = {
        "state": request.state,
        "db": request.state.db
    }
    return res

請求結果

{
  "state": {
    "_state": {
      "db": "Mysql+pymysql//username"
    }
  },
  "db": "Mysql+pymysql//username"
}

獲取檔案上傳資訊

from fastapi import UploadFile, File, Form


@app.post("/file")
async def get_file(*,
                   file: UploadFile = File(...),
                   name: str = Form(...),
                   request: Request):
    form_data = await  request.form()
    res = {
        # 表單資料
        "form": form_data,
        # 檔案物件 UploadFile
        "file": form_data.get("file"),
        # 檔名
        "filename": form_data.get("file").filename,
        # 檔案型別
        "content_type": form_data.get("file").content_type,
        # 檔案內容
        "file_content": await form_data.get("file").read()
    }
    return res

請求結果

{
  "form": {
    "file": {
      "filename": "test.txt",
      "content_type": "text/plain",
      "file": {}
    },
    "name": "表單name！！"
  },
  "file": {
    "filename": "test.txt",
    "content_type": "text/plain",
    "file": {}
  },
  "filename": "test.txt",
  "content_type": "text/plain",
  "file_content": "hello world"
}

UploadFile 物件

• form_data.get("file")返回的是一個starlette.datastructures.UploadFile物件
• filename、content_type 是物件例項屬性
• .read() 是例項方法，讀取檔案

UploadFile 的其他非同步方法

和 Python 普通的檔案物件方法一樣，只不過都是 async 非同步的

• write
• read
• seek
• close