如何在FastAPI中使用Lifespan管理连接池：兼顾生命周期与依赖注入-Python教程-PHP中文网

如何在FastAPI中使用Lifespan管理连接池：兼顾生命周期与依赖注入

聖光之護

发布： 2025-12-13 20:16:29

原创

789人浏览过

如何在fastapi中使用lifespan管理连接池：兼顾生命周期与依赖注入

本文详细阐述了如何在FastAPI应用中，利用其生命周期管理特性（`lifespan`）高效地创建和释放外部资源，如数据库连接池。通过将连接池的初始化和关闭与应用的启动和关闭事件绑定，并结合FastAPI强大的依赖注入系统，实现了在保持代码整洁和便利性的同时，对资源进行全局且优雅的管理。

引言：FastAPI应用中的资源管理挑战

在构建高性能的Web服务时，有效地管理外部资源（如数据库连接、Redis连接池、消息队列客户端等）是至关重要的。这些资源通常需要在应用程序启动时进行初始化，并在应用程序关闭时进行清理，以避免资源泄露和性能问题。然而，如果直接在模块级别创建这些资源，它们会在应用启动之前被初始化，脱离了FastAPI的生命周期管理。同时，开发者也希望能够利用FastAPI强大的依赖注入系统，以简洁优雅的方式在各个API端点中访问这些资源，避免在每个请求处理函数中手动获取或传递。

FastAPI提供了lifespan机制（在旧版本中是on_event），专门用于处理应用程序的启动和关闭事件，这为我们管理全局资源提供了理想的解决方案。

FastAPI的生命周期管理机制

lifespan是FastAPI中一个强大的功能，它允许开发者定义在应用程序启动和关闭时执行的异步代码。它通过一个异步上下文管理器（asynccontextmanager）来实现，确保资源在应用整个生命周期内得到妥善管理。

使用lifespan的主要优势在于：

集中管理：所有全局资源的初始化和清理逻辑都集中在一个地方。
按需启动/关闭：资源只在应用程序真正启动时创建，并在关闭时释放。
优雅停机：确保在应用程序关闭前，所有正在使用的资源都能被安全地关闭，避免数据丢失或连接泄露。

使用lifespan管理连接池

为了在FastAPI中优雅地管理连接池并结合依赖注入，我们通常会采取以下步骤：

定义lifespan上下文管理器：在这个管理器中，我们将在应用启动时创建连接池，并将其存储在app.state对象上，以便全局访问。在应用关闭时，我们将负责关闭这个连接池。
创建获取连接池的依赖函数：这个函数将从app.state中获取已创建的连接池。
创建获取单个连接的依赖函数：这个函数将利用上一步获取的连接池，为每个请求提供一个独立的连接，并在请求结束后将连接返回到池中。

下面，我们将通过一个使用asyncpg（一个异步PostgreSQL客户端库）管理数据库连接池的完整示例来详细说明。

完整示例：PostgreSQL连接池管理

假设我们有一个FastAPI应用，需要连接到PostgreSQL数据库。我们将使用lifespan来管理asyncpg的连接池。

示例代码

首先，确保你已经安装了必要的库：

N世界

一分钟搭建会展元宇宙

138

查看详情

pip install fastapi uvicorn asyncpg

登录后复制

然后，创建以下Python文件（例如 main.py）：

from contextlib import asynccontextmanager
import asyncpg
from fastapi import FastAPI, Depends, Request
from typing import AsyncGenerator

# 数据库连接配置
DATABASE_URL = "postgresql://user:password@localhost/dbname" # 请替换为你的实际数据库URL

