Python Web框架-FastAPI教程-生命周期asynccontextmanager + lifespan

作者：马育民 • • 阅读：10052

提出问题

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

  1. 重复创建连接：每请求一次创建一次数据库连接，数据库压力巨大，极易卡死。

  2. 连接无法关闭：服务关停后，数据库连接残留，产生大量死连接，占用数据库连接池。

  3. 代码散乱难维护：连接初始化、关闭代码分散在不同文件，后期无法统一管理。

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

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

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

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

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

@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。

语法格式

from contextlib import asynccontextmanager

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

执行规则

  • yield 之前：启动代码

  • yield 本身：程序运行中

  • yield 之后：关闭代码

例子

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

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 等连接实例。

代码

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。

原文出处：/show_1GW3HX1SfD8w.html

吉ICP备15003043号