Sunny was born

FastAPI 从入门到落地

本文是基于官方文档精简化提取, 供快速上手使用, 如需完整学习请参阅官方文档<(￣︶￣)

Created | Updated Word count: 10k | Reading time: ≈ 9 min
0%

FastAPI 是一个现代、快速（高性能）的 Web 框架.

官方文档 -> 🚪
项目源码 -> 🚪
Uvicorn -> 🚪

环境准备

pip install fastapi uvicorn[standard]
  • uvicorn: ASGI(异步)服务器

基础例子

创建main.py文件

from typing import Optional

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None):
    return {"item_id": item_id, "q": q}

关于函数是否使用async?

文档参考 -> 🚪

如果不清楚函数内调用是否异步, 那就定义为普通函数, Fastapi会放到thread pool里执行; 如果有使用异步并且是在高并发情景下, 使用async速度会更快.

运行启动

命令行启动

uvicorn main:app --reload
  • main: main.py 文件（一个 Python "模块"）。
  • app: 在 main.py 文件中通过 app = FastAPI() 创建的对象。
  • --reload: 让服务器在更新代码后重新启动。仅在开发时使用该选项。
  • --host: 设置映射IP, 默认127.0.0.1
  • --port: 设置映射端口, 默认8000

访问 http://127.0.0.1:8000/items/5?q=somequery 会返回 {"item_id": 5, "q": "somequery"}

程序内启动

可以用于调试

import uvicorn

# body...

if __name__ == '__main__':
    uvicorn.run(app, host="127.0.0.1", port=8000)

自带文档

  • 访问 http://127.0.0.1:8000/docs 是由Swagger UI自动生成的交互式文档
  • 访问 http://127.0.0.1:8000/redoc 是由ReDoc 生成自动生成的交互式文档

设置枚举类型

导入Enum

from enum import Enum

继承并声明枚举类

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

将其声明为参数

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    if model_name == ModelName.alexnet:
        return {"model_name": model_name, "message": "Deep Learning FTW!"}

    if model_name.value == "lenet":
        return {"model_name": model_name, "message": "LeCNN all the images"}
    return {"model_name": model_name, "message": "Have some residuals"}

设置请求体

基础应用

导入BaseModel

from pydantic import BaseModel

继承并声明数据模型类

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

将其声明为参数

@app.post("/items/")
def create_item(item: Item):
    return item

请求格式应为

{
  "name": "Foo",
  "description": "An optional description",
  "price": 45.2,
  "tax": 3.5
}

多个请求体参数

class User(BaseModel):
    username: str
    full_name: Optional[str] = None
    
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
    results = {"item_id": item_id, "item": item, "user": user}
    return results

请求格式应为

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    }
}

单个请求体参数

导入Body

from fastapi import Body

使用Body()设置

@app.post("/items/{item_id}")
async def update_item(item_id: int, item: Item = Body(..., embed=True)):
    results = {"item_id": item_id, "item": item}
    return results

请求格式应为

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}

嵌套格式请求体

class Image(BaseModel):
    url: HttpUrl
    name: str

class Item(BaseModel):
    image: Optional[Image] = None

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results

请求格式应为

{
    "images": [
        {
            "url": "http://example.com/baz.jpg",
            "name": "The Foo live"
        },
        {
            "url": "http://example.com/dave.jpg",
            "name": "The Baz"
        }
    ]
}

为参数添加校验

查询参数校验

导入Query

from fastapi import Query

将 Query 用作查询参数的默认值，并设置min_length为3,max_length为9

@app.get("/items/")
# Query 的第一个参数用于定义默认值
async def read_items(q: Optional[str] = Query(None, min_length=3, max_length=9)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

其它校验参数(对应类型为int或float)

  • gt：大于（greater than）
  • ge：大于等于（greater than or equal）
  • lt：小于（less than）
  • le：小于等于（less than or equal）

也可以用正则表达式进行判断regex

Query(None, regex=r".{3,9}")

测试一下

  • http://127.0.0.1:8000/items/?q=12
  • http://127.0.0.1:8000/items/?q=123456
  • http://127.0.0.1:8000/items/?q=1234567890

使用参数别名

在Query中使用alias, 设置q别名为item-query

@app.get('/items/')
async def read_items(q: Optional[str] = Query(None, alias="item-query")):
    return {'q': q}

测试一下

  • http://127.0.0.1:8000/items?q=hello
  • http://127.0.0.1:8000/items?item-query=hello

路径参数校验

导入Path

from fastapi import Path

使用方法和Query一样, 第一个参数为默认参数, 使用...表示该参数为必需参数

@app.get("/items/{item_id}")
async def read_items(
    *,
    item_id: int = Path(..., title="The ID of the item to get", ge=0, le=1000),
    size: float = Query(..., gt=0, lt=10.5)
):
    return {"item_id": item_id}

测试一下

  • http://127.0.0.1:8000/items/1000?size=1.2
  • http://127.0.0.1:8000/items/100

请求体字段校验

官方参考 -> 🚪

注: Body、Query、Path的大部分参数是一样的

获取Header和Cookie

导入

from fastapi import Header, Cookie

声明

@app.get("/items/")
def read_items(user_agent: Optional[str] = Header(None), ads_id: Optional[str] = Cookie(None):
    return {
        "User-Agent": user_agent,
    	"ads_id": ads_id
    }

响应和输出

设置输出模型

@app设置response_model参数, 

from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

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

# 输出模型不包含密码
class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Optional[str] = None

# 设置对应输出模型
@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):
    return user

设置响应不包含默认值

@app设置response_model_exclude_unset参数为True, 不想获取过多Json可以用的上.

from typing import List, Optional

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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

items = {
    "foo": {"name": "Foo", "price": 50.2}
}

@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):
    return items[item_id]

