FastAPI Web框架教程 第15章 补充

15-1 事件钩子

在FastAPI中，可以定义一些事件处理器，比如在服务启动时执行，或者在服务关闭时自动执行。

示例1：服务启动时的事件 startup

• 使用 app.on_event(“startup”) 装饰器注册服务启动事件，可以注册多个服务启动事件
• 服务启动后，等所有被注册的 startup事件都执行完成后，才开始接收请求
• 应用场景：初始化db，初始化外部API连接等等

from fastapi import FastAPI

app = FastAPI()

items = {}


@app.on_event("startup")
async def startup_event():
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}


@app.get("/items/{item_id}")
async def read_items(item_id: str):
    return items[item_id]

# 这些事件处理器，可以可以是通过 async def定义的协程函数，也可以是通过 def定义的普通函数。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

示例2：服务关闭事件 shutdown

• 使用 app.on_event(“shutdown”) 装饰器注册服务关闭事件，也可以注册多个

from fastapi import FastAPI

app = FastAPI()


@app.on_event("shutdown")
def shutdown_event():
    print("关机了....")


@app.get("/items/")
async def read_items():
    return [{"name": "Foo"}]
1
2
3
4
5
6
7
8
9
10
11
12
13

15-2 中间件

中间件的含义在不同的领域有不同的具体含义，但是基本差不多。比如在web应用中，客户端向服务端发送请求，然后服务端给客户端响应。那在客户端服务端之间加一层，就是中间件。

• 客户端段的请求都要经过中间件，那中间件就客户在请求来的时候做一些操作。
• 服务端的响应也要经过中间件，那中间件就可在响应数据给到客户端之前做一些操作。
• 简言之：中间件可以预处理请求，可以后处理响应。
• 并且，中间件可以有多个，在请求阶段会依次执行，在响应阶段会按照相反的顺序依次执行。

在FastAPI中使用中间件非常方便，只需要按照指定的格式编写中间件，然后注册就好了。

• 第一步：编写中间件
  • request: Request, call_next是两个必须要的参数
  • 在 await call_next(request) 之前的操作是请求阶段的操作
  • 在 await call_next(request) 之后的操作时响应阶段的操作。
  • await call_next(request) 表示获取调用下一层，来获取响应数据。因此定义的中间件函数必须是协程函数。

async def process_time(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
1
2
3
4
5
6
7
8

• 第二步：注册中间件
  • 使用 @app.middleware("http")这个装饰器注册即可

app = FastAPI()


@app.middleware("http")
async def process_time(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
1
2
3
4
5
6
7
8
9
10
11
12

示例2：注册多个中间件

• 先注册的中间件靠近应用，后注册的中间件在外层
• 所以：请求阶段：先执行后注册的中间件；响应阶段：先执行先注册的中间件。

import time

from fastapi import FastAPI, Request

app = FastAPI()



@app.middleware("http")
async def middleware1(request: Request, call_next):
    print("mid1:请求来了")
    response = await call_next(request)
    print("mid1:响应走了")
    return response


@app.middleware("http")
async def middleware2(request: Request, call_next):
    print("mid2:请求来了")
    response = await call_next(request)
    print("mid2:响应走了")
    return response



@app.get("/")
def index():
    print("处理请求")
    return {"hello": "world"}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

打印结果

mid2:请求来了
mid1:请求来了
处理请求
mid1:响应走了
mid2:响应走了
1
2
3
4
5

15-3 后台任务

需求场景

后台任务在有时是非常有用的，比如：

• 邮件通知，发邮件是一个耗时任务。用户注册后不应该等待通知邮件发出去后再得到响应，应该立即返回响应，然后在后台异步发邮件，这样可以极大提高用户的体验度。
• 处理大文件数据。比如用户上传了需要耗时的任务，正确的处理方式应该为：立即给出响应。然后再后台异步处理这个任务。

FastAPI的解决方式

• FastAPI中我们可以非常方便的使用后台任务，BackgroundTasks
• 第一步：导入BackgroundTasks
• 第二步：在路径函数中定义一个变量bgt，它的类型是BackgroundTasks
• 第三步：编写任务函数，async def或者 def都可以，FastAPI可识别并知道如何运行
• 第四步：使用 bgt.add_task()添加后台任务

注意：

• 在路径函数中定义的后台任务只会在这个接口中使用，一个接口可以定义多个后台任务。
• 编写的任务函数可以是普通函数，也可以是协程函数。

示例：

import time
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def email_notify(user: str, message: str):
    time.sleep(5)
    print(f"给 {user} 发邮邮件了：{message}")

@app.get("/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(email_notify, email, message="some notification")
    return {"message": "Notification sent in the background"}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

补充：

• 对于轻量级的后台任务，可以直接使用FastAPI的BackgroundTasks
• 但是对于重量级的异步任务、定时任务，可以使用 Celery

15-4 静态文件

当我们想要做一个静态网站服务时，可以使用FastAPI中的StaticFiles，不过它其实是从starlette中直接导出的。

示例：

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

app.mount("/", StaticFiles(directory="html"), name="static")
1
2
3
4
5
6

见素材包中两个其前端静态页面

• pet_care
• interior_design

tarlette中直接导出的。

示例：

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

app.mount("/", StaticFiles(directory="html"), name="static")
1
2
3
4
5
6

见素材包中两个其前端静态页面

• pet_care
• interior_design