这里主要介绍fastapi
的Input/Output,或者说Request/Response相关内容,这也是用好fastapi
的关键了。
有兴趣的读者,可以详细的查看:learn fastapi。
Input
我们知道,HTTP支持 POST/GET/PUT/DELETE
。可以使用 @app.post(), @app.get(), @app.put(), @app.delete()
来定义。
关于Path参数
- 基础使用
参数可以在path中声明:
@app.get("/items/{item_id}")
async def read_item(item_id):
return {"item_id": item_id}
还可以给定类型标签,如: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):
......
- 参数带
/
路径的场景
比如有一个文件的路径是 /home/john/myfile.txt
,那我们如果这么传递是有问题的:/files/home/johndoe/myfile.txt
。
可以这样处理:
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
return {"file_path": file_path}
然后传递参数用:/files//home/johndoe/myfile.txt
,注意是 //
不是 /
。
关于Query参数
对于非路径中的参数,如下所示 http://127.0.0.1:8000/items/?skip=0&limit=10
:
其中query参数包括了:skip
,limit
。也可以在参数中来进行定义:
@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
return fake_items_db[skip : skip + limit]
有一点不同的是可选参数:
@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
可选。
可以这么理解:如果有默认值,则表示可选;如果没有默认值,则为必选,如果必选不满足会报错。
请求Body
请求body会用到Pydantic,发送请求body也需要使用POST
,PUT
,DELETE
,PATCH
。
使用BaseModel
,需要从Pydantic
导入。
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() # 转化为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在很多场景下都有使用,比如当用户登录成功后,我们在下次登录时就自动填入用户名和密码。这需要我们使用cookie。cookie一般使用的是key-value
来进行存储的。
在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}
- 关于类型
typing 是Python标准库中的一个模块,支持类型提示(在代码中指定变量、函数参数、返回值的类型的方法)。
Annotated是一个类型提示,Annotated[T, x] 用元数据 x 注解类型 T,比如:
Annotated[int, ValueRange(3, 10), ctype("char")]
Annotated 的第一个参数必须是有效类型,Annotated 至少需要两个参数。
Header
需要导入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会将变量里面的_
转化为-
来进行匹配。
另外,可以支持多个重名的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
# 也可以使用如下导入 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),
]
这么做,没有什么太大的问题,不过有时候我们想做一些转换的操作,比如:
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,会对返回的数据进行转换。注意,上面故意定义了return type为Any
。
返回码(Response Status Code)
我们还可以为上面的操作定义返回码:
- @app.get()
- @app.post()
- @app.put()
- @app.delete()
默认的返回码方式如下:
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]
总结
上面我们只是简单的对fastapi
的一些功能做了介绍,用户可以参考文档再详细了解。