10-04 06:35 阅读 374

关注

FastAPI（54）- 详解 Request 请求对象

背景

• 前面讲了可以自定义 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

来源 https://www.cnblogs.com/poloyy/p/15366904.html