FastApi快速入门（二）

响应模型

基本使用

你可以在任意的路径操作中使用 response_model 参数来声明用于响应的模型：

• @app.get()
• @app.post()
• @app.put()
• @app.delete()
• 等等。

from typing import Any, List, Union

from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn

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/", response_model=Item)
async def create_item(item: Item) -> Any:
    return item


@app.get("/items/", response_model=List[Item])
async def read_items() -> Any:
    return [
        {"name": "Portal Gun", "price": 42.0},
        {"name": "Plumbus", "price": 32.0},
    ]
    
if __name__ == '__main__':
    uvicorn.run(app="main:app",host="0.0.0.0",port=8000,reload=True)

比如在更新用户信息场景中，用户提交的数据包含密码，而返回则不想包含密码，可以通过添加输出模型实现约束。

from typing import Any, Union

from fastapi import FastAPI
import uvicorn
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Union[str, None] = None


class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Union[str, None] = None


@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn) -> Any:
    return user
    
if __name__ == '__main__':
    uvicorn.run(app="main:app",host="0.0.0.0",port=8000,reload=True)

返回结果排除默认值

你可以设置路径操作装饰器的 response_model_exclude_unset=True 参数：

from typing import List, Union

from fastapi import FastAPI
import uvicorn
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: float = 10.5
    tags: List[str] = []


items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}


@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):
    return items[item_id]
    
if __name__ == '__main__':
    uvicorn.run(app="main:app",host="0.0.0.0",port=8000,reload=True)

可以看到，当 item_id 为 bax 时候，有 description 、 tax 、 tags 三个字段和默认值一致，fastapi将正确返回这些实际有的值。

包含与不包含

有时候只定义一个模型，用来作为输入和输出，这时候可以通过 response_model_include 和 response_model_exclude 来实现。

from typing import Union

from fastapi import FastAPI
import uvicorn
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: float = 10.5


items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The Bar fighters", "price": 62, "tax": 20.2},
    "baz": {
        "name": "Baz",
        "description": "There goes my baz",
        "price": 50.2,
        "tax": 10.5,
    },
}


@app.get(
    "/items/{item_id}/name",
    response_model=Item,
    response_model_include=["name", "description"],
)
async def read_item_name(item_id: str):
    return items[item_id]


@app.get("/items/{item_id}/public", response_model=Item, response_model_exclude=["tax"])
async def read_item_public_data(item_id: str):
    return items[item_id]

if __name__ == '__main__':
    uvicorn.run(app="main:app",host="0.0.0.0",port=8000,reload=True)

其中 response_model_include 代表只包含设置的字段， response_model_exclude 在所有字段中，排除设置的字段。

响应状态码

可以通过以下方法自定义返回状态码，其中 status.HTTP_201_CREATED 只是fastapi方便记忆和补全实现的功能，实际上就是一个数值 201 。

from fastapi import FastAPI, status

app = FastAPI()
import uvicorn

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    return {"name": name}
    
if __name__ == '__main__':
    uvicorn.run(app="main:app",host="0.0.0.0",port=8000,reload=True)

表单数据

接收的不是 JSON ，而是表单字段时，要使用 Form 。

要使用表单，需预先安装 python-multipart

pip install python-multipart

基本使用

from fastapi import FastAPI, Form
import uvicorn

app = FastAPI()


@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
    return {"username": username}
    
if __name__ == '__main__':
    uvicorn.run(app="main:app",host="0.0.0.0",port=8000,reload=True)

请求文件

基本使用

上传文件以「表单数据」形式发送，所以所需要的以来和 表单数据Form 一致。

from fastapi import FastAPI, File, UploadFile
import uvicorn

app = FastAPI()


@app.post("/files/")
async def create_file(file: bytes = File()):
    return {"file_size": len(file)}


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    return {"filename": file.filename}
    
if __name__ == '__main__':
    uvicorn.run(app="main:app",host="0.0.0.0",port=8000,reload=True)

文件作为「表单数据」上传。
如果把路径操作函数参数的类型声明为 bytes ，FastAPI 将以 bytes 形式读取和接收文件内容。
这种方式把文件的所有内容都存储在内存里，适用于小型文件。
不过，很多情况下，UploadFile 更好用。

可选文件

from fastapi import FastAPI, File, UploadFile
import uvicorn

app = FastAPI()


@app.post("/files/")
async def create_file(file: bytes | None = File(default=None)):
    if not file:
        return {"message": "No file sent"}
    else:
        return {"file_size": len(file)}


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile | None = None):
    if not file:
        return {"message": "No upload file sent"}
    else:
        return {"filename": file.filename}

if __name__ == '__main__':
    uvicorn.run(app="main:app",host="0.0.0.0",port=8000,reload=True)

带有额外元数据的 UploadFile

您也可以将 File() 与 UploadFile 一起使用，例如，设置额外的元数据:

from fastapi import FastAPI, File, UploadFile
import uvicorn

app = FastAPI()


@app.post("/files/")
async def create_file(file: bytes = File(description="A file read as bytes")):
    return {"file_size": len(file)}


@app.post("/uploadfile/")
async def create_upload_file(
    file: UploadFile = File(description="A file read as UploadFile"),
):
    return {"filename": file.filename}

多文件上传

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse
import uvicorn

app = FastAPI()


@app.post("/files/")
async def create_files(files: list[bytes] = File()):
    return {"file_sizes": [len(file) for file in files]}


@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile]):
    return {"filenames": [file.filename for file in files]}


@app.get("/")
async def main():
    content = """
<body>
<form action="/files/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
<form action="/uploadfiles/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
</body>
    """
    return HTMLResponse(content=content)

if __name__ == '__main__':
    uvicorn.run(app="main:app",host="0.0.0.0",port=8000,reload=True)

处理错误（自定义错误码）

使用 HTTPException

from fastapi import FastAPI, HTTPException
import uvicorn

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}


@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

if __name__ == '__main__':
    uvicorn.run(app="main:app",host="0.0.0.0",port=8000,reload=True)

配置swagger文档中接口信息

请直接参照官网地址。