响应为

{
    "name": "Foo",
    "price": 50.2
}

而不是

{
  "name": "Foo",
  "description": null,
  "price": 50.2,
  "tax": 10.5,
  "tags": []
}

设置响应状态码

@app设置参数status_code, 响应时会返回对应状态码, 不想暴露真实状态码时会很有用.

from fastapi import FastAPI, status

app = FastAPI()

# 使用status自带的变量映射,或者直接使用数字
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    return {"name": name}

PS: 到这里, 简单的API应用就没问题了, 接下来是一些进阶技巧

中间件

"中间件"是一个函数,它在每个请求被特定的路径操作处理之前,以及在每个响应返回之前工作.

使用装饰器 @app.middleware("http")来添加中间件

  • 第一个参数: request
  • 第二个参数: 一个函数 call_next (函数名可自定义)
    • 它将接收 request 作为参数
    • 将 request 传递给相应的 路径操作
    • 然后它将返回由相应的 路径操作 生成的 response.
import time

from fastapi import FastAPI, Request

app = FastAPI()

@app.get("/")
async def test(num: int):
    return {"num": num}

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):

    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    
    # 添加回调时间
    response.headers["X-Process-Time"] = str(process_time)
    return response

@app.middleware("http")
async def set_status_code_eq_200(request: Request, call_next):
    print('set code eq 200')
    
    response = await call_next(request)
    
	# 修改所有响应状态码一直为200
    response.status_code = 200
    return response

if __name__ == '__main__':
    uvicorn.run(app, host="127.0.0.1", port=8000)

使用调试模式测试一下, 顺便观察下中间件的执行顺序

  • http://127.0.0.1:8000/?num=22
  • http://127.0.0.1:8000/?num=str

执行顺序是从下往上先执行call_next前的代码,然后执行call_next后, 再从上往下执行call_next后的部分, 可以看成栈的操作顺序, 先入后出, 后入先出.

跨域资源共享

指浏览器中运行的前端拥有与后端通信的 JavaScript 代码，而后端处于与前端不同的「源」的情况.

源是协议（http，https）、域（myapp.com，localhost，localhost.tiangolo.com）以及端口（80、443、8080）的组合。

因此，这些都是不同的源：

  • http://localhost
  • https://localhost
  • http://localhost:8080

即使它们都在 localhost 中，但是它们使用不同的协议或者端口，所以它们都是不同的「源」

from fastapi import FastAPI
# 导入 CORSMiddleware
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# 创建一个允许的源列表（由字符串组成）
origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

# 将其作为「中间件」添加到你的 FastAPI 应用中
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,	# 允许凭证（授权 headers，Cookies 等）
    allow_methods=["*"],  # 特定的 HTTP 方法（POST，PUT）或者使用通配符 "*" 允许所有方法
    allow_headers=["*"],  # 特定的 HTTP headers 或者使用通配符 "*" 允许所有 headers。
)

@app.get("/")
async def main():
    return {"message": "Hello World"}

其它应用 -> 🚪

创建后台任务

使用场景: 邮件发送, 图片渲染下载...

使用BackgroundTasks

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    
    # 先方法名后传参
    background_tasks.add_task(write_notification, email, message="some notification")
    
    return {"message": "Notification sent in the background"}

其它功能

以下部分属于比较少用到或者是知识量较多需要深入学习的, 故只提供官方链接以供参考

  • 使用其它数据类型
  • 设置请求文件
  • 错误异常处理
  • 请求体更新数据
  • 使用关系数据库
------------ 已触及底线了 感谢您的阅读 ------------
  • 本文作者： OWQ
  • 本文链接： https://www.owq.world/1a0a2507/
  • 版权声明： 本站所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处（￣︶￣）↗