# 1. 定义lifespan上下文管理器
@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    FastAPI应用程序的生命周期管理。
    在启动时创建数据库连接池，在关闭时关闭连接池。
    """
    print("应用程序启动：正在创建数据库连接池...")
    try:
        app.state.db_pool = await asyncpg.create_pool(DATABASE_URL, min_size=1, max_size=10)
        print("数据库连接池创建成功。")
    except Exception as e:
        print(f"数据库连接池创建失败: {e}")
        # 可以在这里选择退出应用或进行其他错误处理
        raise

    yield  # 应用程序在此处运行

    print("应用程序关闭：正在关闭数据库连接池...")
    if hasattr(app.state, 'db_pool') and app.state.db_pool:
        await app.state.db_pool.close()
        await app.state.db_pool.wait_closed() # 确保连接池完全关闭
        print("数据库连接池已关闭。")

# 初始化FastAPI应用，并传入lifespan
app = FastAPI(lifespan=lifespan)

# 2. 创建获取连接池的依赖函数
async def get_db_pool(request: Request) -> asyncpg.pool.Pool:
    """
    依赖注入函数，用于从应用程序状态中获取数据库连接池。
    """
    if not hasattr(request.app.state, 'db_pool') or not request.app.state.db_pool:
        raise RuntimeError("数据库连接池未初始化或已关闭。")
    return request.app.state.db_pool

# 3. 创建获取单个连接的依赖函数
async def get_db_connection(
    pool: asyncpg.pool.Pool = Depends(get_db_pool)
) -> AsyncGenerator[asyncpg.connection.Connection, None]:
    """
    依赖注入函数，用于从连接池中获取一个数据库连接。
    使用yield确保连接在请求结束后返回到池中。
    """
    async with pool.acquire() as connection:
        yield connection

# 示例路由
@app.get("/items/{item_id}")
async def read_item(item_id: int, db: asyncpg.connection.Connection = Depends(get_db_connection)):
    """
    一个示例API端点，演示如何使用依赖注入获取数据库连接。
    """
    # 假设这里有一个items表
    # result = await db.fetchrow("SELECT name FROM items WHERE id = $1", item_id)
    # if result:
    #     return {"item_id": item_id, "name": result["name"]}
    # return {"message": "Item not found"}

    # 为了演示，我们直接返回item_id
    print(f"在请求中使用了数据库连接: {db}")
    return {"item_id": item_id, "message": "Database connection successfully acquired and used."}

@app.get("/status")
async def get_status():
    """
    检查应用状态，确保连接池已初始化。
    """
    if hasattr(app.state, 'db_pool') and app.state.db_pool:
        return {"status": "running", "db_pool_initialized": True}
    return {"status": "running", "db_pool_initialized": False, "message": "DB pool not available"}

登录后复制

代码解析

lifespan 函数：
- 使用@asynccontextmanager装饰器定义，使其成为一个异步上下文管理器。
- 在yield之前，我们调用asyncpg.create_pool()创建数据库连接池，并将其赋值给app.state.db_pool。app.state是FastAPI应用程序对象的一个属性，可以用来存储任何应用程序级别的状态或资源。
- yield语句将控制权交给FastAPI，应用程序开始处理请求。
- 在yield之后（即应用程序关闭时），我们调用app.state.db_pool.close()和app.state.db_pool.wait_closed()来安全地关闭连接池。
get_db_pool 依赖函数：
- 这个函数接收Request对象作为参数，这是FastAPI提供的，用于访问当前请求的上下文。
- 通过request.app.state.db_pool，我们可以获取在lifespan中创建并存储的全局连接池实例。
- 这个函数的作用是将全局连接池“注入”到其他依赖中，避免直接在每个依赖或端点中访问request.app.state，从而保持接口的清洁。
get_db_connection 依赖函数：
- 这个函数声明它依赖于get_db_pool，因此FastAPI会自动为它提供连接池实例。
- 使用async with pool.acquire() as connection:，它从连接池中获取一个数据库连接。
- yield connection将这个连接提供给实际的路由处理函数。
- 当路由处理函数执行完毕后，async with块会确保连接被自动释放并返回到连接池中，无需手动调用connection.release()。
API 端点 (read_item)：
- 路由处理函数只需声明它依赖于get_db_connection，FastAPI就会自动为它提供一个可用的数据库连接。
- 开发者可以在函数内部直接使用db对象执行数据库操作，无需关心连接的获取、释放或连接池的管理细节。

运行应用

使用Uvicorn启动FastAPI应用：

uvicorn main:app --reload

登录后复制

当应用启动时，你将看到控制台输出连接池创建成功的消息。访问/items/123或/status端点，会看到请求成功处理。当关闭Uvicorn进程（通常是Ctrl+C）时，你将看到连接池关闭的消息。

优势与注意事项

优势

资源生命周期与应用同步：连接池的创建和销毁与FastAPI应用的启动和关闭严格同步，避免了资源过早创建或泄露。
依赖注入的便利性：通过Depends机制，API端点可以简洁地声明对数据库连接的需求，而无需关心底层连接池的实现细节。
代码解耦与可维护性：连接池的管理逻辑与业务逻辑分离，提高了代码的可读性和可维护性。
性能优化：连接池避免了为每个请求频繁地创建和关闭数据库连接，显著提升了性能。

注意事项

错误处理：在lifespan中创建资源时，务必添加适当的错误处理机制。如果资源创建失败，应考虑如何处理（例如，抛出异常阻止应用启动，或记录错误并以降级模式运行）。
资源清理：确保在lifespan的yield之后，所有已创建的资源都能被正确、彻底地关闭。对于某些库，可能需要调用wait_closed()等方法确保资源完全释放。
不同库的差异：虽然原理相同，但不同数据库驱动或资源管理库的API可能有所不同（例如，Redis的redis.asyncio库可能与asyncpg有细微差异），需要根据具体库的文档进行调整。
测试：在编写集成测试时，需要考虑如何模拟或管理这些生命周期内的资源，确保测试环境的隔离性。

总结

通过巧妙地结合FastAPI的lifespan生命周期管理和依赖注入系统，我们可以优雅且高效地管理应用程序中的全局资源，如数据库连接池。这种模式不仅确保了资源在整个应用生命周期内的正确初始化和清理，还极大地简化了API端点对这些资源的访问，提升了代码的质量和开发效率。掌握这一模式，是构建健壮、高性能FastAPI应用的关键一步。

以上就是如何在FastAPI中使用Lifespan管理连接池：兼顾生命周期与依赖注入的详细内容，更多请关注php中文网其它相关文章！

大家都在看：

Python如何自动生成考试试卷？从Word文档表格中提取带编号列表的教程 Python docx 库：从Word文档表格中提取带编号列表的教程使用Python提取Word文档表格中带编号列表的文本使用Python docx从Word文档中提取表格内的编号列表