FastAPI/Starlette的依赖注入系统：在ASGI生命周期内管理请求级资源 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

FastAPI/Starlette 依赖注入：ASGI 生命周期内的请求级资源管理

大家好，今天我们要深入探讨 FastAPI 和 Starlette 中强大的依赖注入系统，并重点关注如何在 ASGI 生命周期内管理请求级资源。理解这一点对于构建健壮、可维护且可测试的 Web 应用至关重要。

什么是依赖注入？

首先，让我们回顾一下依赖注入的基本概念。简单来说，依赖注入是一种设计模式，它允许我们从组件自身中移除创建和管理其依赖项的责任。取而代之的是，依赖项由外部“注入”到组件中。这带来了几个重要的好处：

松耦合： 组件不再直接依赖于具体的实现，而是依赖于接口或抽象。这降低了组件之间的耦合度，使得修改和替换组件变得更加容易。
可测试性： 由于依赖项可以被轻松地替换为模拟对象，因此单元测试变得更加容易。
可重用性： 组件可以在不同的上下文中使用，只需注入不同的依赖项即可。
可维护性： 代码变得更清晰、更易于理解和维护。

FastAPI/Starlette 的依赖注入系统

FastAPI 和 Starlette 都内置了一个强大的依赖注入系统，它允许我们以声明式的方式定义和管理应用程序的依赖项。核心概念是依赖项和依赖项解析器。

依赖项 (Dependencies): 依赖项可以是任何 Python 可调用对象，例如函数、类或方法。它们负责提供组件所需的资源或服务。
依赖项解析器 (Dependency Resolver): FastAPI/Starlette 的依赖项解析器负责在运行时解析依赖项之间的依赖关系，并确保它们以正确的顺序创建和注入。

在 FastAPI 中，我们使用 Depends() 函数来声明依赖项。例如：

from fastapi import FastAPI, Depends

app = FastAPI()

async def get_db():
    db = "Database Connection" # 模拟数据库连接
    try:
        yield db
    finally:
        # 关闭数据库连接
        print("Closing database connection")

async def get_current_user(db: str = Depends(get_db)):
    # 这里应该从数据库中获取用户信息，但是为了演示，我们直接返回一个字符串
    return "current_user"

@app.get("/items/")
async def read_items(current_user: str = Depends(get_current_user)):
    return {"message": f"Hello {current_user}, you have accessed items"}

在这个例子中，get_db 和 get_current_user 都是依赖项。read_items 函数依赖于 get_current_user，而 get_current_user 又依赖于 get_db。 FastAPI 会自动解析这些依赖关系，并在调用 read_items 之前创建并注入所需的资源。

ASGI 生命周期与请求级资源

ASGI (Asynchronous Server Gateway Interface) 是一个用于构建异步 Python Web 应用的标准接口。理解 ASGI 生命周期对于有效地管理请求级资源至关重要。

一个典型的 ASGI 请求处理流程如下：

连接建立 (Connection): 客户端与服务器建立连接。
请求接收 (Request): 服务器接收客户端的请求。
请求处理 (Request Handling): 服务器处理请求，包括路由、中间件、依赖项解析等。
响应发送 (Response): 服务器向客户端发送响应。
连接关闭 (Disconnection): 客户端与服务器关闭连接。

请求级资源是指在单个请求的生命周期内创建和使用的资源，例如数据库连接、事务、缓存客户端等。有效地管理这些资源对于确保应用程序的性能、可靠性和安全性至关重要。

在 ASGI 生命周期内管理请求级资源

FastAPI/Starlette 的依赖注入系统提供了一种优雅的方式来管理请求级资源。关键在于利用 yield 语句来实现依赖项的生命周期管理。

当依赖项函数包含 yield 语句时，它会变成一个生成器函数。在这种情况下，FastAPI/Starlette 会在请求开始时执行 yield 之前的代码，将 yield 的值作为依赖项注入到请求处理函数中，然后在请求完成后执行 yield 之后的代码。这允许我们确保资源在请求开始时被创建，并在请求结束时被释放。

让我们回到 get_db 的例子：

async def get_db():
    db = "Database Connection" # 模拟数据库连接
    try:
        yield db
    finally:
        # 关闭数据库连接
        print("Closing database connection")

在这个例子中，yield db 之前的代码可以用来建立数据库连接，yield db 之后的代码可以用来关闭数据库连接。 FastAPI 会确保在调用依赖于 get_db 的函数之前建立数据库连接，并在函数执行完毕后关闭数据库连接。

这种模式非常强大，因为它可以确保资源在使用后被及时释放，避免资源泄漏和性能问题。

中间件与依赖注入的交互

中间件是 ASGI 应用程序中的一个重要组成部分，它们可以拦截和处理请求和响应。中间件也可以使用依赖注入，但需要注意一些细节。

中间件接收三个参数：

app: 下一个中间件或最终的请求处理函数。
scope: 包含有关请求的信息的字典。
receive: 一个异步函数，用于接收请求体。
send: 一个异步函数，用于发送响应。

