@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
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": "***"
}
}
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
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"
}
form_data.get("file")
和 Python 普通的文件对象方法一样,只不过都是 async 异步的