这里主要介绍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的一些功能做了介绍，用户可以参考文档再详细了解。