要在中间件中使用依赖注入，我们可以使用 request.state 对象来存储依赖项。例如：

from fastapi import FastAPI, Request
from starlette.middleware import Middleware
from starlette.responses import Response
from typing import Callable
from fastapi import Depends

app = FastAPI()

async def get_db():
    db = "Database Connection" # 模拟数据库连接
    try:
        yield db
    finally:
        # 关闭数据库连接
        print("Closing database connection")

async def add_db_to_request(request: Request, call_next):
    db = await get_db().__anext__() # 手动调用生成器，获取db
    request.state.db = db
    response = await call_next(request)

    try:
        await get_db().__anext__() # 运行finally代码
    except StopAsyncIteration:
        pass # 忽略 StopAsyncIteration 异常
    return response

middleware = [
    Middleware(add_db_to_request)
]

app = FastAPI(middleware=middleware)

@app.get("/items/")
async def read_items(request: Request):
    db = request.state.db
    return {"message": f"Database connection: {db}"}

在这个例子中，add_db_to_request 中间件使用 get_db 依赖项来获取数据库连接，并将其存储在 request.state.db 中。然后，请求处理函数可以从 request.state 中访问数据库连接。

注意: 在中间件中使用 yield 生成器时，需要手动调用 __anext__() 来启动和结束生成器。并且需要捕获 StopAsyncIteration 异常，以防止程序崩溃。

依赖注入与测试

依赖注入的一个重要好处是它使得单元测试变得更加容易。由于我们可以轻松地替换依赖项为模拟对象，因此我们可以独立地测试组件，而无需依赖于外部资源。

例如，我们可以创建一个模拟的数据库连接，并在单元测试中使用它来测试 read_items 函数：

import pytest
from fastapi.testclient import TestClient
from fastapi import FastAPI, Depends, Request
from starlette.middleware import Middleware
from starlette.responses import Response
from typing import Callable

app = FastAPI()

async def get_db():
    return "Mock Database Connection" # 模拟数据库连接

async def add_db_to_request(request: Request, call_next):
    request.state.db = await get_db()
    response = await call_next(request)
    return response

middleware = [
    Middleware(add_db_to_request)
]

app = FastAPI(middleware=middleware)

@app.get("/items/")
async def read_items(request: Request):
    db = request.state.db
    return {"message": f"Database connection: {db}"}

client = TestClient(app)

def test_read_items():
    response = client.get("/items/")
    assert response.status_code == 200
    assert response.json() == {"message": "Database connection: Mock Database Connection"}

在这个例子中，我们创建了一个模拟的 get_db 函数，它返回一个模拟的数据库连接。然后，我们在单元测试中使用这个模拟的 get_db 函数来测试 read_items 函数。

案例研究：身份验证和授权

让我们看一个更复杂的例子，如何在 FastAPI/Starlette 中使用依赖注入来实现身份验证和授权。

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from typing import Annotated

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    # 模拟从数据库中获取用户信息
    if token == "fake_token":
        return {"username": "testuser"}
    else:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

async def get_current_active_user(current_user: Annotated[dict, Depends(get_current_user)]):
    # 模拟检查用户是否处于活动状态
    if current_user.get("disabled"):
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

@app.get("/users/me")
async def read_users_me(current_user: Annotated[dict, Depends(get_current_active_user)]):
    return current_user

@app.get("/items/")
async def read_items(current_user: Annotated[dict, Depends(get_current_active_user)]):
    return {"item": "Example Item", "owner": current_user["username"]}

在这个例子中，我们使用了 OAuth2PasswordBearer 来获取访问令牌，并使用 get_current_user 依赖项来验证令牌并获取用户信息。然后，我们使用 get_current_active_user 依赖项来检查用户是否处于活动状态。

通过使用依赖注入，我们可以将身份验证和授权逻辑与请求处理函数分离，使得代码更加清晰、易于理解和维护。

总结：利用依赖注入管理请求级资源

FastAPI/Starlette 的依赖注入系统提供了一种强大的方式来管理 ASGI 生命周期内的请求级资源。通过使用 yield 语句，我们可以确保资源在使用后被及时释放，避免资源泄漏和性能问题。此外，依赖注入还使得单元测试变得更加容易，并提高了代码的可维护性。掌握依赖注入的概念和用法对于构建健壮、可扩展的 Web 应用至关重要。

灵活使用 `Depends`

Depends 不仅仅用于函数，可以传入类，可以结合 Annotated 使用，提供了极高的灵活性。

依赖注入的优势

依赖注入能够帮助更好地组织代码，提高可测试性，并方便地管理请求级资源，这是 FastAPI 强大功能的核心组成部分。

掌握生命周期管理

理解 ASGI 生命周期和依赖注入的交互，有助于开发高效、可靠的 FastAPI 应用程序。

更多IT精英技术系列讲座，到智猿学院