Python Web框架-FastAPI教程-生命周期asynccontextmanager + lifespan-马育民老师

# 提出问题

在 FastAPI 开发后端项目时，数据库是必备资源。很多数新手会遇到以下 **3 个致命问题**：

1. **重复创建连接**：每请求一次创建一次数据库连接，数据库压力巨大，极易卡死。
  
2. **连接无法关闭**：服务关停后，数据库连接残留，产生大量死连接，占用数据库连接池。
  
3. **代码散乱难维护**：连接初始化、关闭代码分散在不同文件，后期无法统一管理。

### 错误示范：新手最常写的代码

下面是典型错误写法：接口内部直接创建数据库连接。

```python
@app.get("/user")
async def get_user():
    # 每来一次请求，新建一次连接（极度错误）
    db = AsyncDB()
    await db.connect()
    return {"data": "用户数据"}
    # 没有手动关闭，连接泄漏
```

### 旧方案缺陷：startup/shutdown（淘汰）

有人会使用旧生命周期写法，但是依旧存在缺陷：

```python
@app.on_event("startup")
async def startup():
    # 启动连接
    global db
    db = AsyncDB()
    await db.connect()

@app.on_event("shutdown")
async def shutdown():
    # 关闭连接
    await db.close()
```

### 旧方案存在的问题

- 初始化和关闭**代码分离**，可读性差，维护麻烦
  
- 依赖全局变量，多线程、多进程下不安全
  
- 没有上下文约束，极端情况下依旧资源泄漏

# 解决方案：Lifespan 生命周期

### 什么是 lifespan？

**lifespan** 是 FastAPI 官方推出的应用生命周期管理器，专门用于解决：**数据库、Redis、MQ 等全局资源统一初始化与销毁**。

它把「启动代码」和「关闭代码」写在同一个函数内，天然形成上下文，保证资源成对释放，彻底解决连接泄漏问题。

### 什么是 asynccontextmanager？

`asynccontextmanager` 是 Python 内置库 `contextlib` 提供的**异步上下文管理器装饰器**。

作用：固定格式拆分 **前置代码（启动）** 和 **后置代码（销毁）**，适配 lifespan。

### 语法格式

```python
from contextlib import asynccontextmanager

@asynccontextmanager
async def demo():
    # 前置逻辑：应用启动时执行（创建数据库连接）
    yield
    # 后置逻辑：应用关闭时执行（断开数据库连接）
```

### 执行规则

- **yield 之前：启动代码**
  
- **yield 本身：程序运行中**
  
- **yield 之后：关闭代码**

# 例子

这是官方唯一推荐写法，所有项目通用。

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # ========== 启动阶段：创建数据库连接 ==========
    print("应用启动，准备连接数据库")

yield

# ========== 关闭阶段：断开数据库连接 ==========
    print("应用关闭，释放数据库连接")

app = FastAPI(lifespan=lifespan)
```

### 执行顺序

1. 运行 uvicorn，优先执行 yield 上方代码：连接数据库
  
2. 服务启动，监听端口，接收请求，复用同一个数据库连接
  
3. 按下 Ctrl\+C 关闭服务，执行 yield 下方代码：断开连接

# 案例

数据库全局资源管理

### 工程规范：资源挂载到 app.state

禁止使用普通全局变量。FastAPI 官方规范：所有全局资源挂载至 `app\.state`，隔离性更强、多线程安全，专门用来存放数据库、Redis 等连接实例。

### 代码

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio

# 模拟异步数据库连接（真实项目替换为 XX连接池XX）
class AsyncDB:
    async def connect(self):
        await asyncio.sleep(0.2)
        print("✅ 数据库连接成功")

async def close(self):
        await asyncio.sleep(0.2)
        print("❌ 数据库连接关闭")

# 生命周期管理：统一管控数据库
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 【启动】初始化数据库连接（真实项目是连接池）
    db = AsyncDB()
    await db.connect()
    # 挂载到全局，所有接口共享
    app.state.db = db

yield

# 【关闭】优雅释放数据库连接
    await app.state.db.close()

app = FastAPI(lifespan=lifespan)

# 在接口中使用全局数据库连接
@app.get("/user")
async def get_user():
    # 取出全局数据库实例
    db = app.state.db
    return {"code": 200, "msg": "复用全局数据库连接，无重复创建"}
```

# 新旧写法对比

| 对比项 | lifespan（新版） | startup/shutdown（旧版） |
| --- | --- | --- |
| 代码位置 | 数据库连接/关闭写在一起 | 分散两处，容易遗漏关闭 |
| 连接安全性 | 上下文强制释放，无死连接 | 异常退出容易残留连接 |
| 变量使用 | app\.state 安全挂载 | 依赖全局变量，线程不安全 |
| 官方状态 | 持续维护、主推 | 0\.95\+ 标记废弃 |

# 易错点总结

### 语法易错

- **必须写 yield**：缺少 yield，程序无法切换运行状态，数据库卡死
  
- **必须加 async**：数据库异步连接，lifespan 强制异步函数
  
- **异步操作必须 await**：connect、close 必须加 await

### 工程规范易错

- **禁止** 在 lifespan 写 **同步阻塞代码**，会阻塞数据库初始化
  
- 不要用全局变量存储数据库实例，统一挂载 app.state
  
- 所有初始化资源，必须在关闭阶段手动释放，防止连接堆积

# 阻塞同步代码

将 **阻塞函数名** 写到下面代码中

**注意：** **阻塞函数名** **后面不要加** **()**，否则表示直接调用

```
await asyncio.get_event_loop().run_in_executor(None, 阻塞函数名)
```

### 例子

```
await asyncio.get_event_loop().run_in_executor(None, init)
```

**解释：** `init` 是阻塞同步函数名，后面 **必须不能加** `()`，否则表示直接调用

# 总结

- 数据库怕泄漏，连接要统一；
- 装饰器修饰、函数必须异步；
- yield 分割、前连后闭；
- 资源挂载 state、严禁全局；
- 新项目 lifespan、抛弃 startup。

原文出处：http://www.malaoshi.top/show_1GW3HX1SfD8w.html